coulomb/user-engine
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add registration security conformance

Browse Source
This commit is contained in:
Bernd Worsch
2026-06-15 23:59:45 +02:00
parent aaefa48212
commit 2ceecf6463
10 changed files with 846 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
tests/test_integrated_scenarios.py
View File

@@ -18,6 +18,7 @@ from user_engine.projections import ClaimsEnrichmentProjectionCache
from user_engine.service import REDACTED, UserEngineService
from user_engine.testing.fixtures import sample_application, sample_application_binding
from user_engine.testing.scenarios import (
REGISTRATION_SCENARIO_MATRIX,
SCENARIO_MATRIX,
ScenarioAuthorizationHarness,
StrictFixtureIdentityClaimsAdapter,
@@ -44,7 +45,29 @@ class IntegratedScenarioTests(unittest.TestCase):
"two_applications",
"sensitive_redaction",
"audit_event_replay",
"identity_canon_context",
"family_dataspace_onboarding",
"registration_onboarding_full",
"prepared_account_claim",
"privileged_role_requires_approval",
"eid_assurance_registration",
"tenant_admin_invite",
"group_access_hat",
"denied_cross_tenant_claim",
"ui_registration_access_flow",
},
)
self.assertEqual(
{scenario["id"] for scenario in REGISTRATION_SCENARIO_MATRIX},
{
"self_registration",
"prepared_account_claim",
"privileged_role_requires_approval",
"eid_assurance_registration",
"family_invite",
"tenant_admin_invite",
"group_access",
"denied_cross_tenant_claim",
},
)

576
tests/test_registration_security_conformance.py Normal file
View File

