Implement user-engine portal foundation

tests/test_verified_claims.py

@@ -0,0 +1,31 @@
+import unittest
+
+from user_engine.adapters import VerifiedIdentityClaimsAdapter
+from user_engine.errors import ValidationError
+
+
+class VerifiedIdentityClaimsAdapterTests(unittest.TestCase):
+    def setUp(self):
+        self.adapter = VerifiedIdentityClaimsAdapter(
+            expected_issuer="https://kc.example",
+            expected_audience="user-engine",
+        )
+        self.claims = {
+            "iss": "https://kc.example/",
+            "sub": "person-1",
+            "aud": ["user-engine"],
+            "tenant": "tenant:friendly:binky",
+            "principal_type": "human",
+            "roles": ["tenant-admin"],
+        }
+
+    def test_normalizes_verified_claims(self):
+        actor = self.adapter.normalize(self.claims)
+        self.assertEqual("person-1", actor.subject)
+        self.assertEqual(("tenant-admin",), actor.roles)
+
+    def test_rejects_wrong_issuer_and_audience(self):
+        with self.assertRaises(ValidationError):
+            self.adapter.normalize({**self.claims, "iss": "https://evil.example"})
+        with self.assertRaises(ValidationError):
+            self.adapter.normalize({**self.claims, "aud": ["other"]})

tests/test_web.py

@@ -0,0 +1,89 @@
+import io
+import json
+import unittest
+
+from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort
+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):
+    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",
+    }
+    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"])
+
+
+if __name__ == "__main__":
+    unittest.main()