Add CSRF-safe tenant identity administration

tests/test_web.py

@@ -1,8 +1,11 @@
 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
@@ -11,8 +14,15 @@ 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()
+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,
@@ -21,6 +31,10 @@ def invoke(app, path, *, method="GET", claims=None, marker=SECRET, body=None):
         "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
@@ -84,6 +98,140 @@ class PortalApplicationTests(unittest.TestCase):
         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()