@@ -0,0 +1,576 @@
from dataclasses import replace
from datetime import timedelta
import unittest
from user_engine.adapters.local import InMemoryUserEngineStore, LocalAuthorizationCheckPort
from user_engine.domain import (
AccessMembershipRequirement,
AccessProfile,
AccessScopeType,
AccountStatus,
AttributeDefinition,
Catalog,
CatalogLifecycle,
FactorVerification,
IdentityFactorType,
Mutability,
OnboardingJourneyStatus,
OnboardingTriggerType,
PreparedEntitlement,
PreparedEntitlementKind,
PreparedFactorRequirement,
ProfileScope,
ProjectionType,
Sensitivity,
Visibility,
WelcomeProtocol,
WelcomeProtocolStep,
utc_now,
)
from user_engine.errors import AuthorizationDenied, ConflictError, ValidationError
from user_engine.service import REDACTED, UserEngineService
from user_engine.testing.fixtures import (
FixtureIdentityClaimsAdapter,
human_actor_claims,
sample_application,
sample_application_binding,
)
from user_engine.testing.scenarios import (
ScenarioAuthorizationHarness,
StrictFixtureIdentityClaimsAdapter,
missing_tenant_claims,
)
from user_engine.ui import RegistrationAccessManagementUi, UiViewport
class RegistrationSecurityConformanceTests(unittest.TestCase):
def test_full_registration_claim_hat_onboarding_ui_conformance_path(self):
service, store, _ = _service()
actor = _actor("conformance-user")
_bootstrap_application(service, actor)
service.register_welcome_protocol(
actor,
_prepared_welcome_protocol(),
correlation_id="corr-conf-protocol",
)
prepared = service.prepare_account(
actor,
tenant="tenant:coulomb",
required_factor_matches=(_email_requirement(),),
entitlements=(
PreparedEntitlement(
kind=PreparedEntitlementKind.MEMBERSHIP,
tenant="tenant:coulomb",
scope_type="realm",
scope_id="realm:citadel",
role="operator",
),
PreparedEntitlement(
kind=PreparedEntitlementKind.ONBOARDING_JOURNEY,
tenant="tenant:coulomb",
onboarding_journey="welcome-demo",
),
),
correlation_id="corr-conf-prepare",
)
registration = _complete_registration(service, actor)
claim = service.claim_prepared_account(
actor,
registration.session.registration_id,
prepared_account_id=prepared.prepared_account_id,
correlation_id="corr-conf-claim",
)
profile = service.register_access_profile(
actor,
_operator_access_profile(),
correlation_id="corr-conf-profile",
)
selection = service.select_active_hat(
actor,
registration.user.user_id,
profile.access_profile_id,
correlation_id="corr-conf-hat",
)
projection = service.projection(
actor,
registration.user.user_id,
ProjectionType.CLAIMS_ENRICHMENT,
application_id="app.demo",
tenant="tenant:coulomb",
correlation_id="corr-conf-projection",
)
context = service.identity_context(
actor,
user_id=registration.user.user_id,
tenant="tenant:coulomb",
application_id="app.demo",
correlation_id="corr-conf-context",
)
export = service.export_access_control_facts(
actor,
tenant="tenant:coulomb",
user_id=registration.user.user_id,
correlation_id="corr-conf-export",
)
ui_html = RegistrationAccessManagementUi(service).render_html(
RegistrationAccessManagementUi(service).admin_dashboard(
actor,
tenant="tenant:coulomb",
viewport=UiViewport.DESKTOP,
)
)
self.assertEqual(claim.memberships[0].kind, "operator")
self.assertEqual(selection.active_context.hat, "operator")
self.assertEqual(projection.access_context["active_hat"], "operator")
self.assertTrue(context.onboarding_journeys)
self.assertEqual(
context.onboarding_journeys[0].status,
OnboardingJourneyStatus.IN_PROGRESS,
)
self.assertIn("user", export.manifest["subject_types"])
self.assertIn(
"onboarding_journey.started",
[event.event_type for event in service.outbox_events()],
)
self.assertNotIn("sample.user@example.test", ui_html)
self.assertEqual(store.record_counts()["onboarding_journeys"], 1)
def test_security_negative_paths_fail_closed_with_audit_evidence(self):
service, store, _ = _service()
actor = _actor("security-user")
_bootstrap_application(service, actor)
registration = _complete_registration(service, actor)
other_user = service.create_user(
actor,
display_name="Other",
primary_email="other@example.test",
correlation_id="corr-other",
)
with self.assertRaises(ConflictError):
service.link_identity(
actor,
other_user.user_id,
issuer=actor.issuer,
subject=actor.subject,
correlation_id="corr-duplicate-identity",
)
with self.assertRaises(ValidationError):
service.prepare_account(
actor,
tenant="tenant:coulomb",
required_factor_matches=(
PreparedFactorRequirement(
factor_type=IdentityFactorType.EMAIL,
normalized_value=" ",
),
),
entitlements=(_membership_entitlement(),),
correlation_id="corr-weak-factor",
)
hijack = service.prepare_account(
actor,
tenant="tenant:coulomb",
required_factor_matches=(
PreparedFactorRequirement(
factor_type=IdentityFactorType.EMAIL,
normalized_value="victim@example.test",
),
),
entitlements=(_membership_entitlement(),),
correlation_id="corr-hijack-prepare",
)
with self.assertRaises(ValidationError):
service.claim_prepared_account(
actor,
registration.session.registration_id,
prepared_account_id=hijack.prepared_account_id,
correlation_id="corr-hijack-claim",
)
expired = service.prepare_account(
actor,
tenant="tenant:coulomb",
required_factor_matches=(_email_requirement("expired@example.test"),),
entitlements=(_membership_entitlement(scope_id="realm:expired"),),
expires_at=utc_now() - timedelta(days=1),
correlation_id="corr-expired-prepare",
)
with self.assertRaises(ValidationError):
service.claim_prepared_account(
actor,
registration.session.registration_id,
prepared_account_id=expired.prepared_account_id,
correlation_id="corr-expired-claim",
)
with self.assertRaises(AuthorizationDenied):
service.resolve_tenant_context(actor, "tenant:faraday")
privileged = service.prepare_account(
actor,
tenant="tenant:coulomb",
required_factor_matches=(_email_requirement(),),
entitlements=(
replace(
_membership_entitlement(scope_id="realm:privileged"),
role="admin",
requires_approval=True,
),
),
correlation_id="corr-privileged-prepare",
)
with self.assertRaises(AuthorizationDenied):
service.claim_prepared_account(
actor,
registration.session.registration_id,
prepared_account_id=privileged.prepared_account_id,
correlation_id="corr-privileged-claim",
)
access_profile = service.register_access_profile(
actor,
replace(_operator_access_profile(), requires_approval=True),
correlation_id="corr-stale-approval-profile",
)
with self.assertRaises(AuthorizationDenied):
service.select_active_hat(
actor,
registration.user.user_id,
access_profile.access_profile_id,
correlation_id="corr-stale-approval",
)
audit_summaries = [record.summary for record in service.audit_records()]
self.assertIn(
"prepared account claim denied: factor mismatch or closed",
audit_summaries,
)
self.assertIn(
"prepared account claim denied: approval required",
audit_summaries,
)
self.assertEqual(store.memberships_for_user(other_user.user_id), ())
def test_redaction_and_diagnostics_conformance(self):
service, _, _ = _service()
actor = _actor("redaction-user")
_bootstrap_application(service, actor, catalog=_sensitive_catalog())
registration = _complete_registration(service, actor)
service.set_profile_value(
actor,
registration.user.user_id,
"demo.recovery_hint",
"blue envelope",
tenant="tenant:coulomb",
correlation_id="corr-sensitive-profile",
)
service.prepare_account(
actor,
tenant="tenant:coulomb",
required_factor_matches=(_email_requirement(),),
entitlements=(_membership_entitlement(),),
primary_email="sample.user@example.test",
correlation_id="corr-redaction-prepare",
)
service.register_access_profile(
actor,
replace(
_operator_access_profile(),
claims={"policy_secret": "do-not-render"},
profile_defaults={"landing_hint": "do-not-render"},
),
correlation_id="corr-redaction-profile",
)
projection = service.projection(
actor,
registration.user.user_id,
ProjectionType.APPLICATION_RUNTIME,
application_id="app.demo",
tenant="tenant:coulomb",
correlation_id="corr-sensitive-projection",
)
access_diagnostics = service.access_profile_diagnostics(
actor,
tenant="tenant:coulomb",
correlation_id="corr-access-diagnostics",
)
admin_html = RegistrationAccessManagementUi(service).render_html(
RegistrationAccessManagementUi(service).admin_dashboard(
actor,
tenant="tenant:coulomb",
viewport=UiViewport.DESKTOP,
)
)
event_payloads = repr([event.payload for event in service.outbox_events()])
self.assertEqual(projection.values["demo.recovery_hint"], REDACTED)
self.assertNotIn("blue envelope", repr(access_diagnostics))
self.assertNotIn("do-not-render", repr(access_diagnostics))
self.assertNotIn("sample.user@example.test", admin_html)
self.assertNotIn("sample.user@example.test", event_payloads)
self.assertNotIn("blue envelope", event_payloads)
def test_adapter_conformance_harnesses_without_production_infrastructure(self):
store = InMemoryUserEngineStore()
authz = ScenarioAuthorizationHarness(
action_obligations={"access_control_facts.export": ("acl:sync",)}
)
service = UserEngineService(
store=store,
identity_adapter=StrictFixtureIdentityClaimsAdapter(),
authorization=authz,
factor_verifier=_FixtureFactorVerifier(),
)
actor = service.identity_adapter.normalize(human_actor_claims())
_bootstrap_application(service, actor)
registration = service.start_registration(actor, correlation_id="corr-adapter-start")
service.attach_registration_factor(
actor,
registration.registration_id,
{"type": "eid", "value": "EID-123", "secret": "strip-me"},
correlation_id="corr-adapter-factor",
)
service.attach_registration_factor(
actor,
registration.registration_id,
{"type": "email", "value": "sample.user@example.test"},
correlation_id="corr-adapter-email",
)
completed = service.complete_registration(
actor,
registration.registration_id,
correlation_id="corr-adapter-complete",
)
service.add_membership(
actor,
completed.user.user_id,
tenant="tenant:coulomb",
scope_type="group",
scope_id="group:research",
kind="member",
correlation_id="corr-adapter-group",
)
export = service.export_access_control_facts(
actor,
tenant="tenant:coulomb",
user_id=completed.user.user_id,
correlation_id="corr-adapter-export",
)
protocol = service.register_welcome_protocol(
actor,
WelcomeProtocol(
tenant="tenant:coulomb",
name="Adapter Handoff",
trigger_type=OnboardingTriggerType.MANUAL,
steps=(
WelcomeProtocolStep(
step_key="callback",
title="Callback",
subsystem="crm",
requires_subsystem_callback=True,
),
),
),
correlation_id="corr-adapter-protocol",
)
blocked = service.start_onboarding_journey(
actor,
completed.user.user_id,
protocol.protocol_id,
correlation_id="corr-adapter-onboarding",
).journey
resumed = service.resume_onboarding_journey(
actor,
blocked.journey_id,
callback_refs={"callback": "crm://welcome/callback"},
correlation_id="corr-adapter-resume",
)
self.assertIn("group", export.manifest["subject_types"])
self.assertTrue(any(request.action == "access_control_facts.export" for request in authz.requests))
self.assertNotIn("strip-me", repr(store.factors_for_user(completed.user.user_id)))
self.assertEqual(blocked.status, OnboardingJourneyStatus.BLOCKED)
self.assertEqual(resumed.status, OnboardingJourneyStatus.IN_PROGRESS)
self.assertTrue(service.audit_records())
self.assertTrue(service.outbox_events())
self.assertEqual(service.operability_snapshot().issues, ())
with self.assertRaises(ValidationError):
service.identity_adapter.normalize(missing_tenant_claims())
class _FixtureFactorVerifier:
def normalize(self, proofing_result):
factor_type = IdentityFactorType(str(proofing_result["type"]))
return FactorVerification(
factor_type=factor_type,
normalized_value=str(proofing_result["value"]).casefold(),
display_value=None,
source_system="fixture-proofing",
assurance={"level": "ial2" if factor_type == IdentityFactorType.EID else "ial1"},
)
def _service():
store = InMemoryUserEngineStore()
service = UserEngineService(
store=store,
identity_adapter=FixtureIdentityClaimsAdapter(),
authorization=LocalAuthorizationCheckPort(),
)
return service, store, service.authorization
def _actor(subject: str):
claims = human_actor_claims(subject=subject, tenant="tenant:coulomb")
claims["roles"] = ["tenant-admin"]
return FixtureIdentityClaimsAdapter().normalize(claims)
def _bootstrap_application(
service: UserEngineService,
actor,
*,
catalog: Catalog | None = None,
) -> None:
service.register_application(
actor,
sample_application(),
binding=sample_application_binding(),
correlation_id="corr-bootstrap-app",
)
service.publish_catalog(
actor,
catalog or _simple_catalog(),
correlation_id="corr-bootstrap-catalog",
)
def _complete_registration(service: UserEngineService, actor):
session = service.start_registration(actor, correlation_id="corr-reg-start")
service.attach_registration_factor(
actor,
session.registration_id,
FactorVerification(
factor_type=IdentityFactorType.EMAIL,
normalized_value="sample.user@example.test",
display_value="sample.user@example.test",
source_system="fixture-email",
),
correlation_id="corr-reg-factor",
)
return service.complete_registration(
actor,
session.registration_id,
correlation_id="corr-reg-complete",
)
def _email_requirement(email: str = "sample.user@example.test"):
return PreparedFactorRequirement(
factor_type=IdentityFactorType.EMAIL,
normalized_value=email,
)
def _membership_entitlement(scope_id: str = "realm:citadel"):
return PreparedEntitlement(
kind=PreparedEntitlementKind.MEMBERSHIP,
tenant="tenant:coulomb",
scope_type="realm",
scope_id=scope_id,
role="operator",
)
def _operator_access_profile() -> AccessProfile:
return AccessProfile(
tenant="tenant:coulomb",
display_name="Operator",
hat="operator",
scope_type=AccessScopeType.REALM,
scope_id="realm:citadel",
realm_id="realm:citadel",
service_id="app.demo",
membership_requirements=(
AccessMembershipRequirement(
scope_type="realm",
scope_id="realm:citadel",
kind="operator",
),
),
required_factor_types=(IdentityFactorType.EMAIL,),
claims={"service_role": "operator"},
)
def _prepared_welcome_protocol() -> WelcomeProtocol:
return WelcomeProtocol(
tenant="tenant:coulomb",
name="Prepared Welcome",
trigger_type=OnboardingTriggerType.PREPARED_ACCOUNT,
journey_key="welcome-demo",
prepared_journey="welcome-demo",
steps=(
WelcomeProtocolStep(
step_key="intro",
title="Intro",
subsystem="portal",
callback_ref="portal://welcome",
requires_subsystem_callback=True,
),
),
)
def _simple_catalog() -> Catalog:
return Catalog(
catalog_id="demo-profile",
namespace="demo",
version="0.1.0",
owning_application_id="app.demo",
lifecycle=CatalogLifecycle.ACTIVE,
attributes=(
AttributeDefinition(
key="demo.display_density",
value_type="string",
scope=ProfileScope.APPLICATION,
sensitivity=Sensitivity.INTERNAL,
visibility=(Visibility.USER, Visibility.APPLICATION),
mutability=(Mutability.USER,),
default="comfortable",
validation={"enum": ["compact", "comfortable"]},
),
),
)
def _sensitive_catalog() -> Catalog:
catalog = _simple_catalog()
return Catalog(
catalog_id=catalog.catalog_id,
namespace=catalog.namespace,
version=catalog.version,
owning_application_id=catalog.owning_application_id,
lifecycle=catalog.lifecycle,
attributes=(
*catalog.attributes,
AttributeDefinition(
key="demo.recovery_hint",
value_type="string",
scope=ProfileScope.GLOBAL,
sensitivity=Sensitivity.SENSITIVE,
visibility=(Visibility.USER, Visibility.APPLICATION, Visibility.ADMIN),
mutability=(Mutability.USER, Mutability.ADMIN),
),
),
)
if __name__ == "__main__":
unittest.main()