fix(authority): authenticate revision checks as the installed module

README.md

@@ -319,6 +319,18 @@ consumption still requires a durable replay store the caller owns.
 
 ## Current authorization revision
 
+For a host that requires the calling module's identity, use
+`ModuleMeshGRPCClient(address, module_client)`. Construct `module_client` with
+`saas_sdk.module_authority.ModuleClient(gateway_origin, prefix, internal_credential,
+workload_credential, mesh_protected=...)`; both credentials are async callbacks
+that read this workload's own current credentials. Install the `module-authority`
+and `revision-grpc` extras. The revision client owns and closes the module client.
+Each check mints a module Work Context for the verified caller's organization and
+sends it in `x-codefly-work-context`; the request body retains the caller's owner
+and actor scopes. No credential or authorization decision is cached, and a failed
+mint prevents the revision call. The older clients below remain for hosts that
+explicitly accept the shared internal credential.
+
 `saas_sdk.authorization_revision` implements Accounts' current revision protocol
 for **already verified** `WorkContext` claims. It projects the owner and every
 actor's original scopes, including resource IDs and revision. The consumer keeps

pyproject.toml

@@ -17,6 +17,7 @@ dependencies = ["protobuf>=5.29.3", "cryptography>=41", "googleapis-common-proto
 test = ["pytest>=8", "pytest-asyncio>=0.24"]
 revision-connect = ["httpx>=0.27,<1"]
 revision-grpc = ["grpcio>=1.66"]
+module-authority = ["httpx>=0.27,<1"]
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/saas_sdk/authorization_revision.py

@@ -39,6 +39,8 @@
     "revision_request",
     "ConnectClient",
     "GRPCClient",
+    "MeshGRPCClient",
+    "ModuleMeshGRPCClient",
 ]
 
 try:
@@ -299,13 +301,13 @@ async def _check(self, claims):
         try:
             # One total budget includes credential refresh, connection and RPC body.
             deadline = asyncio.get_running_loop().time() + TIMEOUT
-            token = await _internal_credential(self.credential)
+            metadata = await self._metadata(claims)
             remaining = deadline - asyncio.get_running_loop().time()
             if remaining <= 0:
                 raise TimeoutError
             response = await self.rpc(
                 revision_request(claims),
-                metadata=(("x-codefly-internal-token", token),),
+                metadata=metadata,
                 timeout=remaining,
                 wait_for_ready=False,
             )
@@ -326,6 +328,9 @@ async def _check(self, claims):
     async def aclose(self) -> None:
         await self.channel.close()
 
+    async def _metadata(self, claims):
+        return (("x-codefly-internal-token", await _internal_credential(self.credential)),)
+
 
 class GRPCClient(_GRPCRevision):
     """Server-TLS gRPC on the canonical internal RPC, for an explicit HTTPS origin."""
@@ -380,3 +385,27 @@ def __init__(self, address, credential):
             credential,
         )
 
+
+class ModuleMeshGRPCClient(MeshGRPCClient):
+    """Revision checks authenticated as the installed module in the work organization.
+
+    Owns the supplied module client. The revision subjects remain the verified
+    caller's; the header identifies the service asking about those subjects.
+    """
+
+    def __init__(self, address, module_client):
+        super().__init__(address, None)
+        self.module_client = module_client
+
+    async def _metadata(self, claims):
+        async def credential():
+            return await self.module_client.work_context(claims.tenant_id)
+
+        return (("x-codefly-work-context", await _internal_credential(credential)),)
+
+    async def aclose(self):
+        try:
+            await super().aclose()
+        finally:
+            await self.module_client.aclose()
+

src/saas_sdk/module_authority.py

