user-engine/tests/test_web.py

import io
import json
import unittest
from urllib.parse import urlencode

from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort
from user_engine.oidc import BrowserSession, OIDCClient
from user_engine.ports import ProvisioningResult
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"


def invoke(
    app, path, *, method="GET", claims=None, marker=SECRET, body=None,
    form=None, cookie=None,
):
    payload = (
        urlencode(form).encode()
        if form is not None
        else 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",
    }
    if form is not None:
        environ["CONTENT_TYPE"] = "application/x-www-form-urlencoded"
    if cookie is not None:
        environ["HTTP_COOKIE"] = cookie
    if claims is not None:
        environ["HTTP_X_VERIFIED_OIDC_CLAIMS"] = json.dumps(claims)
        environ["HTTP_X_USER_ENGINE_PROXY_SECRET"] = marker
    captured = {}

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

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


class PortalApplicationTests(unittest.TestCase):
    def setUp(self):
        store = InMemoryUserEngineStore()
        store.migrate()
        service = UserEngineService(
            store=store,
            identity_adapter=FixtureIdentityClaimsAdapter(),
            authorization=LocalAuthorizationCheckPort(),
        )
        self.app = PortalApplication(
            service,
            trusted_proxy_secret=SECRET,
            login_url="https://kc.example/login",
        )
        self.claims = human_actor_claims(tenant="tenant:friendly:binky")

    def test_public_health_and_home(self):
        health, payload = invoke(self.app, "/healthz")
        self.assertEqual("200 OK", health["status"])
        self.assertEqual("no-store", health["headers"]["Cache-Control"])
        self.assertEqual("ok", json.loads(payload)["status"])
        home, html = invoke(self.app, "/")
        self.assertEqual("200 OK", home["status"])
        self.assertIn(b"Sign in with KeyCape", html)

    def test_metrics_expose_only_bounded_aggregate_state(self):
        denied, _ = invoke(self.app, "/metrics", claims={}, marker="")
        self.assertEqual("403 Forbidden", denied["status"])
        result, payload = invoke(self.app, "/metrics", claims={})
        self.assertEqual("200 OK", result["status"])
        self.assertEqual(
            "text/plain; version=0.0.4; charset=utf-8",
            result["headers"]["Content-Type"],
        )
        self.assertIn(b"user_engine_ready 1", payload)
        self.assertIn(b'user_engine_records{kind="users"} 0', payload)
        self.assertNotIn(SECRET.encode(), payload)

    def test_protected_route_rejects_untrusted_claim_header(self):
        result, payload = invoke(
            self.app, "/api/v1/me", claims=self.claims, marker="attacker"
        )
        self.assertEqual("403 Forbidden", result["status"])
        self.assertNotIn(b"attacker", payload)

    def test_verified_claims_create_current_user(self):
        result, payload = invoke(self.app, "/api/v1/me", claims=self.claims)
        self.assertEqual("200 OK", result["status"])
        decoded = json.loads(payload)
        self.assertEqual("tenant:friendly:binky", decoded["actor"]["tenant"])

    def test_registration_api_is_correlated(self):
        result, payload = invoke(
            self.app,
            "/api/v1/registrations",
            method="POST",
            claims=self.claims,
            body={"tenant": "tenant:friendly:binky"},
        )
        self.assertEqual("201 Created", result["status"])
        self.assertEqual("corr_test", result["headers"]["X-Request-ID"])
        self.assertEqual("factor_pending", json.loads(payload)["status"])

    def test_provision_api_links_provider_subject(self):
        self.app.provisioning = FakeProvisioning()
        created, payload = invoke(
            self.app,
            "/api/v1/tenants/tenant:friendly:binky/users",
            method="POST",
            claims=self.claims,
            body={
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("201 Created", created["status"])
        user_id = json.loads(payload)["user"]["user_id"]
        provisioned, payload = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision",
            method="POST",
            claims=self.claims,
        )
        # The helper does not set an idempotency header.
        self.assertEqual("400 Bad Request", provisioned["status"])
        result, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision",
            self.claims,
        )
        self.assertEqual("200 OK", result["status"])
        self.assertEqual("ada", json.loads(payload)["identity"]["subject"])
        changed, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}",
            self.claims,
            method="PATCH",
            body={"status": "suspended"},
        )
        self.assertEqual("200 OK", changed["status"])
        self.assertEqual("suspended", json.loads(payload)["status"])
        self.assertIn(("suspend", "ada"), self.app.provisioning.actions)

    def test_admin_form_requires_csrf_and_supports_two_step_provisioning(self):
        oidc = OIDCClient(
            issuer="https://kc.example",
            client_id="portal",
            redirect_uri="https://users.example/oidc/callback",
            audience="portal",
        )
        oidc.sessions["browser"] = BrowserSession(
            claims=self.claims,
            expires_at=9999999999,
            csrf_token="csrf-test-token",
        )
        self.app.oidc_client = oidc
        self.app.provisioning = FakeProvisioning()
        denied, _ = invoke(
            self.app,
            "/admin/tenant:friendly:binky/users",
            method="POST",
            cookie="ue_session=browser",
            form={
                "csrf_token": "wrong",
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("403 Forbidden", denied["status"])
        created, _ = invoke(
            self.app,
            "/admin/tenant:friendly:binky/users",
            method="POST",
            cookie="ue_session=browser",
            form={
                "csrf_token": "csrf-test-token",
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("303 See Other", created["status"])
        page, html = invoke(
            self.app,
            "/admin/tenant:friendly:binky",
            cookie="ue_session=browser",
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"ada@example.test", html)
        self.assertIn(b"Create login", html)
        user_id = next(iter(self.app.service.store.users))
        handoff, html = invoke(
            self.app,
            f"/admin/tenant:friendly:binky/users/{user_id}/provision",
            method="POST",
            cookie="ue_session=browser",
            form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("200 OK", handoff["status"])
        self.assertIn(b"Continue to password setup", html)
        self.assertIn(b"https://kc.example/setup/password?token=opaque", html)
        page, html = invoke(
            self.app,
            "/admin/tenant:friendly:binky",
            cookie="ue_session=browser",
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"Create password setup link", html)


class FakeProvisioning:
    def __init__(self):
        self.actions = []

    def provision(self, request):
        self.actions.append(("provision", request.primary_email))
        return ProvisioningResult(
            provider="netkingdom-lldap",
            external_subject=request.primary_email.split("@")[0],
            status="password_setup_required",
            password_setup_url="https://kc.example/setup/password?token=opaque",
        )

    def suspend(self, *, external_subject, idempotency_key, correlation_id):
        self.actions.append(("suspend", external_subject))
        return ProvisioningResult("netkingdom-lldap", external_subject, "suspended")

    def reactivate(self, *, external_subject, idempotency_key, correlation_id):
        self.actions.append(("reactivate", external_subject))
        return ProvisioningResult("netkingdom-lldap", external_subject, "active")


def invoke_with_idempotency(app, path, claims, *, method="POST", body=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",
        "HTTP_X_VERIFIED_OIDC_CLAIMS": json.dumps(claims),
        "HTTP_X_USER_ENGINE_PROXY_SECRET": SECRET,
        "HTTP_IDEMPOTENCY_KEY": "test-idempotency-123456",
    }
    captured = {}
    response = b"".join(app(
        environ,
        lambda status, headers: captured.update(
            {"status": status, "headers": dict(headers)}
        ),
    ))
    return captured, response


if __name__ == "__main__":
    unittest.main()