935 lines
40 KiB
Python
935 lines
40 KiB
Python
import io
|
|
import json
|
|
import re
|
|
import unittest
|
|
from dataclasses import replace
|
|
from datetime import timedelta
|
|
from urllib.parse import 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,
|
|
VerifiedRegistrationApplicant,
|
|
)
|
|
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
|
|
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"<main>", 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()
|
|
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, "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_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_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_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_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
|
|
|
|
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.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:
|
|
def create_tenant(self, *, tenant, display_name, idempotency_key, correlation_id):
|
|
return TenantProvisioningResult(tenant=tenant, status="created")
|
|
|
|
|
|
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()
|