249 lines
9 KiB
Python
249 lines
9 KiB
Python
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
|
|
|
|
|
|
SECRET = "test-proxy-secret-with-adequate-length"
|
|
|
|
|
|
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,
|
|
"QUERY_STRING": "",
|
|
"CONTENT_LENGTH": str(len(payload)),
|
|
"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
|
|
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"])
|
|
|
|
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)
|
|
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)
|
|
|
|
|
|
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",
|
|
password_setup_url="https://kc.example/setup/password?token=opaque",
|
|
)
|
|
|
|
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()
|