feat: add workload-authenticated account-authority HTTP lookup (USER-WP-0037-T03 partial)

Exposes POST /api/v1/platform/account-authority/lookup over the existing
lookup_account_authority service method, gated by a dedicated Hub workload
bearer distinct from end-user OIDC claims and the trusted proxy secret. The
route is absent unless USER_ENGINE_HUB_AUTHORITY_TOKEN is configured.

T02 (root-grant custody/mapping) and the remainder of T03 (real workload
credential admission) depend on decisions outside this repo, per
credential-routing.md and NetKingdom's root-custody canon. Requested via
State Hub messages to net-kingdom and state-hub; workplan set to blocked.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 227621@bnt-lap001
Assistant-Session: ffcc35e9-8ffa-4881-a473-02ca94d6b366

src/user_engine/runtime.py

@@ -14,6 +14,7 @@ from user_engine.adapters import (
     HTTPIdentityProvisioningAdapter,
     HTTPTenantManagementAdapter,
 )
+from user_engine.domain import Actor, PrincipalType
 from user_engine.service import UserEngineService
 from user_engine.identity_state import AutheliaIdentityState
 from user_engine.oidc import OIDCClient
@@ -127,6 +128,19 @@ def create_application() -> PortalApplication:
             if os.environ.get("USER_ENGINE_IDENTITY_STATE_URL")
             else None
         ),
+        hub_authority_caller=(
+            Actor(
+                issuer=_required("USER_ENGINE_HUB_AUTHORITY_ISSUER"),
+                subject=_required("USER_ENGINE_HUB_AUTHORITY_SUBJECT"),
+                tenant=os.environ.get("USER_ENGINE_HUB_AUTHORITY_TENANT", "").strip()
+                or "platform:root",
+                principal_type=PrincipalType.SERVICE,
+                audience=(),
+            )
+            if os.environ.get("USER_ENGINE_HUB_AUTHORITY_TOKEN")
+            else None
+        ),
+        hub_authority_token=os.environ.get("USER_ENGINE_HUB_AUTHORITY_TOKEN", "").strip(),
     )
 
     from user_engine.operations_status import check_services

src/user_engine/web.py

@@ -92,6 +92,8 @@ class PortalApplication:
         registration_rate_limit: int = 10,
         registration_rate_window_seconds: int = 60,
         identity_lookup: Any | None = None,
+        hub_authority_caller: Actor | None = None,
+        hub_authority_token: str = "",
     ) -> None:
         if len(trusted_proxy_secret) < 24:
             raise ValueError("trusted proxy secret must contain at least 24 characters")
@@ -128,6 +130,12 @@ class PortalApplication:
         self._registration_attempts: dict[str, deque[float]] = {}
         self._registration_attempts_lock = Lock()
         self.identity_lookup = identity_lookup
+        if hub_authority_token and len(hub_authority_token) < 24:
+            raise ValueError("hub authority token must contain at least 24 characters")
+        if bool(hub_authority_caller) != bool(hub_authority_token):
+            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
 
     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)}"
@@ -397,6 +405,9 @@ class PortalApplication:
                 environ, start_response, correlation_id
             )
 
+        if path == "/api/v1/platform/account-authority/lookup" and method == "POST":
+            return self._account_authority_lookup(environ, start_response, correlation_id)
+
         actor = self._actor(environ)
         self._set_account_navigation(environ, actor)
         privileged = path == "/platform" or path.startswith(("/platform/", "/admin/", "/api/v1/platform/", "/api/v1/tenants/"))
@@ -1401,6 +1412,31 @@ class PortalApplication:
     def _actor(self, environ: Mapping[str, Any]) -> Any:
         return self.service.identity_adapter.normalize(self._claims(environ))
 