@@ -0,0 +1,133 @@
+"""The host module broker's organization-scoped workload credential exchange.
+
+Credential callbacks read the composing runtime's own credentials. No user
+credential, authorization decision, retry or cross-organization cache lives here.
+"""
+
+import asyncio
+import ipaddress
+import json
+from urllib.parse import urlsplit
+from uuid import UUID
+
+import httpx
+
+from .authorization_revision import RevisionUnavailable, _decode, _internal_credential
+
+
+def _origin(value, mesh_protected):
+    try:
+        url = urlsplit(value)
+        host = url.hostname or ""
+        port = url.port
+        try:
+            loopback = ipaddress.ip_address(host).is_loopback
+        except ValueError:
+            loopback = host == "localhost"
+        mesh = mesh_protected and host.endswith((".svc", ".svc.cluster.local"))
+        if (
+            not host
+            or url.username is not None
+            or url.password is not None
+            or url.path not in ("", "/")
+            or url.query
+            or url.fragment
+            or port == 0
+            or "\\" in value
+            or any(c.isspace() for c in value)
+            or url.scheme not in ("http", "https")
+            or (url.scheme == "http" and not (loopback or mesh))
+        ):
+            raise ValueError
+    except ValueError:
+        raise ValueError(
+            "Module broker requires HTTPS or an explicitly protected mesh origin"
+        ) from None
+    return value.rstrip("/")
+
+
+class ModuleClient:
+    """Mint this workload's identity for one organization, without caching it."""
+
+    def __init__(
+        self,
+        origin,
+        prefix,
+        internal_credential,
+        workload_credential,
+        *,
+        mesh_protected=False,
+        transport=None,
+    ):
+        self.endpoint = _origin(origin, mesh_protected) + "/modules/_work-context"
+        if not prefix or len(prefix) > 128 or any(c.isspace() for c in prefix):
+            raise ValueError("A module prefix is required")
+        self.prefix = prefix
+        self.internal_credential = internal_credential
+        self.workload_credential = workload_credential
+        self.client = httpx.AsyncClient(
+            transport=transport, trust_env=False, follow_redirects=False
+        )
+
+    async def work_context(self, organization):
+        try:
+            return await asyncio.wait_for(self._mint(organization), timeout=3)
+        except (
+            asyncio.TimeoutError,
+            httpx.HTTPError,
+            OSError,
+            ValueError,
+            TypeError,
+            AttributeError,
+            TimeoutError,
+            RevisionUnavailable,
+        ):
+            # Broker bodies, transport details and credentials never escape.
+            raise RevisionUnavailable("Module identity is unavailable") from None
+
+    async def _mint(self, organization):
+        if str(UUID(organization)) != organization:
+            raise ValueError("Organization must be a canonical UUID")
+        internal = await _internal_credential(self.internal_credential)
+        workload = await _internal_credential(self.workload_credential)
+        request = httpx.Request(
+            "POST",
+            self.endpoint,
+            headers={
+                "content-type": "application/json",
+                "accept": "application/json",
+                "accept-encoding": "identity",
+                "x-codefly-internal-token": internal,
+                "authorization": "Bearer " + workload,
+            },
+            content=json.dumps(
+                {"prefix": self.prefix, "organization_id": organization}
+            ).encode(),
+        )
+        response = await self.client.send(request, stream=True, auth=None)
+        try:
+            if (
+                response.status_code != 200
+                or response.headers.get("content-type", "").split(";", 1)[0]
+                != "application/json"
+                or response.headers.get("content-encoding", "identity") != "identity"
+            ):
+                raise ValueError("Module broker refused the exchange")
+            data = bytearray()
+            async for chunk in response.aiter_bytes():
+                data.extend(chunk)
+                if len(data) > 65536:
+                    raise ValueError("Module broker response exceeds bound")
+            payload = _decode(data)
+            if payload.get("tenant") != organization or not payload.get("principalId"):
+                raise ValueError("Module broker returned a different identity")
+
+            async def minted():
+                return payload.get("token")
+
+            return await _internal_credential(minted)
+        finally:
+            await response.aclose()
+
+    async def aclose(self):
+        await self.client.aclose()

tests/test_authorization_revision.py

@@ -313,6 +313,7 @@ async def check(request, context):
         # The current host's transport: h2c, with the mesh securing the hop.
         port = server.add_insecure_port("localhost:0")
     await server.start()
+    state["address"] = f"localhost:{port}"
     if request.param == "tls":
         tls = ssl.create_default_context(cadata=cert_bytes.decode())
         client = revision.GRPCClient(f"https://localhost:{port}", token, tls_context=tls)
@@ -348,6 +349,39 @@ async def rotated():
         )
 
 
