diff --git a/docs/configuration.md b/docs/configuration.md index 2aecaef..a2be3cd 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -83,6 +83,41 @@ contacted; the platform event never carries `client_id`. An invalid value stops startup — it never falls back to the legacy shape. +## Application access readback + +Off unless `USER_ENGINE_READBACK_URL` is set. When it is set, these are also +needed: + +- `USER_ENGINE_READBACK_URL`: base URL of access-engine's dedicated + `flex-auth-application-readback` pin; +- `USER_ENGINE_READBACK_TOKEN_FILE`: the readback caller token, read on every + call. This is a separate file from `USER_ENGINE_FLEX_AUTH_TOKEN_FILE`. The pin + admits it only under the `application-readback` binding, so it can never act + on `/v1/check`; +- `USER_ENGINE_READBACK_TIMEOUT`: optional, in seconds (default `3`). + +The account page (`/onboarding`) then lists each workload bound in all three +records: + +- the user-engine application record; +- its binding's `protected_system_id`; +- the readback system's `client_binding`, whose issuer and client id must equal + our OIDC issuer and the binding's `oidc_client_id`. + +The rows are rendered as follows. No row is ever inferred as allow: + +| access-engine answer | Shown | +| --- | --- | +| `allow` before `expires_at` | Available | +| `deny` | Not available | +| `not_applicable` | Not available: not offered here | +| failure, missing row, or `allow` past or without `expires_at` | Not available: could not be checked | + +Answers are read only on account-page views. They are reused per session and +subject for at most 60 seconds, and never past the earliest allow's +`expires_at`. Logout drops them. A failed read is not reused. Contract: +access-engine `docs/application-access-readback.md` (FLEX-DEC-2026-019). + ## Authenticator management handoff `USER_ENGINE_MFA_MANAGEMENT_URL` is optional and empty by default. When set, the diff --git a/layer.yaml b/layer.yaml index f652b8e..5da25ee 100644 --- a/layer.yaml +++ b/layer.yaml @@ -64,6 +64,14 @@ non_tooling_clients: write: false note: Engine API. The shape §5 prescribes for a PEP-shaped consumer. + - id: access-engine-application-readback + target: access-engine + layer: engine + module: src/user_engine/adapters/application_access.py + operation: "HTTP GET /v1/readback/systems and POST /v1/readback/application_access" + write: false + note: "Engine API, dedicated readback pin (FLEX-DEC-2026-019). Display only; readback decisions authorize nothing and no row is inferred as allow." + - id: oidc-claim-input target: key-cape layer: tooling-as-claim-input diff --git a/src/user_engine/adapters/application_access.py b/src/user_engine/adapters/application_access.py new file mode 100644 index 0000000..d5fecb7 --- /dev/null +++ b/src/user_engine/adapters/application_access.py @@ -0,0 +1,128 @@ +"""access-engine application access readback HTTP adapter. + +Calls the dedicated readback pin with its own caller token. The token file is +separate from the ``/v1/check`` caller token and never authorizes a decision. +Any failure raises ``ReadbackUnavailable``; callers render "could not be checked". +""" + +from __future__ import annotations + +import json +from collections.abc import Sequence +from datetime import datetime +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +from user_engine.application_access import ( + MAX_SYSTEMS_PER_REQUEST, + ReadbackResult, + ReadbackSystem, + ReadbackUnavailable, +) +from user_engine.domain import Actor + +_EFFECTS = frozenset({"allow", "deny", "not_applicable"}) + + +class ApplicationAccessReadbackHTTPAdapter: + """Read ``/v1/readback/systems`` and ``/v1/readback/application_access``.""" + + def __init__( + self, + *, + base_url: str, + bearer_token_file: str, + timeout_seconds: float = 3.0, + ) -> None: + self.base_url = base_url.rstrip("/") + self.bearer_token_file = bearer_token_file + self.timeout_seconds = timeout_seconds + + def systems(self) -> tuple[ReadbackSystem, ...]: + body = self._call("GET", "/v1/readback/systems") + try: + systems = [] + for item in body["systems"]: + binding = item.get("client_binding") or {} + systems.append(ReadbackSystem( + system_id=str(item["id"]), + name=str(item.get("name") or item["id"]), + readback=bool(item["readback"]), + client_issuer=str(binding["issuer"]) if binding else None, + client_id=str(binding["client_id"]) if binding else None, + )) + return tuple(systems) + except (KeyError, TypeError, AttributeError) as error: + raise ReadbackUnavailable("malformed readback systems response") from error + + def read(self, actor: Actor, systems: Sequence[str]) -> tuple[ReadbackResult, ...]: + requested = list(dict.fromkeys(systems)) + if not 1 <= len(requested) <= MAX_SYSTEMS_PER_REQUEST: + raise ValueError("readback requires 1 to 64 unique systems") + payload = { + "tenant": actor.tenant, + # Same subject shape as FlexAuthHTTPAdapter's /v1/check request. + "subject": { + "id": actor.subject, + "type": actor.principal_type.value, + "tenant": actor.tenant, + "attributes": { + "issuer": actor.issuer, + "roles": list(actor.roles), + "tenant_roles": list(actor.tenant_roles), + "groups": list(actor.groups), + "scopes": list(actor.scopes), + "assurance": dict(actor.assurance), + }, + }, + "systems": requested, + } + body = self._call("POST", "/v1/readback/application_access", payload) + try: + results = [] + for item in body["results"]: + effect = str(item["effect"]) + if effect not in _EFFECTS: + raise ValueError(f"unknown readback effect {effect!r}") + lifetime = item.get("lifetime") or {} + results.append(ReadbackResult( + system_id=str(item["system"]), + effect=effect, + expires_at=_timestamp(lifetime.get("expires_at")), + decision_id=item.get("decision_id"), + )) + except (KeyError, TypeError, ValueError, AttributeError) as error: + raise ReadbackUnavailable("malformed readback response") from error + if [result.system_id for result in results] != requested: + raise ReadbackUnavailable("readback results do not match the request") + return tuple(results) + + def _call(self, method: str, path: str, payload: Any = None) -> Any: + try: + # Read per call so projected ServiceAccount-token rotation needs no restart. + with open(self.bearer_token_file, encoding="utf-8") as token_file: + token = token_file.read().strip() + if not token: + raise ValueError("empty readback caller token") + headers = {"Authorization": f"Bearer {token}"} + data = None + if payload is not None: + headers["Content-Type"] = "application/json" + data = json.dumps(payload).encode() + response = urlopen( + Request(self.base_url + path, data=data, headers=headers, method=method), + timeout=self.timeout_seconds, + ) + return json.load(response) + except (HTTPError, URLError, TimeoutError, OSError, ValueError) as error: + raise ReadbackUnavailable("application access readback unavailable") from error + + +def _timestamp(value: Any) -> datetime | None: + if not value: + return None + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + if parsed.tzinfo is None: + raise ValueError("readback expires_at must carry a timezone") + return parsed diff --git a/src/user_engine/adapters/local.py b/src/user_engine/adapters/local.py index c5290ba..5724675 100644 --- a/src/user_engine/adapters/local.py +++ b/src/user_engine/adapters/local.py @@ -158,6 +158,9 @@ class InMemoryUserEngineStore: def binding(self, application_id: str) -> ApplicationBinding | None: return self.bindings.get(application_id) + def all_bindings(self) -> tuple[ApplicationBinding, ...]: + return tuple(self.bindings.values()) + def save_catalog(self, catalog: Catalog) -> None: self.catalogs[catalog.catalog_id] = catalog diff --git a/src/user_engine/adapters/postgres.py b/src/user_engine/adapters/postgres.py index 5325d09..10df95b 100644 --- a/src/user_engine/adapters/postgres.py +++ b/src/user_engine/adapters/postgres.py @@ -233,6 +233,9 @@ class PostgresUserEngineStore: def catalog(self, catalog_id: str) -> Catalog | None: return cast(Catalog | None, self._get_record("catalogs", catalog_id)) + def all_bindings(self) -> tuple[ApplicationBinding, ...]: + return cast(tuple[ApplicationBinding, ...], self._query_records("application_bindings")) + def all_catalogs(self) -> tuple[Catalog, ...]: return cast(tuple[Catalog, ...], self._query_records("catalogs")) diff --git a/src/user_engine/application_access.py b/src/user_engine/application_access.py new file mode 100644 index 0000000..8d1efda --- /dev/null +++ b/src/user_engine/application_access.py @@ -0,0 +1,181 @@ +"""Application access readback: which bound workloads a signed-in person may use. + +access-engine is the only decision point. This module joins its readback +answers to the workload catalogue and renders them. It never infers an allow: +a missing, failed, expired, denied or not-applicable answer is "not available". +Contract: access-engine ``docs/application-access-readback.md`` +(FLEX-DEC-2026-019). +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from datetime import datetime, timedelta + +from user_engine.domain import Actor, Application, ApplicationBinding, utc_now + +MAX_SYSTEMS_PER_REQUEST = 64 +DEFAULT_CACHE_TTL = timedelta(seconds=60) + + +class ReadbackUnavailable(Exception): + """access-engine returned no usable readback answer.""" + + +@dataclass(frozen=True) +class ReadbackSystem: + system_id: str + name: str + readback: bool + client_issuer: str | None = None + client_id: str | None = None + + +@dataclass(frozen=True) +class ReadbackResult: + system_id: str + effect: str + expires_at: datetime | None = None + decision_id: str | None = None + + +@dataclass(frozen=True) +class Workload: + """A workload bound in all three records: user-engine, key-cape, access-engine.""" + + application_id: str + display_name: str + system_id: str + + +@dataclass(frozen=True) +class ApplicationAccessRow: + workload: Workload + # available | denied | not_offered | unchecked + status: str + expires_at: datetime | None = None + + @property + def available(self) -> bool: + return self.status == "available" + + +def workload_catalogue( + applications: Sequence[tuple[Application, ApplicationBinding]], + systems: Sequence[ReadbackSystem], + *, + issuer: str, +) -> tuple[Workload, ...]: + """List only workloads whose three records exist and agree.""" + by_id = {system.system_id: system for system in systems} + expected_issuer = issuer.rstrip("/") + workloads = [] + for application, binding in applications: + system = by_id.get(binding.protected_system_id or "") + if ( + system is None + or not binding.oidc_client_id + or system.client_id != binding.oidc_client_id + or (system.client_issuer or "").rstrip("/") != expected_issuer + ): + continue + workloads.append(Workload( + application_id=application.application_id, + display_name=application.display_name or system.name or system.system_id, + system_id=system.system_id, + )) + return tuple(sorted(workloads, key=lambda item: (item.display_name, item.application_id))) + + +def render_rows( + workloads: Sequence[Workload], + results: Sequence[ReadbackResult] | None, + *, + now: datetime, +) -> tuple[ApplicationAccessRow, ...]: + by_system = {result.system_id: result for result in results or ()} + rows = [] + for workload in workloads: + result = by_system.get(workload.system_id) + if result is None: + status, expires_at = "unchecked", None + elif result.effect == "allow": + live = result.expires_at is not None and result.expires_at > now + status, expires_at = ("available", result.expires_at) if live else ("unchecked", None) + elif result.effect == "deny": + status, expires_at = "denied", None + elif result.effect == "not_applicable": + status, expires_at = "not_offered", None + else: + status, expires_at = "unchecked", None + rows.append(ApplicationAccessRow(workload, status, expires_at)) + return tuple(rows) + + +@dataclass +class _CacheEntry: + workloads: tuple[Workload, ...] + results: tuple[ReadbackResult, ...] | None + reuse_until: datetime + + +class ApplicationAccessReader: + """Read and reuse answers per session and subject, never past expiry.""" + + def __init__( + self, + readback: object, + *, + issuer: str, + bound_applications: Callable[[], Sequence[tuple[Application, ApplicationBinding]]], + cache_ttl: timedelta = DEFAULT_CACHE_TTL, + clock: Callable[[], datetime] = utc_now, + ) -> None: + self.readback = readback + self.issuer = issuer + self.bound_applications = bound_applications + self.cache_ttl = cache_ttl + self.clock = clock + self._cache: dict[tuple[str, str, str], _CacheEntry] = {} + + def rows(self, actor: Actor, session_key: str) -> tuple[ApplicationAccessRow, ...] | None: + """Return rendered rows, or None when the catalogue itself is unreadable.""" + now = self.clock() + key = (session_key, actor.issuer, actor.subject) + entry = self._cache.get(key) + if entry is None or entry.reuse_until <= now: + entry = self._read(actor, now) + if entry is None: + self._cache.pop(key, None) + return None + self._cache[key] = entry + return render_rows(entry.workloads, entry.results, now=now) + + def forget(self, session_key: str) -> None: + for key in [key for key in self._cache if key[0] == session_key]: + del self._cache[key] + + def _read(self, actor: Actor, now: datetime) -> _CacheEntry | None: + try: + systems = self.readback.systems() + except ReadbackUnavailable: + return None + workloads = workload_catalogue( + self.bound_applications(), systems, issuer=self.issuer + )[:MAX_SYSTEMS_PER_REQUEST] + results: tuple[ReadbackResult, ...] | None = () + if workloads: + try: + results = tuple(self.readback.read( + actor, [workload.system_id for workload in workloads] + )) + except ReadbackUnavailable: + results = None + reuse_until = now + self.cache_ttl + for result in results or (): + if result.effect == "allow" and result.expires_at is not None: + reuse_until = min(reuse_until, result.expires_at) + if results is None: + reuse_until = now + return _CacheEntry(workloads, results, reuse_until) diff --git a/src/user_engine/ports.py b/src/user_engine/ports.py index bb8d783..38597d9 100644 --- a/src/user_engine/ports.py +++ b/src/user_engine/ports.py @@ -9,7 +9,7 @@ from __future__ import annotations from contextlib import AbstractContextManager from dataclasses import dataclass -from typing import Any, Iterable, Mapping, Protocol +from typing import Any, Iterable, Mapping, Protocol, Sequence from user_engine.domain import ( Account, @@ -303,6 +303,9 @@ class UserEngineStore(Protocol): def binding(self, application_id: str) -> ApplicationBinding | None: """Return an application binding by application id.""" + def all_bindings(self) -> tuple[ApplicationBinding, ...]: + """Return all application bindings.""" + def save_catalog(self, catalog: Catalog) -> None: """Create or replace a catalog.""" @@ -591,3 +594,16 @@ class SecretProvider(Protocol): def get(self, name: str) -> str: """Return a secret value by logical name.""" + + +class ApplicationAccessReadbackPort(Protocol): + """Read access-engine application access answers; never decide locally. + + Both calls raise ``ReadbackUnavailable`` when no usable answer exists. + """ + + def systems(self) -> tuple[Any, ...]: + """Return the readback pin's registry projection (``ReadbackSystem``).""" + + def read(self, actor: Actor, systems: Sequence[str]) -> tuple[Any, ...]: + """Return one ``ReadbackResult`` per requested system.""" diff --git a/src/user_engine/runtime.py b/src/user_engine/runtime.py index dc65648..38f2bda 100644 --- a/src/user_engine/runtime.py +++ b/src/user_engine/runtime.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from typing import Any from wsgiref.simple_server import make_server from user_engine.adapters import ( @@ -145,6 +146,29 @@ def create_application() -> PortalApplication: hub_authority_token=os.environ.get("USER_ENGINE_HUB_AUTHORITY_TOKEN", "").strip(), ) + if os.environ.get("USER_ENGINE_READBACK_URL"): + from user_engine.adapters.application_access import ( + ApplicationAccessReadbackHTTPAdapter, + ) + from user_engine.application_access import ApplicationAccessReader + + def bound_applications() -> list[tuple[Any, Any]]: + return [ + (application, binding) + for binding in store.all_bindings() + if (application := store.application(binding.application_id)) is not None + ] + + app.application_access = ApplicationAccessReader( + ApplicationAccessReadbackHTTPAdapter( + base_url=_required("USER_ENGINE_READBACK_URL"), + bearer_token_file=_required("USER_ENGINE_READBACK_TOKEN_FILE"), + timeout_seconds=float(os.environ.get("USER_ENGINE_READBACK_TIMEOUT", "3")), + ), + issuer=_required("USER_ENGINE_OIDC_ISSUER"), + bound_applications=bound_applications, + ) + from user_engine.operations_status import check_services app.operations_probe = lambda: check_services(app.oidc_client, app.outbox_delivery) from user_engine.authentication_policy import PolicyClient diff --git a/src/user_engine/web.py b/src/user_engine/web.py index f6c0ad3..d12ef55 100644 --- a/src/user_engine/web.py +++ b/src/user_engine/web.py @@ -94,6 +94,7 @@ class PortalApplication: identity_lookup: Any | None = None, hub_authority_caller: Actor | None = None, hub_authority_token: str = "", + application_access: Any | None = None, ) -> None: if len(trusted_proxy_secret) < 24: raise ValueError("trusted proxy secret must contain at least 24 characters") @@ -136,6 +137,7 @@ class PortalApplication: raise ValueError("hub authority lookup needs both a caller and a token, or neither") self.hub_authority_caller = hub_authority_caller self.hub_authority_token = hub_authority_token + self.application_access = application_access def __call__(self, environ: Mapping[str, Any], start_response: StartResponse) -> Iterable[bytes]: correlation_id = environ.get("HTTP_X_REQUEST_ID") or f"corr_{secrets.token_hex(12)}" @@ -311,6 +313,8 @@ class PortalApplication: if path == "/logout" and method == "POST": session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session") body = self._form_body(environ) + if session_id and self.application_access is not None: + self.application_access.forget(session_id) if session_id and self.oidc_client and self.oidc_client.claims(session_id) is not None: self._require_csrf(environ, str(body.get("csrf_token", ""))) self.oidc_client.logout(session_id) @@ -444,6 +448,10 @@ class PortalApplication: start_response, self._onboarding( session, memberships, journeys, self._csrf_token(environ), + application_access=self._application_access_section( + session.actor, + cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session") or "", + ), ), correlation_id, ) @@ -2273,7 +2281,7 @@ Use the login name they provide; it may differ from your display name.

