user-engine/tests/test_web_account_authority.py

import io
import json
import unittest

from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort
from user_engine.domain import Actor, AuthorizationEffect, PrincipalType
from user_engine.service import UserEngineService
from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
from user_engine.web import PortalApplication

SECRET = "test-proxy-secret-with-adequate-length"
HUB_TOKEN = "hub-workload-token-with-adequate-length"
HUB_CALLER = Actor(
    issuer="https://hub.example", subject="hub-workload",
    tenant="platform:root", principal_type=PrincipalType.SERVICE, audience=(),
)


def invoke(app, path, *, method="POST", body=None, authorization=None):
    payload = json.dumps(body or {}).encode()
    environ = {
        "REQUEST_METHOD": method,
        "PATH_INFO": path,
        "QUERY_STRING": "",
        "CONTENT_LENGTH": str(len(payload)),
        "wsgi.input": io.BytesIO(payload),
        "HTTP_X_REQUEST_ID": "corr_test",
        "REMOTE_ADDR": "127.0.0.1",
    }
    if authorization is not None:
        environ["HTTP_AUTHORIZATION"] = authorization
    captured = {}

    def start_response(status, headers):
        captured["status"] = status
        captured["headers"] = dict(headers)

    response = b"".join(app(environ, start_response))
    return captured, response


class AccountAuthorityRouteTests(unittest.TestCase):
    def setUp(self):
        self.store = InMemoryUserEngineStore()
        self.store.migrate()
        self.policy = LocalAuthorizationCheckPort()
        self.service = UserEngineService(
            store=self.store,
            identity_adapter=FixtureIdentityClaimsAdapter(),
            authorization=self.policy,
        )
        self.session = self.service.me(human_actor_claims())
        self.app = PortalApplication(
            self.service,
            trusted_proxy_secret=SECRET,
            login_url="https://kc.example/login",
            hub_authority_caller=HUB_CALLER,
            hub_authority_token=HUB_TOKEN,
        )
        self.body = {
            "issuer": self.session.actor.issuer,
            "subject": self.session.actor.subject,
            "tenant": self.session.actor.tenant,
        }

    def test_route_disabled_without_configuration(self):
        app = PortalApplication(
            self.service, trusted_proxy_secret=SECRET, login_url="https://kc.example/login",
        )
        result, _ = invoke(
            app, "/api/v1/platform/account-authority/lookup",
            body=self.body, authorization=f"Bearer {HUB_TOKEN}",
        )
        self.assertEqual("404 Not Found", result["status"])

    def test_missing_or_wrong_bearer_is_denied(self):
        missing, _ = invoke(self.app, "/api/v1/platform/account-authority/lookup", body=self.body)
        self.assertEqual("403 Forbidden", missing["status"])
        wrong, _ = invoke(
            self.app, "/api/v1/platform/account-authority/lookup",
            body=self.body, authorization="Bearer not-the-token",
        )
        self.assertEqual("403 Forbidden", wrong["status"])

    def test_denial_precedes_any_domain_lookup(self):
        self.policy.default_effect = AuthorizationEffect.DENY
        before_users = dict(self.store.users)
        result, _ = invoke(
            self.app, "/api/v1/platform/account-authority/lookup",
            body=self.body, authorization=f"Bearer {HUB_TOKEN}",
        )
        self.assertEqual("403 Forbidden", result["status"])
        self.assertEqual(self.store.users, before_users)

    def test_valid_workload_bearer_returns_current_state_without_provisioning(self):
        result, payload = invoke(
            self.app, "/api/v1/platform/account-authority/lookup",
            body=self.body, authorization=f"Bearer {HUB_TOKEN}",
        )
        self.assertEqual("200 OK", result["status"])
        data = json.loads(payload)
        self.assertEqual(data["identity_id"], self.session.identities[0].identity_id)
        self.assertEqual(data["account_status"], "active")
        self.assertNotIn("roles", data)
        self.assertNotIn("primary_email", data)
        request = self.policy.requests[-1]
        self.assertEqual(request.actor.subject, "hub-workload")
        self.assertNotEqual(request.actor.subject, self.session.actor.subject)

    def test_unknown_identity_is_not_found_and_does_not_provision(self):
        before = repr(self.store.__dict__)
        body = dict(self.body, subject="unknown-subject")
        result, _ = invoke(
            self.app, "/api/v1/platform/account-authority/lookup",
            body=body, authorization=f"Bearer {HUB_TOKEN}",
        )
        self.assertEqual("404 Not Found", result["status"])
        self.assertEqual(repr(self.store.__dict__), before)

    def test_end_user_claims_headers_do_not_satisfy_workload_auth(self):
        environ_extra = {
            "HTTP_X_VERIFIED_OIDC_CLAIMS": json.dumps(human_actor_claims()),
            "HTTP_X_USER_ENGINE_PROXY_SECRET": SECRET,
        }

        def call():
            payload = json.dumps(self.body).encode()
            environ = {
                "REQUEST_METHOD": "POST",
                "PATH_INFO": "/api/v1/platform/account-authority/lookup",
                "QUERY_STRING": "",
                "CONTENT_LENGTH": str(len(payload)),
                "wsgi.input": io.BytesIO(payload),
                "HTTP_X_REQUEST_ID": "corr_test",
                "REMOTE_ADDR": "127.0.0.1",
                **environ_extra,
            }
            captured = {}

            def start_response(status, headers):
                captured["status"] = status

            response = b"".join(self.app(environ, start_response))
            return captured, response

        result, _ = call()
        self.assertEqual("403 Forbidden", result["status"])

    def test_construction_requires_both_caller_and_token(self):
        with self.assertRaises(ValueError):
            PortalApplication(
                self.service, trusted_proxy_secret=SECRET,
                login_url="https://kc.example/login", hub_authority_caller=HUB_CALLER,
            )
        with self.assertRaises(ValueError):
            PortalApplication(
                self.service, trusted_proxy_secret=SECRET,
                login_url="https://kc.example/login", hub_authority_token=HUB_TOKEN,
            )