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_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) 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", ) 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()