+@pytest.mark.parametrize("grpc_endpoint", ["mesh"], indirect=True)
+async def test_module_revision_uses_work_organization_and_separate_module_proof(grpc_endpoint):
+    _, state = grpc_endpoint
+
+    class Module:
+        def __init__(self):
+            self.organizations = []
+            self.closed = False
+
+        async def work_context(self, org):
+            self.organizations.append(org)
+            return "own-module-proof"
+
+        async def aclose(self):
+            self.closed = True
+
+    module = Module()
+    client = revision.ModuleMeshGRPCClient(state["address"], module)
+    try:
+        await client.check(claims())
+        state["status"] = grpc.StatusCode.FAILED_PRECONDITION
+        with pytest.raises(revision.RevisionDenied):
+            await client.check(claims())
+        assert module.organizations == [claims().tenant_id] * 2
+        for request, metadata in state["calls"]:
+            assert request == revision.revision_request(claims())
+            assert metadata["x-codefly-work-context"] == "own-module-proof"
+            assert not {"authorization", "cookie", "x-codefly-internal-token"}.intersection(metadata)
+    finally:
+        await client.aclose()
+    assert module.closed
+
+
 @pytest.mark.parametrize(
     "status,error",
     [

tests/test_module_authority.py

@@ -0,0 +1,125 @@
+import json
+
+import httpx
+import pytest
+
+from saas_sdk.authorization_revision import RevisionUnavailable
+from saas_sdk.module_authority import ModuleClient
+
+pytestmark = pytest.mark.asyncio
+ORG = "11111111-1111-4111-8111-111111111111"
+OTHER = "22222222-2222-4222-8222-222222222222"
+
+
+async def internal():
+    return "internal-fixture"
+
+
+async def workload():
+    return "workload-fixture"
+
+
+def client(respond):
+    return ModuleClient(
+        "https://broker.invalid",
+        "example",
+        internal,
+        workload,
+        transport=httpx.MockTransport(respond),
+    )
+
+
+async def test_organization_and_rotating_own_credential_on_every_exchange():
+    seen = []
+
+    async def respond(request):
+        body = json.loads(request.content)
+        seen.append((body, dict(request.headers)))
+        assert request.url.path == "/modules/_work-context"
+        return httpx.Response(
+            200,
+            json={
+                "token": "module-" + body["organization_id"],
+                "tenant": body["organization_id"],
+                "principalId": "example",
+            },
+        )
+
+    c = client(respond)
+    try:
+        assert await c.work_context(ORG) == "module-" + ORG
+
+        async def rotated():
+            return "rotated-fixture"
+
+        c.workload_credential = rotated
+        assert await c.work_context(OTHER) == "module-" + OTHER
+        assert [b for b, _ in seen] == [
+            {"prefix": "example", "organization_id": ORG},
+            {"prefix": "example", "organization_id": OTHER},
+        ]
+        assert [h["authorization"] for _, h in seen] == [
+            "Bearer workload-fixture",
+            "Bearer rotated-fixture",
+        ]
+        assert all(h["x-codefly-internal-token"] == "internal-fixture" for _, h in seen)
+        assert all(
+            "x-codefly-work-context" not in h and "cookie" not in h for _, h in seen
+        )
+    finally:
+        await c.aclose()
+
+
+@pytest.mark.parametrize(
+    "status,payload",
+    [
+        (401, {"error": "private-secret"}),
+        (403, {}),
+        (302, {}),
+        (503, {}),
+        (200, {"token": "private-secret", "tenant": OTHER, "principalId": "example"}),
+        (200, {"token": "bad\ncredential", "tenant": ORG, "principalId": "example"}),
+        (200, {"token": "x" * 9000, "tenant": ORG, "principalId": "example"}),
+    ],
+)
+async def test_refusal_is_redacted_and_not_retried(status, payload):
+    calls = []
+
+    async def respond(request):
+        calls.append(request)
+        return httpx.Response(status, json=payload)
+
+    c = client(respond)
+    try:
+        with pytest.raises(
+            RevisionUnavailable, match="^Module identity is unavailable$"
+        ):
+            await c.work_context(ORG)
+        assert len(calls) == 1
+    finally:
+        await c.aclose()
+
+
+async def test_invalid_organization_never_sends_credentials():
+    def respond(request):
+        pytest.fail("invalid selector reached broker")
+
+    c = client(respond)
+    try:
+        with pytest.raises(RevisionUnavailable):
+            await c.work_context("not-an-organization")
+    finally:
+        await c.aclose()
+
+
+@pytest.mark.parametrize(
+    "origin",
+    [
+        "http://public.invalid",
+        "https://user:secret@broker.invalid",
+        "https://broker.invalid/other",
+    ],
+)
+async def test_refuse_unsafe_broker(origin):
+    with pytest.raises(ValueError, match="Module broker requires"):
+        ModuleClient(origin, "example", internal, workload, mesh_protected=True)