+    def _account_authority_lookup(
+        self, environ: Mapping[str, Any], start_response: StartResponse, correlation_id: str,
+    ) -> Iterable[bytes]:
+        """Independently authenticated workload read; never end-user claims.
+
+        The caller identity is the fixed configured Hub workload, not
+        anything forwarded from a human bearer or verified OIDC claims.
+        """
+        if self.hub_authority_caller is None or not self.hub_authority_token:
+            raise NotFoundError("account authority lookup is unavailable")
+        header = str(environ.get("HTTP_AUTHORIZATION", ""))
+        prefix = "Bearer "
+        supplied = header[len(prefix):] if header.startswith(prefix) else ""
+        if not supplied or not secrets.compare_digest(supplied, self.hub_authority_token):
+            raise AuthorizationDenied("workload authentication required")
+        body = self._body(environ)
+        result = self.service.lookup_account_authority(
+            self.hub_authority_caller,
+            issuer=str(body.get("issuer") or ""),
+            subject=str(body.get("subject") or ""),
+            tenant=str(body.get("tenant") or ""),
+            correlation_id=correlation_id,
+        )
+        return self._json(start_response, "200 OK", _jsonable(result), correlation_id)
+
     def _start_public_registration(
         self,
         environ: Mapping[str, Any],

tests/test_web_account_authority.py

@@ -0,0 +1,159 @@
+import io
+import json
+import unittest
+
+from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort
+from user_engine.domain import Actor, AuthorizationEffect, PrincipalType
+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"
+HUB_TOKEN = "hub-workload-token-with-adequate-length"
+HUB_CALLER = Actor(
+    issuer="https://hub.example", subject="hub-workload",
+    tenant="platform:root", principal_type=PrincipalType.SERVICE, audience=(),
+)
+
+
+def invoke(app, path, *, method="POST", body=None, authorization=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",
+        "REMOTE_ADDR": "127.0.0.1",
+    }
+    if authorization is not None:
+        environ["HTTP_AUTHORIZATION"] = authorization
+    captured = {}
+
+    def start_response(status, headers):
+        captured["status"] = status
+        captured["headers"] = dict(headers)
+
+    response = b"".join(app(environ, start_response))
+    return captured, response
+
+
+class AccountAuthorityRouteTests(unittest.TestCase):
+    def setUp(self):
+        self.store = InMemoryUserEngineStore()
+        self.store.migrate()
+        self.policy = LocalAuthorizationCheckPort()
+        self.service = UserEngineService(
+            store=self.store,
+            identity_adapter=FixtureIdentityClaimsAdapter(),
+            authorization=self.policy,
+        )
+        self.session = self.service.me(human_actor_claims())
+        self.app = PortalApplication(
+            self.service,
+            trusted_proxy_secret=SECRET,
+            login_url="https://kc.example/login",
+            hub_authority_caller=HUB_CALLER,
+            hub_authority_token=HUB_TOKEN,
+        )
+        self.body = {
+            "issuer": self.session.actor.issuer,
+            "subject": self.session.actor.subject,
+            "tenant": self.session.actor.tenant,
+        }
+
+    def test_route_disabled_without_configuration(self):
+        app = PortalApplication(
+            self.service, trusted_proxy_secret=SECRET, login_url="https://kc.example/login",
+        )
+        result, _ = invoke(
+            app, "/api/v1/platform/account-authority/lookup",
+            body=self.body, authorization=f"Bearer {HUB_TOKEN}",
+        )
+        self.assertEqual("404 Not Found", result["status"])
+
+    def test_missing_or_wrong_bearer_is_denied(self):
+        missing, _ = invoke(self.app, "/api/v1/platform/account-authority/lookup", body=self.body)
+        self.assertEqual("403 Forbidden", missing["status"])
+        wrong, _ = invoke(
+            self.app, "/api/v1/platform/account-authority/lookup",
+            body=self.body, authorization="Bearer not-the-token",
+        )
+        self.assertEqual("403 Forbidden", wrong["status"])
+
+    def test_denial_precedes_any_domain_lookup(self):
+        self.policy.default_effect = AuthorizationEffect.DENY
+        before_users = dict(self.store.users)
+        result, _ = invoke(
+            self.app, "/api/v1/platform/account-authority/lookup",
+            body=self.body, authorization=f"Bearer {HUB_TOKEN}",
+        )
+        self.assertEqual("403 Forbidden", result["status"])
+        self.assertEqual(self.store.users, before_users)
+
+    def test_valid_workload_bearer_returns_current_state_without_provisioning(self):
+        result, payload = invoke(
+            self.app, "/api/v1/platform/account-authority/lookup",
+            body=self.body, authorization=f"Bearer {HUB_TOKEN}",
+        )
+        self.assertEqual("200 OK", result["status"])
+        data = json.loads(payload)
+        self.assertEqual(data["identity_id"], self.session.identities[0].identity_id)
+        self.assertEqual(data["account_status"], "active")
+        self.assertNotIn("roles", data)
+        self.assertNotIn("primary_email", data)
+        request = self.policy.requests[-1]
+        self.assertEqual(request.actor.subject, "hub-workload")
+        self.assertNotEqual(request.actor.subject, self.session.actor.subject)
+
+    def test_unknown_identity_is_not_found_and_does_not_provision(self):
+        before = repr(self.store.__dict__)
+        body = dict(self.body, subject="unknown-subject")
+        result, _ = invoke(
+            self.app, "/api/v1/platform/account-authority/lookup",
+            body=body, authorization=f"Bearer {HUB_TOKEN}",
+        )
+        self.assertEqual("404 Not Found", result["status"])
+        self.assertEqual(repr(self.store.__dict__), before)
+
+    def test_end_user_claims_headers_do_not_satisfy_workload_auth(self):
+        environ_extra = {
+            "HTTP_X_VERIFIED_OIDC_CLAIMS": json.dumps(human_actor_claims()),
+            "HTTP_X_USER_ENGINE_PROXY_SECRET": SECRET,
+        }
+
+        def call():
+            payload = json.dumps(self.body).encode()
+            environ = {
+                "REQUEST_METHOD": "POST",
+                "PATH_INFO": "/api/v1/platform/account-authority/lookup",
+                "QUERY_STRING": "",
+                "CONTENT_LENGTH": str(len(payload)),
+                "wsgi.input": io.BytesIO(payload),
+                "HTTP_X_REQUEST_ID": "corr_test",
+                "REMOTE_ADDR": "127.0.0.1",
+                **environ_extra,
+            }
+            captured = {}
+
+            def start_response(status, headers):
+                captured["status"] = status
+
+            response = b"".join(self.app(environ, start_response))
+            return captured, response
+
+        result, _ = call()
+        self.assertEqual("403 Forbidden", result["status"])
+
+    def test_construction_requires_both_caller_and_token(self):
+        with self.assertRaises(ValueError):
+            PortalApplication(
+                self.service, trusted_proxy_secret=SECRET,
+                login_url="https://kc.example/login", hub_authority_caller=HUB_CALLER,
+            )
+        with self.assertRaises(ValueError):
+            PortalApplication(
+                self.service, trusted_proxy_secret=SECRET,
+                login_url="https://kc.example/login", hub_authority_token=HUB_TOKEN,
+            )

workplans/USER-WP-0037-hub-authority-lookup.md

@@ -4,7 +4,7 @@ type: workplan
 title: "Read-only account authority for Hub integration"
 domain: communication
 repo: user-engine
-status: active
+status: blocked
 owner: codex
 topic_slug: communication
 created: "2026-09-28"
@@ -48,7 +48,7 @@ production deployment was added. T02/T03 retain those outstanding owner gates.
 
 ```task
 id: USER-WP-0037-T02
-status: todo
+status: wait
 priority: high
 state_hub_task_id: "23a5c211-277b-561e-a4b3-0f8e99537cac"
 ```
@@ -58,11 +58,19 @@ platform:root to tenant:platform mapping. Do not derive a grant from JWT roles,
 a username, or Hub's configured root subject. Define retainable evidence for
 account and entitlement observations.
 
+Blocked 2026-09-28: this is a NetKingdom canon decision, not a user-engine
+implementation choice. `net-kingdom/docs/platform-root-custody.md` covers
+OpenBao/king-credential bootstrap custody but does not yet define how a
+workload-facing root-grant fact is derived, represented, or revoked for a
+consumer like the Hub. Requested via State Hub message
+`379d266b-b738-4772-82c0-318e67a50bc8` to `net-kingdom`. `AccountAuthority`
+deliberately still carries no root-grant field pending that decision.
+
 ## Authenticated workload transport and private acceptance
 
 ```task
 id: USER-WP-0037-T03
-status: todo
+status: wait
 priority: high
 state_hub_task_id: "c5d67725-c685-50f6-9db8-28114a34014b"
 ```
@@ -72,3 +80,21 @@ workload, policy action account.authority.read, exact target binding and no
 end-user bearer forwarding. Add HTTP schema and denial/no-provisioning tests,
 then admit real caller credentials through their owner and verify withdrawal,
 outages and private Hub integration. A service method alone is not this gate.
+
+Implemented 2026-09-28: `POST /api/v1/platform/account-authority/lookup`
+(`src/user_engine/web.py`), gated by a dedicated `hub_authority_caller` /
+`hub_authority_token` pair distinct from `trusted_proxy_secret` and end-user
+OIDC claims — a bearer mismatch or unconfigured route denies/404s before any
+domain lookup runs. Wired in `runtime.py` behind
+`USER_ENGINE_HUB_AUTHORITY_TOKEN` (+ `_ISSUER`/`_SUBJECT`/`_TENANT`), unset by
+default so the route stays absent until provisioned. Schema, denial, and
+no-provisioning tests: `tests/test_web_account_authority.py` (7 tests, all
+green against `make test`, 277 tests total).
+
+Blocked 2026-09-28 on the remaining half of this gate: admitting a *real*
+Hub workload credential is a credential-routing matter
+(`.claude/rules/credential-routing.md`) — OpenBao/ops-warden issue it, this
+repo only consumes it. Requested via State Hub message
+`f3bcc516-7241-4050-ba52-5723bb174936` to `state-hub`. Withdrawal, outage,
+and private-integration verification with the real credential are still
+owed once it exists.