Start user-engine implementation scaffold

This commit is contained in:
2026-05-22 20:55:27 +02:00
parent e618b4e286
commit 58d9de26d3
14 changed files with 763 additions and 7 deletions

6
Makefile Normal file
View File

@@ -0,0 +1,6 @@
.PHONY: test
PYTHON ?= python3
test:
PYTHONPATH=src $(PYTHON) -m unittest discover -s tests -p 'test_*.py'

View File

@@ -1 +1,10 @@
Headless multi-application, multi-tenant user mangement engine.
Headless multi-application, multi-tenant user management engine.
## Development
```bash
make test
```
See `docs/development.md` and `docs/configuration.md` for the initial
implementation boundaries.

44
docs/configuration.md Normal file
View File

@@ -0,0 +1,44 @@
# Configuration Boundaries
## Standalone Mode
Standalone mode is for local development, tests, prototypes, and small
single-service deployments.
Expected characteristics:
- local configuration file or environment variables;
- local database or file-backed persistence during early development;
- fixture or local identity claims adapter;
- deterministic authorization test adapter;
- no password, MFA, or token issuance responsibility inside user-engine.
## Platform Mode
Platform mode is for a NetKingdom-aligned shared service deployment.
Expected characteristics:
- verified IAM Profile claims arrive from an identity layer;
- authorization decisions are requested through the authorization check port;
- runtime secrets are delivered through a scoped secret provider;
- audit records and outbox events are correlated with platform sinks;
- tenant and application bindings are explicit.
## Secret Names
The code should refer to logical secret names, not platform paths. Concrete
secret lookup is owned by the active `SecretProvider` adapter.
Initial logical names:
- `database.url`
- `event.signing_key`
- `webhook.shared_secret`
## Production Guardrails
- Local issuers must be rejected by production adapters.
- Sensitive writes must fail closed when authorization is unavailable.
- Claims enrichment must be optional and must not make user-engine a token
issuer.

42
docs/development.md Normal file
View File

@@ -0,0 +1,42 @@
# Development
## Stack
The initial implementation uses Python 3.12 and the standard library. The
first slice intentionally avoids runtime dependencies so the repository can be
tested immediately in local and agent environments.
## Layout
```text
src/user_engine/
domain/ transport- and persistence-neutral domain schemas
ports.py adapter protocols for identity, authorization, events, audit,
membership export, application bindings, and secrets
testing/ local fixtures for tests and examples
tests/ standard-library unittest suite
```
The domain layer should not import HTTP frameworks, database clients, or
platform-specific SDKs. Those integrations belong behind ports.
## Commands
```bash
make test
```
The command runs:
```bash
PYTHONPATH=src python3 -m unittest discover -s tests -p 'test_*.py'
```
## Implementation Rule
Add new behavior in this order:
1. domain schema or port;
2. local fixture or adapter;
3. test that proves the boundary;
4. infrastructure adapter or API surface.

9
pyproject.toml Normal file
View File

@@ -0,0 +1,9 @@
[project]
name = "user-engine"
version = "0.0.0"
description = "Headless user-domain and profile engine."
requires-python = ">=3.12"
[tool.user-engine]
package = "user_engine"
workplan = "USER-WP-0001"

View File

@@ -0,0 +1,5 @@
"""Headless user-domain and profile engine."""
__all__ = ["__version__"]
__version__ = "0.0.0"

View File

@@ -0,0 +1,53 @@
"""Domain schemas for user-engine."""
from user_engine.domain.models import (
Account,
AccountStatus,
Actor,
Application,
ApplicationBinding,
AttributeDefinition,
AuditRecord,
AuthorizationDecision,
AuthorizationEffect,
AuthorizationRequest,
Catalog,
CatalogLifecycle,
ExternalIdentity,
Membership,
Mutability,
OutboxEvent,
PrincipalType,
ProfileScope,
ProfileValue,
ProjectionType,
Sensitivity,
User,
Visibility,
)
__all__ = [
"Account",
"AccountStatus",
"Actor",
"Application",
"ApplicationBinding",
"AttributeDefinition",
"AuditRecord",
"AuthorizationDecision",
"AuthorizationEffect",
"AuthorizationRequest",
"Catalog",
"CatalogLifecycle",
"ExternalIdentity",
"Membership",
"Mutability",
"OutboxEvent",
"PrincipalType",
"ProfileScope",
"ProfileValue",
"ProjectionType",
"Sensitivity",
"User",
"Visibility",
]