str: journey_items = "".join( self._onboarding_journey_item(item, csrf_token) for item in journeys @@ -2292,6 +2300,7 @@ Use the login name they provide; it may differ from your display name.

Save profile {self._allowed_tenants(actor, memberships)} {self._allowed_workloads(memberships)} +{application_access}

Onboarding progress

""", ) @@ -2426,6 +2435,35 @@ Use the login name they provide; it may differ from your display name.

Each application checks access when you open it. Tenant membership alone does not grant access to every application.

" ) + _APPLICATION_ACCESS_TEXT = { + "available": "Available", + "denied": "Not available", + "not_offered": "Not available: not offered here", + "unchecked": "Not available: could not be checked", + } + + def _application_access_section(self, actor: Any, session_key: str) -> str: + """Render access-engine readback answers; never infer an allow.""" + if self.application_access is None: + return "" + rows = self.application_access.rows(actor, session_key) + if rows is None: + items = "
  • Application access could not be checked. Try again later.
    • " + elif not rows: + items = "
  • No application is registered for an access check.
    • " + else: + items = "".join( + f'
  • {escape(row.workload.display_name)} - ' + f"{escape(self._APPLICATION_ACCESS_TEXT[row.status])}
    • " + for row in rows + ) + return ( + '

    Application access

    ' + f'' + "

    Checked with the access service for your current sign-in. Each application checks again when you open it. " + "This list covers registered applications only and is not a complete inventory.

    " + ) + @staticmethod def _onboarding_journey_item(journey: Any, csrf_token: str) -> str: steps = "".join( diff --git a/tests/test_application_access.py b/tests/test_application_access.py new file mode 100644 index 0000000..514e669 --- /dev/null +++ b/tests/test_application_access.py @@ -0,0 +1,330 @@ +import io +import json +import os +import tempfile +import unittest +from datetime import UTC, datetime, timedelta +from unittest.mock import patch +from urllib.error import HTTPError, URLError + +from user_engine.adapters.application_access import ApplicationAccessReadbackHTTPAdapter +from user_engine.adapters.flex_auth import FlexAuthHTTPAdapter +from user_engine.application_access import ( + ApplicationAccessReader, + ReadbackResult, + ReadbackSystem, + ReadbackUnavailable, + Workload, + render_rows, + workload_catalogue, +) +from user_engine.domain import ( + Actor, + Application, + ApplicationBinding, + AuthorizationRequest, + PrincipalType, +) + +ISSUER = "https://key-cape.example/realms/coulomb" +NOW = datetime(2026, 10, 10, 12, 0, tzinfo=UTC) + + +class _Response(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *_args): + self.close() + + +def _actor(subject="subject-1"): + return Actor( + issuer=ISSUER, subject=subject, tenant="tenant-a", + principal_type=PrincipalType.HUMAN, audience=("user-engine",), + roles=("member",), tenant_roles=("CUS",), groups=("tenant:a:users",), + scopes=("openid",), assurance={"level": "aal2", "mfa": True}, + ) + + +def _bound(app_id, system_id, client_id): + return ( + Application(application_id=app_id, display_name=app_id.title(), owner="owner"), + ApplicationBinding(application_id=app_id, oidc_client_id=client_id, + protected_system_id=system_id), + ) + + +def _system(system_id, client_id, issuer=ISSUER): + return ReadbackSystem(system_id=system_id, name=system_id, readback=True, + client_issuer=issuer, client_id=client_id) + + +class WorkloadCatalogueTests(unittest.TestCase): + def test_lists_only_workloads_bound_in_all_three_records(self): + applications = [ + _bound("informed", "informed-decision", "informed-client"), + _bound("unbound", None, "unbound-client"), + _bound("wrongclient", "wrong-client", "client-a"), + _bound("wrongissuer", "other-issuer", "client-b"), + _bound("nobinding", "no-client-binding", "client-c"), + _bound("unknown", "not-in-registry", "client-d"), + ] + systems = [ + _system("informed-decision", "informed-client"), + _system("wrong-client", "client-z"), + _system("other-issuer", "client-b", issuer="https://elsewhere"), + ReadbackSystem("no-client-binding", "x", readback=True), + ] + workloads = workload_catalogue(applications, systems, issuer=ISSUER + "/") + self.assertEqual([item.system_id for item in workloads], ["informed-decision"]) + + +class RenderTests(unittest.TestCase): + workloads = tuple(Workload(f"app-{n}", f"App {n}", f"sys-{n}") for n in range(5)) + + def test_no_row_is_inferred_as_allow(self): + rows = render_rows(self.workloads, [ + ReadbackResult("sys-0", "allow", NOW + timedelta(seconds=30)), + ReadbackResult("sys-1", "allow", NOW - timedelta(seconds=1)), + ReadbackResult("sys-2", "deny"), + ReadbackResult("sys-3", "not_applicable"), + ], now=NOW) + self.assertEqual( + [row.status for row in rows], + ["available", "unchecked", "denied", "not_offered", "unchecked"], + ) + self.assertEqual([row.available for row in rows], [True, False, False, False, False]) + + def test_allow_without_expiry_is_not_available(self): + rows = render_rows(self.workloads[:1], [ReadbackResult("sys-0", "allow")], now=NOW) + self.assertEqual(rows[0].status, "unchecked") + + def test_failed_read_renders_every_row_unchecked(self): + rows = render_rows(self.workloads, None, now=NOW) + self.assertEqual({row.status for row in rows}, {"unchecked"}) + + +class _FakeReadback: + def __init__(self, results, systems=None, fail_read=False, fail_systems=False): + self.results = results + self._systems = systems or [_system("informed-decision", "informed-client")] + self.fail_read = fail_read + self.fail_systems = fail_systems + self.reads = [] + + def systems(self): + if self.fail_systems: + raise ReadbackUnavailable("down") + return tuple(self._systems) + + def read(self, actor, systems): + self.reads.append((actor.subject, tuple(systems))) + if self.fail_read: + raise ReadbackUnavailable("down") + return tuple(self.results) + + +class ReaderTests(unittest.TestCase): + def setUp(self): + self.now = NOW + self.applications = [_bound("informed", "informed-decision", "informed-client")] + + def _reader(self, readback): + return ApplicationAccessReader( + readback, issuer=ISSUER, bound_applications=lambda: self.applications, + clock=lambda: self.now, + ) + + def test_reuses_rows_until_allow_expiry_and_never_past_it(self): + readback = _FakeReadback([ + ReadbackResult("informed-decision", "allow", NOW + timedelta(seconds=30), "decision:1"), + ]) + reader = self._reader(readback) + self.assertEqual(reader.rows(_actor(), "s1")[0].status, "available") + self.now = NOW + timedelta(seconds=20) + self.assertEqual(reader.rows(_actor(), "s1")[0].status, "available") + self.assertEqual(len(readback.reads), 1) + self.now = NOW + timedelta(seconds=30) + reader.rows(_actor(), "s1") + self.assertEqual(len(readback.reads), 2) + + def test_cache_is_per_subject_and_dropped_on_forget(self): + readback = _FakeReadback([ReadbackResult("informed-decision", "deny")]) + reader = self._reader(readback) + reader.rows(_actor("alice"), "s1") + reader.rows(_actor("bob"), "s1") + self.assertEqual([item[0] for item in readback.reads], ["alice", "bob"]) + reader.forget("s1") + reader.rows(_actor("alice"), "s1") + self.assertEqual(len(readback.reads), 3) + + def test_read_failure_is_unchecked_and_not_reused(self): + readback = _FakeReadback([], fail_read=True) + reader = self._reader(readback) + self.assertEqual(reader.rows(_actor(), "s1")[0].status, "unchecked") + reader.rows(_actor(), "s1") + self.assertEqual(len(readback.reads), 2) + + def test_unreadable_registry_returns_none(self): + reader = self._reader(_FakeReadback([], fail_systems=True)) + self.assertIsNone(reader.rows(_actor(), "s1")) + + def test_no_bound_workload_makes_no_decision_request(self): + self.applications = [] + readback = _FakeReadback([]) + self.assertEqual(self._reader(readback).rows(_actor(), "s1"), ()) + self.assertEqual(readback.reads, []) + + +class HTTPAdapterTests(unittest.TestCase): + def setUp(self): + handle, self.token_file = tempfile.mkstemp() + os.write(handle, b"readback-token\n") + os.close(handle) + self.addCleanup(os.unlink, self.token_file) + self.adapter = ApplicationAccessReadbackHTTPAdapter( + base_url="http://readback/", bearer_token_file=self.token_file, + ) + + def _patch(self, **kwargs): + return patch("user_engine.adapters.application_access.urlopen", **kwargs) + + def test_read_sends_check_subject_shape_and_maps_results(self): + body = _Response(json.dumps({ + "registry_snapshot_digest": "sha256:" + "0" * 64, + "results": [ + {"system": "informed-decision", "effect": "allow", "decision_id": "decision:1", + "lifetime": {"kind": "ttl", "expires_at": "2026-10-10T12:00:30Z"}}, + {"system": "markitect", "effect": "not_applicable", "reason": "no rule"}, + ], + }).encode()) + with self._patch(return_value=body) as call: + results = self.adapter.read(_actor(), ["informed-decision", "markitect"]) + request = call.call_args.args[0] + self.assertEqual(request.full_url, "http://readback/v1/readback/application_access") + self.assertEqual(request.get_header("Authorization"), "Bearer readback-token") + payload = json.loads(request.data) + self.assertEqual(set(payload), {"tenant", "subject", "systems"}) + self.assertEqual(payload["subject"]["tenant"], "tenant-a") + check = _check_payload(_actor()) + self.assertEqual(payload["subject"], check["subject"]) + self.assertEqual(results[0].effect, "allow") + self.assertEqual(results[0].expires_at, NOW + timedelta(seconds=30)) + self.assertEqual(results[1].effect, "not_applicable") + + def test_mismatched_or_unknown_results_are_unavailable(self): + bodies = [ + {"registry_snapshot_digest": "x", "results": [{"system": "other", "effect": "allow"}]}, + {"registry_snapshot_digest": "x", "results": [{"system": "a", "effect": "permit"}]}, + {"registry_snapshot_digest": "x", "results": []}, + ] + for body in bodies: + with self.subTest(body=body), self._patch(return_value=_Response(json.dumps(body).encode())): + with self.assertRaises(ReadbackUnavailable): + self.adapter.read(_actor(), ["a"]) + + def test_transport_failures_are_unavailable(self): + for error in (URLError("down"), HTTPError("http://readback", 503, "x", {}, None), TimeoutError()): + with self.subTest(error=error), self._patch(side_effect=error): + with self.assertRaises(ReadbackUnavailable): + self.adapter.systems() + + def test_empty_token_never_calls(self): + with open(self.token_file, "w", encoding="utf-8"): + pass + with self._patch() as call, self.assertRaises(ReadbackUnavailable): + self.adapter.systems() + call.assert_not_called() + + def test_systems_maps_client_binding(self): + body = _Response(json.dumps({ + "registry_snapshot_digest": "sha256:" + "0" * 64, + "systems": [ + {"id": "informed-decision", "name": "Informed Decision", "readback": True, + "client_binding": {"issuer": ISSUER, "client_id": "informed-client"}}, + {"id": "markitect", "readback": False}, + ], + }).encode()) + with self._patch(return_value=body): + systems = self.adapter.systems() + self.assertEqual(systems[0].client_id, "informed-client") + self.assertIsNone(systems[1].client_id) + + def test_request_size_is_bounded(self): + with self.assertRaises(ValueError): + self.adapter.read(_actor(), [f"s{n}" for n in range(65)]) + + +def _check_payload(actor): + body = _Response(json.dumps({"id": "decision:1", "effect": "deny"}).encode()) + with patch("user_engine.adapters.flex_auth.urlopen", return_value=body) as call: + FlexAuthHTTPAdapter(base_url="http://flex-auth").check(AuthorizationRequest( + actor=actor, resource_type="user-engine:user", resource_id="u", + action="user.read", tenant=actor.tenant, correlation_id="c", + )) + return json.loads(call.call_args.args[0].data) + + +if __name__ == "__main__": + unittest.main() + + +class PortalApplicationAccessTests(unittest.TestCase): + def setUp(self): + import test_portal_navigation + + test_portal_navigation.PortalNavigationTests.setUp(self) + store = self.app.service.store + for app_id, system_id, client_id in [ + ("informed", "informed-decision", "informed-client"), + ("markitect", "markitect", "markitect-client"), + ]: + application, binding = _bound(app_id, system_id, client_id) + store.save_application(application) + store.save_binding(binding) + self.readback = _FakeReadback( + [ + ReadbackResult("informed-decision", "allow", + datetime.now(UTC) + timedelta(seconds=30), "decision:1"), + ReadbackResult("markitect", "not_applicable"), + ], + systems=[_system("informed-decision", "informed-client"), + _system("markitect", "markitect-client")], + ) + self.app.application_access = ApplicationAccessReader( + self.readback, issuer=ISSUER, + bound_applications=lambda: [ + (store.application(binding.application_id), binding) + for binding in store.all_bindings() + ], + ) + + def _onboarding(self): + from test_web import invoke + + return invoke(self.app, "/onboarding", cookie="ue_session=member")[1] + + def test_account_page_renders_readback_rows(self): + body = self._onboarding() + self.assertIn(b'
  • Informed - Available
    • ', body) + self.assertIn(b'
  • Markitect - Not available: not offered here
    • ', body) + self.assertIn(b"is not a complete inventory", body) + + def test_unreadable_registry_says_could_not_be_checked(self): + self.readback.fail_systems = True + self.assertIn(b"Application access could not be checked.", self._onboarding()) + + def test_logout_drops_the_cached_rows(self): + from test_web import invoke + + self._onboarding() + self._onboarding() + self.assertEqual(len(self.readback.reads), 1) + invoke(self.app, "/logout", method="POST", cookie="ue_session=member", form={}) + self._onboarding() + self.assertEqual(len(self.readback.reads), 2) + + def test_section_absent_without_readback(self): + self.app.application_access = None + self.assertNotIn(b"application-access-list", self._onboarding()) diff --git a/workplans/USER-WP-0026-account-recovery.md b/workplans/USER-WP-0026-account-recovery.md index b4cf10e..12405e3 100644 --- a/workplans/USER-WP-0026-account-recovery.md +++ b/workplans/USER-WP-0026-account-recovery.md @@ -101,6 +101,18 @@ deny, not_applicable, failure and expired as not available, never as allow. Use a separate `USER_ENGINE_READBACK_TOKEN_FILE`. Wait stays on access-engine T02/T03 code and its T04 founder go. +2026-10-10: access-engine folded amendments (a)-(c) into the contract, +accepted FLEX-DEC-2026-019 and shipped T02/T03 (8f2ba75). T05 (b5da745) gives +informed-decision a separate `application.access` rule. user-engine's client is +now in the repo, off by default: `application_access.py`, +`adapters/application_access.py`, the account-page section, and +`USER_ENGINE_READBACK_URL` / `USER_ENGINE_READBACK_TOKEN_FILE` (see +docs/configuration.md). A workload is listed only when all three records are +bound. Rows follow the agreed rendering table, and none is inferred as allow. +Remaining before this task is done: access-engine T04 (readback pin values, +digest-equality test, founder go), operator-installed readback URL and token +file, and a live check against the pin. + ### Live recovery rollout — 2026-09-12 Recovery is deployed in KeyCape 4d8b8fe, User Engine e54b6ee and Vergabe c067993