import io import hashlib import json import re import unittest from dataclasses import replace from datetime import timedelta from urllib.parse import quote, urlencode from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort from user_engine.domain import ( OnboardingJourney, OnboardingJourneyStatus, OnboardingStep, OnboardingStepStatus, OnboardingTriggerType, SubsystemHandoff, ) from user_engine.oidc import BrowserSession, OIDCClient from user_engine.ports import ( IdentityDriftResult, ProvisioningResult, RegistrationVerificationReceipt, TenantProvisioningResult, TenantRecord, VerifiedRegistrationApplicant, ) from user_engine.errors import ConflictError, NotFoundError, ValidationError from user_engine.service import UserEngineService from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims from user_engine.web import PortalApplication from user_engine.domain import utc_now SECRET = "test-proxy-secret-with-adequate-length" def invoke( app, path, *, method="GET", claims=None, marker=SECRET, body=None, form=None, cookie=None, headers=None, query="", remote_addr="127.0.0.1", ): payload = ( urlencode(form).encode() if form is not None else json.dumps(body or {}).encode() ) environ = { "REQUEST_METHOD": method, "PATH_INFO": path, "QUERY_STRING": query, "CONTENT_LENGTH": str(len(payload)), "wsgi.input": io.BytesIO(payload), "HTTP_X_REQUEST_ID": "corr_test", "REMOTE_ADDR": remote_addr, } 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 if path == "/api/v1/public/registrations" and method == "POST": environ["HTTP_IDEMPOTENCY_KEY"] = hashlib.sha256(payload).hexdigest() environ.update(headers or {}) 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 platform_claims(self): claims = human_actor_claims(subject="platform-operator", tenant="platform:root") claims["roles"] = ["platform-operator"] return claims 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) self.assertIn(b'name="viewport"', html) self.assertIn(b"focus-visible", html) self.assertIn(b"
", 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_expired_browser_session_and_provider_outage_fail_closed(self): oidc = OIDCClient( issuer="https://kc.example", client_id="portal", redirect_uri="https://users.example/oidc/callback", audience="portal", ) oidc.sessions["expired"] = BrowserSession( claims=self.claims, expires_at=0, csrf_token="expired-csrf" ) self.app.oidc_client = oidc expired, _ = invoke( self.app, "/onboarding", cookie="ue_session=expired" ) self.assertEqual("403 Forbidden", expired["status"]) self.app.oidc_client = None self.app.provisioning = FailingProvisioning() created, payload = invoke( self.app, "/api/v1/tenants/tenant:friendly:binky/users", method="POST", claims=self.claims, body={"primary_email": "failure@example.test", "role": "user"}, ) user_id = json.loads(payload)["user"]["user_id"] before = self.app.service.store.tenant_account( "tenant:friendly:binky", user_id ).status failed, payload = invoke_with_idempotency( self.app, f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision", self.claims, ) self.assertEqual("502 Bad Gateway", failed["status"]) self.assertNotIn(b"provider-secret", payload) self.assertEqual( before, self.app.service.store.tenant_account( "tenant:friendly:binky", user_id ).status, ) 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"]) updated, payload = invoke_with_idempotency( self.app, "/api/v1/me/profile", self.claims, method="PATCH", body={ "display_name": "Sample Person", "consent_accepted": True, "consent_version": "portal-terms-v1", }, ) self.assertEqual("200 OK", updated["status"]) self.assertEqual("Sample Person", json.loads(payload)["display_name"]) 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_public_registration_start_and_verified_factor_are_bound(self): verifier = FakeRegistrationVerification() self.app.registration_verification = verifier self.app.provisioning = FakeProvisioning() self.app.registration_clients = frozenset({"coulomb-social"}) self.app.registration_tenants = frozenset({"tenant:coulomb"}) self.app.registration_oidc_issuer = "https://kc.example" self.app.registration_password_setup_origins = frozenset( {"https://kc.example"} ) started, payload = invoke( self.app, "/api/v1/public/registrations", method="POST", body={ "username": "New.Person", "email": "New.Person@Example.Test", "display_name": "New Person", "client_id": "coulomb-social", "tenant": "tenant:coulomb", "return_to": "https://evil.example", }, ) self.assertEqual("202 Accepted", started["status"]) self.assertEqual({"status": "verification_requested"}, json.loads(payload)) self.assertEqual("new.person", verifier.requested.preferred_username) self.assertEqual("new.person@example.test", verifier.requested.normalized_email) verifier.registration_id = verifier.requested.registration_id verified, payload = invoke( self.app, "/api/v1/public/registrations/verify", method="POST", body={"handle": "x" * 32}, ) self.assertEqual("303 See Other", verified["status"]) self.assertEqual( "https://kc.example/setup/password?token=opaque", verified["headers"]["Location"], ) self.assertEqual(1, self.app.service.operability_snapshot().metrics["users"]) session = self.app.service.store.registration_session(verifier.registration_id) self.assertEqual("completed", session.status.value) identities = self.app.service.store.identities_for_user(session.user_id) self.assertTrue(any( identity.issuer == "https://kc.example" and identity.subject == "new.person" for identity in identities )) provision = self.app.provisioning.requests[-1] self.assertEqual(("user",), provision.roles) self.assertEqual("new.person", provision.preferred_username) self.assertEqual( f"public-registration-{verifier.registration_id}", provision.idempotency_key, ) def test_public_registration_browser_journey_uses_csrf_and_confirmation(self): verifier = FakeRegistrationVerification() self.app.registration_verification = verifier self.app.provisioning = FakeProvisioning() self.app.registration_clients = frozenset({"coulomb-social"}) self.app.registration_tenants = frozenset({"tenant:coulomb"}) self.app.registration_oidc_issuer = "https://kc.example" self.app.registration_password_setup_origins = frozenset({"https://kc.example"}) page, html = invoke(self.app, "/register") self.assertEqual("200 OK", page["status"]) self.assertIn(b"Create your account", html) cookie = page["headers"]["Set-Cookie"] token = re.search(rb'name="csrf_token" value="([^"]+)"', html).group(1).decode() idempotency_key = re.search( rb'name="idempotency_key" value="([^"]+)"', html ).group(1).decode() denied, _ = invoke(self.app, "/register", method="POST", form={ "csrf_token": "wrong", "username": "new.person", "email": "new@example.test", "client_id": "coulomb-social", "tenant": "tenant:coulomb", }, cookie=cookie) self.assertEqual("403 Forbidden", denied["status"]) started, html = invoke(self.app, "/register", method="POST", form={ "csrf_token": token, "idempotency_key": idempotency_key, "username": "new.person", "email": "new@example.test", "display_name": "New Person", "client_id": "coulomb-social", "tenant": "tenant:coulomb", }, cookie=cookie) self.assertEqual("200 OK", started["status"]) self.assertIn(b"Check your email", html) verifier.registration_id = verifier.requested.registration_id confirmation, html = invoke( self.app, "/registration/verify", query="handle=" + "x" * 32 ) self.assertEqual("200 OK", confirmation["status"]) self.assertIn(b"Verify and continue", html) cookie = confirmation["headers"]["Set-Cookie"] token = re.search(rb'name="csrf_token" value="([^"]+)"', html).group(1).decode() completed, html = invoke( self.app, "/registration/verify", method="POST", cookie=cookie, form={"csrf_token": token, "handle": "x" * 32}, ) self.assertEqual("200 OK", completed["status"]) self.assertIn(b"Create password", html) def test_public_registration_browser_cancellation_is_bound_and_terminal(self): verifier = FakeRegistrationVerification() self.app.registration_verification = verifier self.app.registration_clients = frozenset({"coulomb-social"}) self.app.registration_tenants = frozenset({"tenant:coulomb"}) invoke(self.app, "/api/v1/public/registrations", method="POST", body={ "username": "cancel.person", "email": "cancel@example.test", "client_id": "coulomb-social", "tenant": "tenant:coulomb", }) verifier.registration_id = verifier.requested.registration_id confirmation, html = invoke( self.app, "/registration/cancel", query="handle=" + "x" * 32 ) cookie = confirmation["headers"]["Set-Cookie"] token = re.search(rb'name="csrf_token" value="([^"]+)"', html).group(1).decode() canceled, html = invoke( self.app, "/registration/cancel", method="POST", cookie=cookie, form={"csrf_token": token, "handle": "x" * 32}, ) self.assertEqual("200 OK", canceled["status"]) self.assertIn(b"Registration canceled", html) session = self.app.service.store.registration_session(verifier.registration_id) self.assertEqual("abandoned", session.status.value) verify, _ = invoke( self.app, "/api/v1/public/registrations/verify", method="POST", body={"handle": "x" * 32}, remote_addr="127.0.0.2", ) self.assertEqual("400 Bad Request", verify["status"]) def test_public_registration_rejects_untrusted_password_setup_origin(self): verifier = FakeRegistrationVerification() self.app.registration_verification = verifier self.app.provisioning = FakeProvisioning( password_setup_url="https://evil.example/setup" ) self.app.registration_clients = frozenset({"coulomb-social"}) self.app.registration_tenants = frozenset({"tenant:coulomb"}) self.app.registration_oidc_issuer = "https://kc.example" self.app.registration_password_setup_origins = frozenset( {"https://kc.example"} ) invoke( self.app, "/api/v1/public/registrations", method="POST", body={ "username": "person", "email": "person@example.test", "client_id": "coulomb-social", "tenant": "tenant:coulomb", }, ) verifier.registration_id = verifier.requested.registration_id result, _ = invoke( self.app, "/api/v1/public/registrations/verify", method="POST", body={"handle": "x" * 32}, ) self.assertEqual("400 Bad Request", result["status"]) def test_public_registration_resumes_after_provider_outage(self): verifier = FakeRegistrationVerification() provisioning = FailingOnceProvisioning() self.app.registration_verification = verifier self.app.provisioning = provisioning self.app.registration_clients = frozenset({"coulomb-social"}) self.app.registration_tenants = frozenset({"tenant:coulomb"}) self.app.registration_oidc_issuer = "https://kc.example" self.app.registration_password_setup_origins = frozenset({"https://kc.example"}) invoke(self.app, "/api/v1/public/registrations", method="POST", body={ "username": "retry.person", "email": "retry@example.test", "client_id": "coulomb-social", "tenant": "tenant:coulomb", }) verifier.registration_id = verifier.requested.registration_id pending, payload = invoke( self.app, "/api/v1/public/registrations/verify", method="POST", body={"handle": "x" * 32}, ) self.assertEqual("202 Accepted", pending["status"]) recovery = json.loads(payload) resumed, _ = invoke( self.app, "/api/v1/public/registrations/resume", method="POST", body={"registration_id": recovery["registration_id"], "resume_handle": recovery["resume_handle"]}, ) self.assertEqual("303 See Other", resumed["status"]) self.assertEqual(2, len(provisioning.requests)) self.assertEqual( provisioning.requests[0].idempotency_key, provisioning.requests[1].idempotency_key, ) replay, _ = invoke( self.app, "/api/v1/public/registrations/resume", method="POST", body={"registration_id": recovery["registration_id"], "resume_handle": recovery["resume_handle"]}, ) self.assertEqual("403 Forbidden", replay["status"]) def test_public_registration_duplicate_inputs_never_take_over_identity(self): verifier = FakeRegistrationVerification() self.app.registration_verification = verifier self.app.provisioning = FakeProvisioning() self.app.registration_clients = frozenset({"coulomb-social"}) self.app.registration_tenants = frozenset({"tenant:coulomb"}) self.app.registration_oidc_issuer = "https://kc.example" self.app.registration_password_setup_origins = frozenset({"https://kc.example"}) def register(username, email): invoke(self.app, "/api/v1/public/registrations", method="POST", body={ "username": username, "email": email, "client_id": "coulomb-social", "tenant": "tenant:coulomb", }) registration_id = verifier.requested.registration_id verifier.registration_id = registration_id result, _ = invoke( self.app, "/api/v1/public/registrations/verify", method="POST", body={"handle": "x" * 32}, ) return result, self.app.service.store.registration_session(registration_id) first, first_session = register("first.person", "shared@example.test") second, second_session = register("second.person", "shared@example.test") collision, collision_session = register("first.person", "other@example.test") self.assertEqual("303 See Other", first["status"]) self.assertEqual("303 See Other", second["status"]) self.assertNotEqual(first_session.user_id, second_session.user_id) self.assertEqual("409 Conflict", collision["status"]) linked = self.app.service.store.find_identity( "https://kc.example", "first.person" ) self.assertEqual(first_session.user_id, linked.user_id) self.assertNotEqual(collision_session.user_id, linked.user_id) def test_public_registration_rejects_unregistered_client(self): self.app.registration_verification = FakeRegistrationVerification() self.app.registration_clients = frozenset({"coulomb-social"}) self.app.registration_tenants = frozenset({"tenant:coulomb"}) result, _ = invoke( self.app, "/api/v1/public/registrations", method="POST", body={ "username": "person", "email": "person@example.test", "client_id": "unknown", "tenant": "tenant:coulomb", }, ) self.assertEqual("400 Bad Request", result["status"]) def test_public_registration_start_is_idempotent_and_key_is_payload_bound(self): verifier = FakeRegistrationVerification() self.app.registration_verification = verifier self.app.registration_clients = frozenset({"coulomb-social"}) self.app.registration_tenants = frozenset({"tenant:coulomb"}) body = { "username": "idem.person", "email": "idem@example.test", "client_id": "coulomb-social", "tenant": "tenant:coulomb", } headers = {"HTTP_IDEMPOTENCY_KEY": "registration-key-123456789"} first, _ = invoke( self.app, "/api/v1/public/registrations", method="POST", body=body, headers=headers, ) registration_id = verifier.requested.registration_id replay, _ = invoke( self.app, "/api/v1/public/registrations", method="POST", body=body, headers=headers, ) conflicting, _ = invoke( self.app, "/api/v1/public/registrations", method="POST", body={**body, "email": "different@example.test"}, headers=headers, ) sessions = self.app.service.store.all_registration_sessions() self.assertEqual("202 Accepted", first["status"]) self.assertEqual("202 Accepted", replay["status"]) self.assertEqual("409 Conflict", conflicting["status"]) self.assertEqual(1, len(sessions)) self.assertEqual(1, verifier.request_count) self.assertEqual(registration_id, sessions[0].registration_id) self.assertNotIn("registration-key-123456789", repr(sessions[0])) def test_public_registration_rate_limit_uses_peer_not_forwarded_header(self): self.app.registration_verification = FakeRegistrationVerification() self.app.registration_rate_limit = 2 self.app.registration_rate_window_seconds = 60 body = { "username": "person", "email": "person@example.test", "client_id": "unknown", "tenant": "tenant:coulomb", } first, _ = invoke( self.app, "/api/v1/public/registrations", method="POST", body=body, headers={"HTTP_X_FORWARDED_FOR": "198.51.100.1"}, ) second, _ = invoke( self.app, "/api/v1/public/registrations", method="POST", body=body, headers={"HTTP_X_FORWARDED_FOR": "198.51.100.2"}, ) limited, payload = invoke( self.app, "/api/v1/public/registrations", method="POST", body=body, headers={"HTTP_X_FORWARDED_FOR": "198.51.100.3"}, ) other_peer, _ = invoke( self.app, "/api/v1/public/registrations", method="POST", body=body, remote_addr="127.0.0.2", ) self.assertEqual("400 Bad Request", first["status"]) self.assertEqual("400 Bad Request", second["status"]) self.assertEqual("429 Too Many Requests", limited["status"]) self.assertEqual("rate_limited", json.loads(payload)["error"]["code"]) self.assertEqual("400 Bad Request", other_peer["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) removed, payload = invoke_with_idempotency( self.app, f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}", self.claims, method="DELETE", ) self.assertEqual("200 OK", removed["status"]) self.assertEqual("removed", json.loads(payload)["status"]) self.assertIn(("deprovision", "ada"), self.app.provisioning.actions) def test_invitation_lifecycle_is_versioned_and_replay_safe(self): created, payload = invoke( self.app, "/api/v1/tenants/tenant:friendly:binky/invitations", method="POST", claims=self.claims, body={"primary_email": "invitee@example.test", "role": "user"}, ) self.assertEqual("201 Created", created["status"]) invitation = json.loads(payload)["invitation"] invitation_id = invitation["invitation_id"] self.assertEqual(1, invitation["version"]) self.assertIsNotNone(invitation["expires_at"]) duplicate, _ = invoke( self.app, "/api/v1/tenants/tenant:friendly:binky/invitations", method="POST", claims=self.claims, body={"primary_email": "INVITEE@example.test", "role": "user"}, ) self.assertEqual("409 Conflict", duplicate["status"]) listed, payload = invoke( self.app, "/api/v1/tenants/tenant:friendly:binky/invitations", claims=self.claims, ) self.assertEqual("200 OK", listed["status"]) self.assertEqual(1, len(json.loads(payload)["items"])) resent, payload = invoke( self.app, f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/resend", method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"1"'}, ) self.assertEqual("200 OK", resent["status"]) self.assertEqual(2, json.loads(payload)["version"]) conflict, _ = invoke( self.app, f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/expire", method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"1"'}, ) self.assertEqual("409 Conflict", conflict["status"]) expired, payload = invoke( self.app, f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/expire", method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"2"'}, ) self.assertEqual("200 OK", expired["status"]) self.assertEqual("revoked", json.loads(payload)["status"]) claimable, payload = invoke( self.app, "/api/v1/tenants/tenant:friendly:binky/invitations", method="POST", claims=self.claims, body={"primary_email": "sample.user@example.test", "role": "user"}, ) self.assertEqual("201 Created", claimable["status"]) claim_id = json.loads(payload)["invitation"]["invitation_id"] claimed, _ = invoke( self.app, f"/api/v1/invitations/{claim_id}/claim", method="POST", claims=self.claims, ) self.assertEqual("200 OK", claimed["status"]) replayed, _ = invoke( self.app, f"/api/v1/invitations/{claim_id}/claim", method="POST", claims=self.claims, ) self.assertEqual("400 Bad Request", replayed["status"]) timed, payload = invoke( self.app, "/api/v1/tenants/tenant:friendly:binky/invitations", method="POST", claims=self.claims, body={"primary_email": "expired@example.test", "role": "user"}, ) timed_id = json.loads(payload)["invitation"]["invitation_id"] timed_invitation = self.app.service.store.family_invitation(timed_id) self.app.service.store.save_family_invitation(replace( timed_invitation, expires_at=utc_now() - timedelta(seconds=1) )) expired_claims = dict(self.claims) expired_claims["email"] = "expired@example.test" timed_out, _ = invoke( self.app, f"/api/v1/invitations/{timed_id}/claim", method="POST", claims=expired_claims, ) self.assertEqual("400 Bad Request", timed_out["status"]) def test_outbox_delivery_replay_and_dead_letter(self): invoke(self.app, "/api/v1/me", claims=self.claims) actor = self.app.service.identity_adapter.normalize(self.claims) event = self.app.service.outbox_events()[0] def unavailable(_event): raise RuntimeError("notification provider unavailable; token=redacted") first = self.app.service.deliver_outbox( actor, unavailable, worker_id="worker-a", max_attempts=2 )[0] self.assertEqual(1, first.delivery_attempts) self.assertIsNotNone(first.failed_at) self.assertIsNone(first.dead_lettered_at) replayed = self.app.service.replay_outbox(actor, event.event_id) self.assertIsNone(replayed.failure_reason) second = self.app.service.deliver_outbox( actor, unavailable, worker_id="worker-a", max_attempts=2 )[0] self.assertIsNotNone(second.dead_lettered_at) self.assertLessEqual(len(second.failure_reason), 200) replayed = self.app.service.replay_outbox(actor, event.event_id) delivered = self.app.service.deliver_outbox( actor, lambda _event: None, worker_id="worker-b" )[0] self.assertEqual(replayed.event_id, delivered.event_id) self.assertIsNotNone(delivered.delivered_at) self.assertNotIn(delivered, self.app.service.outbox_events()) def test_platform_tenant_bootstrap_recovery_and_outbox_transport(self): self.app.tenant_management = FakeTenantManagement() self.app.provisioning = FakeProvisioning() claims = self.platform_claims() denied, _ = invoke_with_idempotency( self.app, "/api/v1/platform/tenants", self.claims, body={"tenant": "tenant:friendly:new", "display_name": "New"}, ) self.assertEqual("403 Forbidden", denied["status"]) created, payload = invoke_with_idempotency( self.app, "/api/v1/platform/tenants", claims, body={ "tenant": "tenant:friendly:new", "display_name": "New Tenant", "first_admin": {"primary_email": "admin@new.test", "display_name": "Admin"}, }, ) self.assertEqual("201 Created", created["status"]) decoded = json.loads(payload) self.assertEqual("created", decoded["tenant"]["status"]) self.assertEqual("tenant-admin", decoded["first_admin"]["membership"]["kind"]) user_id = decoded["first_admin"]["user"]["user_id"] recovered, payload = invoke_with_idempotency( self.app, f"/api/v1/platform/tenants/tenant:friendly:new/users/{user_id}/recover", claims, ) self.assertEqual("200 OK", recovered["status"]) self.assertEqual("active", json.loads(payload)["tenant_account"]["status"]) self.app.outbox_delivery = lambda _event: None delivered, payload = invoke( self.app, "/api/v1/platform/outbox/deliver", method="POST", claims=claims, body={"worker_id": "test-worker"}, ) self.assertEqual("200 OK", delivered["status"]) self.assertTrue(json.loads(payload)["items"]) def test_platform_tenant_lifecycle_is_delegated_to_the_authority(self): authority = FakeTenantManagement() self.app.tenant_management = authority claims = self.platform_claims() tenant = "tenant:friendly:lifecycle" invoke_with_idempotency( self.app, "/api/v1/platform/tenants", claims, body={"tenant": tenant, "display_name": "Lifecycle"}, ) path = f"/api/v1/platform/tenants/{tenant}" denied, _ = invoke(self.app, path, claims=self.claims) self.assertEqual("403 Forbidden", denied["status"]) read, payload = invoke(self.app, path, claims=claims) self.assertEqual("200 OK", read["status"]) record = json.loads(payload) self.assertEqual("active", record["lifecycle"]) self.assertEqual(1, record["version"]) def mutate(suffix, *, method, body, version, key): return invoke( self.app, path + suffix, method=method, claims=claims, body=body, headers={ "HTTP_IF_MATCH": f'"{version}"', "HTTP_IDEMPOTENCY_KEY": key, }, ) unconditional, _ = invoke( self.app, path, method="PATCH", claims=claims, body={"metadata": {"display_name": "X"}, "reason": "rename"}, headers={"HTTP_IDEMPOTENCY_KEY": "tenant-update-0000000000"}, ) self.assertEqual("400 Bad Request", unconditional["status"]) unreasoned, _ = mutate( "", method="PATCH", body={"metadata": {"display_name": "X"}}, version=1, key="tenant-update-0000000001", ) self.assertEqual("400 Bad Request", unreasoned["status"]) updated, payload = mutate( "", method="PATCH", body={"metadata": {"display_name": "Renamed"}, "reason": "operator rename"}, version=1, key="tenant-update-0000000002", ) self.assertEqual("200 OK", updated["status"]) self.assertEqual("Renamed", json.loads(payload)["display_name"]) self.assertEqual(2, json.loads(payload)["version"]) replayed, payload = mutate( "", method="PATCH", body={"metadata": {"display_name": "Renamed"}, "reason": "operator rename"}, version=1, key="tenant-update-0000000002", ) self.assertEqual("200 OK", replayed["status"]) self.assertTrue(json.loads(payload)["replayed"]) self.assertEqual(2, json.loads(payload)["version"]) stale, _ = mutate( "", method="PATCH", body={"metadata": {"display_name": "Again"}, "reason": "second rename"}, version=1, key="tenant-update-0000000003", ) self.assertEqual("409 Conflict", stale["status"]) retired, payload = mutate( "/retire", method="POST", body={"reason": "contract ended"}, version=2, key="tenant-retire-0000000001", ) self.assertEqual("200 OK", retired["status"]) self.assertEqual("retired", json.loads(payload)["lifecycle"]) while_retired, _ = mutate( "", method="PATCH", body={"metadata": {"display_name": "Nope"}, "reason": "late rename"}, version=3, key="tenant-update-0000000004", ) self.assertEqual("409 Conflict", while_retired["status"]) double, _ = mutate( "/retire", method="POST", body={"reason": "again"}, version=3, key="tenant-retire-0000000002", ) self.assertEqual("409 Conflict", double["status"]) reactivated, payload = mutate( "/reactivate", method="POST", body={"reason": "contract renewed"}, version=3, key="tenant-reactivate-000001", ) self.assertEqual("200 OK", reactivated["status"]) self.assertEqual("active", json.loads(payload)["lifecycle"]) # user-engine keeps no tenant table of its own: every read and write # above went to the authority. self.assertEqual({tenant}, set(authority.records)) missing, _ = invoke( self.app, "/api/v1/platform/tenants/tenant:friendly:absent/retire", method="POST", claims=claims, body={"reason": "unknown"}, headers={ "HTTP_IF_MATCH": '"1"', "HTTP_IDEMPOTENCY_KEY": "tenant-retire-0000000009", }, ) self.assertEqual("404 Not Found", missing["status"]) def test_platform_browser_tenant_lifecycle_controls(self): authority = FakeTenantManagement() claims = self.platform_claims() oidc = OIDCClient( issuer="https://kc.example", client_id="portal", redirect_uri="https://users.example/oidc/callback", audience="portal", ) oidc.sessions["platform"] = BrowserSession( claims=claims, expires_at=9999999999, csrf_token="platform-csrf", ) self.app.oidc_client = oidc self.app.tenant_management = authority tenant = "tenant:friendly:browserlifecycle" authority.create_tenant( tenant=tenant, display_name="Browser Lifecycle", idempotency_key="seed", correlation_id="corr", ) quoted = quote(tenant, safe="") lookup, _ = invoke( self.app, "/platform/tenant", cookie="ue_session=platform", query=urlencode({"tenant": tenant}), ) self.assertEqual("303 See Other", lookup["status"]) self.assertEqual(f"/platform/tenants/{quoted}", dict(lookup["headers"])["Location"]) page, html = invoke( self.app, f"/platform/tenants/{quoted}", cookie="ue_session=platform" ) self.assertEqual("200 OK", page["status"]) self.assertIn(b"Retire tenant", html) self.assertIn(b'name="version" value="1"', html) forged, _ = invoke( self.app, f"/platform/tenants/{quoted}", method="POST", cookie="ue_session=platform", form={ "csrf_token": "wrong", "operation": "retire", "version": "1", "reason": "forged", }, ) self.assertEqual("403 Forbidden", forged["status"]) self.assertEqual("active", authority.records[tenant].lifecycle) renamed, html = invoke( self.app, f"/platform/tenants/{quoted}", method="POST", cookie="ue_session=platform", form={ "csrf_token": "platform-csrf", "operation": "update", "version": "1", "display_name": "Renamed In Browser", "reason": "operator rename", }, ) self.assertEqual("200 OK", renamed["status"]) self.assertIn(b"Renamed In Browser", html) retired, html = invoke( self.app, f"/platform/tenants/{quoted}", method="POST", cookie="ue_session=platform", form={ "csrf_token": "platform-csrf", "operation": "retire", "version": "2", "reason": "contract ended", }, ) self.assertEqual("200 OK", retired["status"]) self.assertIn(b"Reactivate tenant", html) # A retired tenant offers no metadata form, matching the authority. self.assertNotIn(b"Save metadata", html) resubmitted, html = invoke( self.app, f"/platform/tenants/{quoted}", method="POST", cookie="ue_session=platform", form={ "csrf_token": "platform-csrf", "operation": "retire", "version": "2", "reason": "contract ended", }, ) self.assertEqual("200 OK", resubmitted["status"]) self.assertIn(b"replayed", html) self.assertEqual(3, authority.records[tenant].version) platform, html = invoke( self.app, "/platform", cookie="ue_session=platform" ) self.assertIn(b"Manage an existing tenant", html) def test_platform_browser_tenant_and_first_admin_bootstrap(self): claims = self.platform_claims() oidc = OIDCClient( issuer="https://kc.example", client_id="portal", redirect_uri="https://users.example/oidc/callback", audience="portal", ) oidc.sessions["platform"] = BrowserSession( claims=claims, expires_at=9999999999, csrf_token="platform-csrf", ) self.app.oidc_client = oidc self.app.tenant_management = FakeTenantManagement() self.app.provisioning = FakeProvisioning() page, html = invoke(self.app, "/platform", cookie="ue_session=platform") self.assertEqual("200 OK", page["status"]) self.assertIn(b"First administrator", html) denied, _ = invoke( self.app, "/platform/tenants", method="POST", cookie="ue_session=platform", form={ "csrf_token": "wrong", "tenant": "tenant:friendly:browser", "display_name": "Browser Tenant", }, ) self.assertEqual("403 Forbidden", denied["status"]) created, html = invoke( self.app, "/platform/tenants", method="POST", cookie="ue_session=platform", form={ "csrf_token": "platform-csrf", "tenant": "tenant:friendly:browser", "display_name": "Browser Tenant", "admin_display_name": "First Admin", "admin_email": "first-admin@browser.test", }, ) self.assertEqual("200 OK", created["status"]) self.assertIn(b"awaiting onboarding", html) memberships = self.app.service.store.memberships_for_tenant( "tenant:friendly:browser" ) self.assertEqual("tenant-admin", memberships[0].kind) admin_page, html = invoke( self.app, "/admin/tenant:friendly:browser", cookie="ue_session=platform" ) self.assertEqual("200 OK", admin_page["status"]) self.assertIn(b"Lifecycle diagnostics", html) self.assertIn(b"Recover identity", html) recovered, _ = invoke( self.app, f"/admin/tenant:friendly:browser/users/{memberships[0].user_id}/recover", method="POST", cookie="ue_session=platform", form={"csrf_token": "platform-csrf"}, ) self.assertEqual("303 See Other", recovered["status"]) 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) self.assertIn(b"Invite a user", html) self.assertIn(b"Remove account", html) def test_browser_invitation_acceptance_and_onboarding_status(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 created, _ = invoke( self.app, "/admin/tenant:friendly:binky/invitations", method="POST", cookie="ue_session=browser", form={ "csrf_token": "csrf-test-token", "display_name": "Invitee", "primary_email": "invitee@example.test", "role": "user", }, ) self.assertEqual("303 See Other", created["status"]) invitation = next(iter(self.app.service.store.family_invitations.values())) page, html = invoke( self.app, f"/invitations/{invitation.invitation_id}", cookie="ue_session=browser", ) self.assertEqual("200 OK", page["status"]) self.assertIn(b"Accept invitation", html) wrong, _ = invoke( self.app, f"/invitations/{invitation.invitation_id}", method="POST", cookie="ue_session=browser", form={"csrf_token": "csrf-test-token"}, ) self.assertEqual("403 Forbidden", wrong["status"]) invitee_claims = human_actor_claims( subject="invitee", tenant="tenant:friendly:binky" ) invitee_claims["email"] = "invitee@example.test" oidc.sessions["browser"] = BrowserSession( claims=invitee_claims, expires_at=9999999999, csrf_token="csrf-test-token", ) accepted, _ = invoke( self.app, f"/invitations/{invitation.invitation_id}", method="POST", cookie="ue_session=browser", form={"csrf_token": "csrf-test-token"}, ) self.assertEqual("303 See Other", accepted["status"]) self.assertEqual("/onboarding", accepted["headers"]["Location"]) own_journey = OnboardingJourney( tenant="tenant:friendly:binky", user_id=invitation.user_id, protocol_id="protocol-self", trigger_type=OnboardingTriggerType.INVITATION, status=OnboardingJourneyStatus.IN_PROGRESS, active_step_key="profile-review", steps=(OnboardingStep( step_key="profile-review", title="Review your profile", subsystem="user-engine", status=OnboardingStepStatus.IN_PROGRESS, ),), ) provider_journey = OnboardingJourney( tenant="tenant:friendly:binky", user_id=invitation.user_id, protocol_id="protocol-provider", trigger_type=OnboardingTriggerType.INVITATION, status=OnboardingJourneyStatus.BLOCKED, active_step_key="mfa", steps=(OnboardingStep( step_key="mfa", title="Enroll MFA", subsystem="key-cape", status=OnboardingStepStatus.BLOCKED, handoff=SubsystemHandoff( subsystem="key-cape", status=OnboardingStepStatus.BLOCKED, ), ),), ) self.app.service.store.save_onboarding_journey(own_journey) self.app.service.store.save_onboarding_journey(provider_journey) onboarding, html = invoke( self.app, "/onboarding", cookie="ue_session=browser" ) self.assertEqual("200 OK", onboarding["status"]) self.assertIn(b"Onboarding progress", html) self.assertIn(b"tenant:friendly:binky", html) self.assertIn(b"Profile and consent", html) self.assertIn(b"Mark complete", html) self.assertIn(b"provider-owned surface", html) completed, _ = invoke( self.app, f"/onboarding/{own_journey.journey_id}/steps/profile-review/complete", method="POST", cookie="ue_session=browser", form={"csrf_token": "csrf-test-token"}, ) self.assertEqual("303 See Other", completed["status"]) provider_denied, _ = invoke( self.app, f"/onboarding/{provider_journey.journey_id}/steps/mfa/complete", method="POST", cookie="ue_session=browser", form={"csrf_token": "csrf-test-token"}, ) self.assertEqual("403 Forbidden", provider_denied["status"]) saved, _ = invoke( self.app, "/onboarding/profile", method="POST", cookie="ue_session=browser", form={ "csrf_token": "csrf-test-token", "display_name": "Updated Invitee", "consent_accepted": "yes", }, ) self.assertEqual("303 See Other", saved["status"]) user = self.app.service.store.user(invitation.user_id) self.assertEqual("Updated Invitee", user.display_name) self.assertEqual("invitee@example.test", user.primary_email) self.assertEqual("portal-terms-v1", user.consent_version) self.assertIsNotNone(user.profile_completed_at) class FakeRegistrationVerification: def __init__(self): self.requested = None self.registration_id = None self.request_count = 0 def request(self, request): self.request_count += 1 self.requested = request return RegistrationVerificationReceipt(request_id="vrq_test") def consume(self, opaque_handle): return VerifiedRegistrationApplicant( verification_id="fvr_test", registration_id=self.registration_id, normalized_email=self.requested.normalized_email, preferred_username=self.requested.preferred_username, client_id=self.requested.client_id, tenant=self.requested.tenant, source_system="mail-verifier", assurance={"mailbox_control": True}, display_name=self.requested.display_name, ) def cancel(self, opaque_handle): return self.consume(opaque_handle) class FakeProvisioning: def __init__(self, password_setup_url="https://kc.example/setup/password?token=opaque"): self.actions = [] self.requests = [] self.password_setup_url = password_setup_url def provision(self, request): self.requests.append(request) self.actions.append(("provision", request.primary_email)) return ProvisioningResult( provider="netkingdom-lldap", external_subject=request.preferred_username or request.primary_email.split("@")[0], status="password_setup_required", password_setup_url=self.password_setup_url, ) 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 deprovision(self, *, external_subject, idempotency_key, correlation_id): self.actions.append(("deprovision", external_subject)) return ProvisioningResult("netkingdom-lldap", external_subject, "removed") def reconcile(self, request, *, external_subject, desired_status="active"): self.actions.append(("reconcile", external_subject)) return IdentityDriftResult( "netkingdom-lldap", external_subject, "in_sync", changed=("status",) ) class FailingOnceProvisioning(FakeProvisioning): def provision(self, request): self.requests.append(request) if len(self.requests) == 1: raise RuntimeError("provider unavailable") return ProvisioningResult( provider="netkingdom-lldap", external_subject=request.preferred_username, status="password_setup_required", password_setup_url=self.password_setup_url, ) class FakeTenantManagement: """Stands in for tenant-engine, including its compare-and-swap semantics.""" def __init__(self): self.records = {} self.receipts = {} self.reasons = [] def create_tenant(self, *, tenant, display_name, idempotency_key, correlation_id): self.records.setdefault(tenant, TenantRecord( tenant=tenant, external_ref=tenant, lifecycle="active", version=1, display_name=display_name, )) return TenantProvisioningResult(tenant=tenant, status="created") def tenant(self, *, tenant, correlation_id): record = self.records.get(tenant) if record is None: raise NotFoundError("tenant not found") return record def update_tenant(self, *, tenant, metadata, expected_version, reason, idempotency_key, correlation_id): record = self._mutate(tenant, expected_version, reason, idempotency_key) if record is not None: return record current = self.records[tenant] if current.lifecycle == "retired": raise ConflictError("invalid_lifecycle_transition") return self._commit(idempotency_key, replace( current, version=current.version + 1, display_name=metadata.get("display_name", current.display_name), contact_email=metadata.get("contact_email", current.contact_email), )) def retire_tenant(self, *, tenant, expected_version, reason, idempotency_key, correlation_id): return self._transition( tenant, "retired", expected_version, reason, idempotency_key ) def reactivate_tenant(self, *, tenant, expected_version, reason, idempotency_key, correlation_id): return self._transition( tenant, "active", expected_version, reason, idempotency_key ) def _transition(self, tenant, lifecycle, expected_version, reason, idempotency_key): record = self._mutate(tenant, expected_version, reason, idempotency_key) if record is not None: return record current = self.records[tenant] if current.lifecycle == lifecycle: raise ConflictError("invalid_lifecycle_transition") return self._commit(idempotency_key, replace( current, lifecycle=lifecycle, version=current.version + 1 )) def _mutate(self, tenant, expected_version, reason, idempotency_key): if idempotency_key in self.receipts: return replace(self.receipts[idempotency_key], replayed=True) if tenant not in self.records: raise NotFoundError("tenant not found") if not reason: raise ValidationError("invalid_update") if expected_version != self.records[tenant].version: raise ConflictError("version_conflict") self.reasons.append(reason) return None def _commit(self, idempotency_key, record): self.records[record.tenant] = record self.receipts[idempotency_key] = record return record class FailingProvisioning(FakeProvisioning): def provision(self, request): raise RuntimeError("provider-secret must never escape") 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()