View File

@@ -0,0 +1,276 @@
"""Core user-engine domain schemas.
These dataclasses are deliberately persistence- and transport-neutral. API
handlers, databases, and platform adapters should translate into and out of
these shapes instead of putting domain rules in infrastructure code.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import StrEnum
from typing import Any, Mapping
from uuid import uuid4
def new_id(prefix: str) -> str:
"""Return an opaque local identifier with a readable type prefix."""
return f"{prefix}_{uuid4().hex}"
def utc_now() -> datetime:
return datetime.now(timezone.utc)
class PrincipalType(StrEnum):
HUMAN = "human"
SERVICE = "service"
AGENT = "agent"
class AccountStatus(StrEnum):
INVITED = "invited"
ACTIVE = "active"
DISABLED = "disabled"
SUSPENDED = "suspended"
DELETION_PENDING = "deletion_pending"
DELETED = "deleted"
class ManagementMode(StrEnum):
LOCAL = "local"
EXTERNALLY_PROVISIONED = "externally_provisioned"
FEDERATED = "federated"
SERVICE_MANAGED = "service_managed"
class ProfileScope(StrEnum):
GLOBAL = "global"
TENANT = "tenant"
APPLICATION = "application"
MEMBERSHIP = "membership"
class ProjectionType(StrEnum):
SELF_SERVICE = "self_service"
ADMIN = "admin"
APPLICATION_RUNTIME = "application_runtime"
AUDIT = "audit"
AGENT_CONTEXT = "agent_context"
CLAIMS_ENRICHMENT = "claims_enrichment"
class Sensitivity(StrEnum):
PUBLIC = "public"
INTERNAL = "internal"
PERSONAL = "personal"
SENSITIVE = "sensitive"
SECRET = "secret"
class Visibility(StrEnum):
USER = "user"
ADMIN = "admin"
APPLICATION = "application"
SYSTEM = "system"
class Mutability(StrEnum):
USER = "user"
ADMIN = "admin"
APPLICATION = "application"
SYSTEM = "system"
READ_ONLY = "read_only"
class CatalogLifecycle(StrEnum):
DRAFT = "draft"
ACTIVE = "active"
DEPRECATED = "deprecated"
RETIRED = "retired"
class AuthorizationEffect(StrEnum):
ALLOW = "allow"
DENY = "deny"
REDACT = "redact"
AUDIT_ONLY = "audit_only"
NOT_APPLICABLE = "not_applicable"
@dataclass(frozen=True)
class Actor:
issuer: str
subject: str
tenant: str
principal_type: PrincipalType
audience: tuple[str, ...]
roles: tuple[str, ...] = ()
groups: tuple[str, ...] = ()
scopes: tuple[str, ...] = ()
assurance: Mapping[str, Any] = field(default_factory=dict)
authorized_party: str | None = None
preferred_username: str | None = None
claims: Mapping[str, Any] = field(default_factory=dict)
agent: Mapping[str, Any] = field(default_factory=dict)
@property
def identity_key(self) -> tuple[str, str]:
return (self.issuer, self.subject)
@dataclass(frozen=True)
class User:
user_id: str = field(default_factory=lambda: new_id("usr"))
display_name: str | None = None
primary_email: str | None = None
created_at: datetime = field(default_factory=utc_now)
@dataclass(frozen=True)
class Account:
account_id: str
user_id: str
status: AccountStatus = AccountStatus.INVITED
management_mode: ManagementMode = ManagementMode.LOCAL
created_at: datetime = field(default_factory=utc_now)
@dataclass(frozen=True)
class ExternalIdentity:
identity_id: str
user_id: str
issuer: str
subject: str
provider: str | None = None
linked_at: datetime = field(default_factory=utc_now)
@property
def identity_key(self) -> tuple[str, str]:
return (self.issuer, self.subject)
@dataclass(frozen=True)
class Application:
application_id: str
display_name: str
owner: str
allowed_profile_scopes: tuple[ProfileScope, ...] = (ProfileScope.GLOBAL,)
allowed_projection_types: tuple[ProjectionType, ...] = (
ProjectionType.APPLICATION_RUNTIME,
)
@dataclass(frozen=True)
class ApplicationBinding:
application_id: str
oidc_client_id: str | None = None
protected_system_id: str | None = None
catalog_namespaces: tuple[str, ...] = ()
event_source: str | None = None
deployment_ref: str | None = None
@dataclass(frozen=True)
class AttributeDefinition:
key: str
value_type: str
scope: ProfileScope
sensitivity: Sensitivity
visibility: tuple[Visibility, ...]
mutability: tuple[Mutability, ...]
required: bool = False
default: Any = None
validation: Mapping[str, Any] = field(default_factory=dict)
@dataclass(frozen=True)
class Catalog:
catalog_id: str
namespace: str
version: str
owning_application_id: str
lifecycle: CatalogLifecycle = CatalogLifecycle.DRAFT
attributes: tuple[AttributeDefinition, ...] = ()
def attribute_keys(self) -> set[str]:
return {attribute.key for attribute in self.attributes}
@dataclass(frozen=True)
class ProfileValue:
user_id: str
attribute_key: str
value: Any
scope: ProfileScope = ProfileScope.GLOBAL
scope_id: str | None = None
source: str = "user-engine"
updated_at: datetime = field(default_factory=utc_now)
@dataclass(frozen=True)
class Membership:
membership_id: str
user_id: str
tenant: str
scope_type: str
scope_id: str
kind: str
source_system: str = "user-engine"
owning_system: str = "user-engine"
freshness_version: str | None = None
created_at: datetime = field(default_factory=utc_now)
@dataclass(frozen=True)
class AuthorizationRequest:
actor: Actor
resource_type: str
resource_id: str
action: str
tenant: str
correlation_id: str
application_id: str | None = None
target_user_id: str | None = None
context: Mapping[str, Any] = field(default_factory=dict)
@dataclass(frozen=True)
class AuthorizationDecision:
effect: AuthorizationEffect
decision_id: str = field(default_factory=lambda: new_id("dec"))
reason: str | None = None
obligations: tuple[str, ...] = ()
@property
def allowed(self) -> bool:
return self.effect in {
AuthorizationEffect.ALLOW,
AuthorizationEffect.AUDIT_ONLY,
}
@dataclass(frozen=True)
class AuditRecord:
audit_id: str
actor: Actor
action: str
subject: str
tenant: str
correlation_id: str
decision_id: str | None = None
application_id: str | None = None
summary: str | None = None
recorded_at: datetime = field(default_factory=utc_now)
@dataclass(frozen=True)
class OutboxEvent:
event_id: str
event_type: str
aggregate_id: str
payload: Mapping[str, Any]
tenant: str
correlation_id: str
occurred_at: datetime = field(default_factory=utc_now)

83
src/user_engine/ports.py Normal file
View File

@@ -0,0 +1,83 @@
"""Implementation ports for user-engine adapters.
The domain layer should depend on these protocols. Concrete implementations
can be local test adapters, HTTP clients, database-backed stores, or platform
adapters without changing domain code.
"""
from __future__ import annotations
from typing import Any, Iterable, Mapping, Protocol
from user_engine.domain import (
Actor,
ApplicationBinding,
AuditRecord,
AuthorizationDecision,
AuthorizationRequest,
Membership,
OutboxEvent,
)
class IdentityClaimsAdapter(Protocol):
"""Normalize verified identity claims into a user-engine actor."""
def normalize(self, claims: Mapping[str, Any]) -> Actor:
"""Return a normalized actor from already-verified claims."""
def identity_key(self, actor: Actor) -> tuple[str, str]:
"""Return the stable external identity link key."""
class AuthorizationCheckPort(Protocol):
"""Ask whether an actor may perform an action."""
def check(self, request: AuthorizationRequest) -> AuthorizationDecision:
"""Return the authorization decision for one request."""
def batch_check(
self, requests: Iterable[AuthorizationRequest]
) -> tuple[AuthorizationDecision, ...]:
"""Return decisions in request order."""
class ApplicationBindingStore(Protocol):
"""Store links between user-engine applications and external systems."""
def get(self, application_id: str) -> ApplicationBinding | None:
"""Return a binding by user-engine application id."""
def save(self, binding: ApplicationBinding) -> None:
"""Create or replace an application binding."""
class MembershipFactExporter(Protocol):
"""Export membership facts as read models for authorization systems."""
def export(self, memberships: Iterable[Membership]) -> Mapping[str, Any]:
"""Return an adapter-neutral membership fact manifest."""
class EventOutbox(Protocol):
"""Persist and publish durable domain events."""
def append(self, event: OutboxEvent) -> None:
"""Append an event in the same unit of work as its mutation."""
def pending(self) -> tuple[OutboxEvent, ...]:
"""Return events waiting for delivery."""
class AuditWriter(Protocol):
"""Persist local audit records and support platform audit export."""
def record(self, audit_record: AuditRecord) -> None:
"""Persist an audit record."""
class SecretProvider(Protocol):
"""Load runtime secret material from the active environment."""
def get(self, name: str) -> str:
"""Return a secret value by logical name."""

View File

@@ -0,0 +1 @@
"""Testing helpers and local fixtures for user-engine."""

View File

@@ -0,0 +1,151 @@
"""Local fixtures used by early user-engine tests and examples."""
from __future__ import annotations
from typing import Iterable, Mapping
from user_engine.domain import (
Actor,
Application,
ApplicationBinding,
AttributeDefinition,
AuthorizationDecision,
AuthorizationEffect,
AuthorizationRequest,
Catalog,
CatalogLifecycle,
Mutability,
PrincipalType,
ProfileScope,
ProjectionType,
Sensitivity,
Visibility,
)
from user_engine.ports import AuthorizationCheckPort, IdentityClaimsAdapter
class FixtureIdentityClaimsAdapter:
"""Normalize dictionary fixtures that already passed token verification."""
def normalize(self, claims: Mapping[str, object]) -> Actor:
scopes = claims.get("scope", ())
if isinstance(scopes, str):
scopes = tuple(part for part in scopes.split(" ") if part)
return Actor(
issuer=str(claims["iss"]),
subject=str(claims["sub"]),
tenant=str(claims["tenant"]),
principal_type=PrincipalType(str(claims["principal_type"])),
audience=tuple(_as_tuple(claims.get("aud", ()))),
roles=tuple(_as_tuple(claims.get("roles", ()))),
groups=tuple(_as_tuple(claims.get("groups", ()))),
scopes=tuple(_as_tuple(scopes)),
assurance=dict(claims.get("assurance", {})),
authorized_party=_optional_str(claims.get("azp") or claims.get("client_id")),
preferred_username=_optional_str(claims.get("preferred_username")),
claims=dict(claims),
agent=dict(claims.get("agent", {})),
)
def identity_key(self, actor: Actor) -> tuple[str, str]:
return actor.identity_key
class StaticAuthorizationCheckPort:
"""Deterministic authorization adapter for tests and local examples."""
def __init__(self, effect: AuthorizationEffect = AuthorizationEffect.ALLOW):
self.effect = effect
self.requests: list[AuthorizationRequest] = []
def check(self, request: AuthorizationRequest) -> AuthorizationDecision:
self.requests.append(request)
return AuthorizationDecision(effect=self.effect, reason="fixture")
def batch_check(
self, requests: Iterable[AuthorizationRequest]
) -> tuple[AuthorizationDecision, ...]:
return tuple(self.check(request) for request in requests)
def human_actor_claims(
*,
issuer: str = "https://issuer.example.test",
subject: str = "user-123",
tenant: str = "tenant:coulomb",
) -> dict[str, object]:
return {
"iss": issuer,
"sub": subject,
"aud": ["user-engine"],
"tenant": tenant,
"principal_type": "human",
"groups": ["tenant:coulomb:users"],
"roles": ["user"],
"scope": "openid profile email",
"preferred_username": "sample.user",
"email": "sample.user@example.test",
"assurance": {
"level": "aal2",
"methods": ["pwd", "otp"],
"mfa": True,
"source": "fixture",
},
}
def sample_application() -> Application:
return Application(
application_id="app.demo",
display_name="Demo Application",
owner="team:demo",
allowed_profile_scopes=(ProfileScope.GLOBAL, ProfileScope.APPLICATION),
allowed_projection_types=(ProjectionType.APPLICATION_RUNTIME,),
)
def sample_application_binding() -> ApplicationBinding:
return ApplicationBinding(
application_id="app.demo",
oidc_client_id="demo-client",
protected_system_id="user-engine.demo",
catalog_namespaces=("demo",),
event_source="user-engine.demo",
deployment_ref="local",
)
def sample_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 _as_tuple(value: object) -> tuple[str, ...]:
if value is None:
return ()
if isinstance(value, str):
return (value,)
return tuple(str(item) for item in value)
def _optional_str(value: object) -> str | None:
if value is None:
return None
return str(value)

View File

@@ -0,0 +1,28 @@
import unittest
from user_engine.domain import (
AuthorizationDecision,
AuthorizationEffect,
CatalogLifecycle,
)
from user_engine.testing.fixtures import sample_catalog
class DomainModelTests(unittest.TestCase):
def test_catalog_exposes_attribute_keys(self):
catalog = sample_catalog()
self.assertEqual(catalog.lifecycle, CatalogLifecycle.ACTIVE)
self.assertEqual(catalog.attribute_keys(), {"demo.display_density"})
def test_authorization_decision_allowed_property(self):
self.assertTrue(
AuthorizationDecision(effect=AuthorizationEffect.ALLOW).allowed
)
self.assertFalse(
AuthorizationDecision(effect=AuthorizationEffect.DENY).allowed
)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,49 @@
import unittest
from user_engine.domain import AuthorizationEffect, AuthorizationRequest
from user_engine.testing.fixtures import (
FixtureIdentityClaimsAdapter,
StaticAuthorizationCheckPort,
human_actor_claims,
sample_application_binding,
)
class PortFixtureTests(unittest.TestCase):
def test_fixture_identity_claims_adapter_normalizes_actor(self):
adapter = FixtureIdentityClaimsAdapter()
actor = adapter.normalize(human_actor_claims())
self.assertEqual(actor.identity_key, ("https://issuer.example.test", "user-123"))
self.assertEqual(actor.tenant, "tenant:coulomb")
self.assertIn("profile", actor.scopes)
def test_static_authorization_check_records_requests(self):
adapter = FixtureIdentityClaimsAdapter()
actor = adapter.normalize(human_actor_claims())
authz = StaticAuthorizationCheckPort(effect=AuthorizationEffect.ALLOW)
request = AuthorizationRequest(
actor=actor,
resource_type="user-engine:profile",
resource_id="usr_123",
action="read",
tenant="tenant:coulomb",
correlation_id="corr-1",
target_user_id="usr_123",
)
decision = authz.check(request)
self.assertTrue(decision.allowed)
self.assertEqual(authz.requests, [request])
def test_sample_application_binding_keeps_external_ids_separate(self):
binding = sample_application_binding()
self.assertEqual(binding.application_id, "app.demo")
self.assertEqual(binding.oidc_client_id, "demo-client")
self.assertEqual(binding.protected_system_id, "user-engine.demo")
if __name__ == "__main__":
unittest.main()

View File

@@ -4,7 +4,7 @@ type: workplan
title: "User Engine Preparation And Interface Adoption"
domain: netkingdom
repo: user-engine
status: ready
status: finished
owner: codex
topic_slug: netkingdom
planning_priority: high
@@ -25,7 +25,7 @@ from the PRD, architecture blueprint, and NetKingdom interface guidance.
```task
id: USER-WP-0001-T1
status: todo
status: done
priority: high
state_hub_task_id: "7f91d466-2f97-4acf-832a-e9de0b21be04"
```
@@ -35,7 +35,7 @@ test commands, and module boundaries.
```task
id: USER-WP-0001-T2
status: todo
status: done
priority: high
state_hub_task_id: "940ce697-fbad-4ef9-9024-175f482b2d3a"
```
@@ -46,7 +46,7 @@ audit writer, and secret provider.
```task
id: USER-WP-0001-T3
status: todo
status: done
priority: high
state_hub_task_id: "c16bd102-0f86-4951-86ee-8d6088ac888b"
```
@@ -56,7 +56,7 @@ catalogs, profile values, memberships, audit records, and outbox events.
```task
id: USER-WP-0001-T4
status: todo
status: done
priority: medium
state_hub_task_id: "9e962947-6d1f-49ae-a362-d8a13d430b87"
```
@@ -66,7 +66,7 @@ authorization decisions.
```task
id: USER-WP-0001-T5
status: todo
status: done
priority: medium
state_hub_task_id: "c5b24341-ae26-48a0-97ac-6fbcf108b0e0"
```