Implement verified public registration flow

tests/test_registration_verification_adapter.py

@@ -0,0 +1,68 @@
+import io
+import json
+import unittest
+from unittest.mock import patch
+
+from user_engine.adapters.registration_verification import (
+    HTTPRegistrationVerificationAdapter,
+)
+from user_engine.ports import RegistrationVerificationRequest
+
+
+class Response(io.BytesIO):
+    def __enter__(self): return self
+    def __exit__(self, *args): return None
+
+
+class RegistrationVerificationAdapterTests(unittest.TestCase):
+    @patch("user_engine.adapters.registration_verification.urlopen")
+    def test_request_sends_bound_context_and_returns_only_opaque_receipt(self, opener):
+        opener.return_value = Response(
+            json.dumps({"request_id": "vrq_opaque", "accepted": True}).encode()
+        )
+        adapter = HTTPRegistrationVerificationAdapter(
+            base_url="http://verification", bearer_token="secret\n"
+        )
+        receipt = adapter.request(RegistrationVerificationRequest(
+            registration_id="reg_1",
+            normalized_email="person@example.test",
+            preferred_username="person",
+            client_id="coulomb-social",
+            tenant="tenant:coulomb",
+            correlation_id="corr_1",
+        ))
+        self.assertEqual("vrq_opaque", receipt.request_id)
+        request = opener.call_args.args[0]
+        self.assertEqual("Bearer secret", request.headers["Authorization"])
+        self.assertNotIn(b"return_to", request.data)
+
+    @patch("user_engine.adapters.registration_verification.urlopen")
+    def test_consume_requires_registration_purpose_and_normalizes_email(self, opener):
+        opener.return_value = Response(json.dumps({
+            "purpose": "public-registration",
+            "verification_id": "fvr_1",
+            "registration_id": "reg_1",
+            "email": "Person@Example.Test ",
+            "preferred_username": "person",
+            "client_id": "coulomb-social",
+            "tenant": "tenant:coulomb",
+            "source_system": "mail-verifier",
+            "assurance": {"mailbox_control": True},
+        }).encode())
+        adapter = HTTPRegistrationVerificationAdapter(
+            base_url="http://verification", bearer_token="secret"
+        )
+        evidence = adapter.consume("x" * 32)
+        self.assertEqual("person@example.test", evidence.normalized_email)
+        self.assertTrue(evidence.assurance["mailbox_control"])
+
+    @patch("user_engine.adapters.registration_verification.urlopen")
+    def test_consume_rejects_wrong_purpose(self, opener):
+        opener.return_value = Response(json.dumps({
+            "purpose": "password-reset"
+        }).encode())
+        adapter = HTTPRegistrationVerificationAdapter(
+            base_url="http://verification", bearer_token="secret"
+        )
+        with self.assertRaisesRegex(RuntimeError, "wrong purpose"):
+            adapter.consume("x" * 32)

tests/test_web.py

@@ -11,7 +11,13 @@ from user_engine.domain import (
     OnboardingStepStatus, OnboardingTriggerType, SubsystemHandoff,
 )
 from user_engine.oidc import BrowserSession, OIDCClient
-from user_engine.ports import IdentityDriftResult, ProvisioningResult, TenantProvisioningResult
+from user_engine.ports import (
+    IdentityDriftResult,
+    ProvisioningResult,
+    RegistrationVerificationReceipt,
+    TenantProvisioningResult,
+    VerifiedRegistrationApplicant,
+)
 from user_engine.service import UserEngineService
 from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
 from user_engine.web import PortalApplication
@@ -176,6 +182,112 @@ class PortalApplicationTests(unittest.TestCase):
         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_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_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_provision_api_links_provider_subject(self):
         self.app.provisioning = FakeProvisioning()
         created, payload = invoke(
@@ -602,17 +714,43 @@ class PortalApplicationTests(unittest.TestCase):
         self.assertIsNotNone(user.profile_completed_at)
 
 
-class FakeProvisioning:
+class FakeRegistrationVerification:
     def __init__(self):
+        self.requested = None
+        self.registration_id = None
+
+    def request(self, request):
+        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,
+        )
+
+
+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.primary_email.split("@")[0],
+            external_subject=request.preferred_username or request.primary_email.split("@")[0],
             status="password_setup_required",
-            password_setup_url="https://kc.example/setup/password?token=opaque",
+            password_setup_url=self.password_setup_url,
         )
 
     def suspend(self, *, external_subject, idempotency_key, correlation_id):