feat: reach the host's internal revision RPC over the mesh (h2c)

README.md

@@ -114,7 +114,9 @@ Verification enforces the two-segment shape, the signature, structural
 validation (including the proto's `min_len` constraints on optional fields and
 monotonic scope attenuation), the time window, and the caller's expectations,
 raising `WorkContextError` (or `WorkContextDenied` from `require_scope`) on any
-failure. Replay policy is reported on the claims; enforcing single-use
+failure. When the key set itself cannot be fetched or read, `JWKSVerifier` raises
+`WorkContextUnavailable`, a `WorkContextError` subclass, so a consumer can report
+an authority outage as unavailable rather than as an invalid capability. Replay policy is reported on the claims; enforcing single-use
 consumption still requires a durable replay store the caller owns.
 
 ## Current authorization revision
@@ -125,16 +127,25 @@ actor's original scopes, including resource IDs and revision. The consumer keeps
 signature/lifetime verification, action/resource policy and public error mapping.
 
 ```python
-from saas_sdk.authorization_revision import GRPCClient, RevisionDenied, RevisionUnavailable
+from saas_sdk.authorization_revision import MeshGRPCClient, RevisionDenied, RevisionUnavailable
 
-client = GRPCClient("https://accounts.internal:50051", current_internal_credential)
+# The host:port Codefly resolved for accounts' private `rest` endpoint.
+client = MeshGRPCClient(resolved_accounts_rest_address, current_internal_credential)
 try:
     await client.check(verified_claims)
 finally:
     await client.aclose()
 ```
 
-Install `saas-sdk-python[revision-grpc]` for the canonical TLS gRPC client. Where an
+A current host serves this RPC as cleartext HTTP/2 (h2c) multiplexed onto accounts'
+private `rest` endpoint, and leaves in-cluster transport security to the service
+mesh (module-saas-starter `module/INTERNAL_TRANSPORT.md`). `MeshGRPCClient` dials
+that resolved `host:port` and adds no TLS; it refuses a URL, a path or credentials
+in the address. Use it only where the mesh enforces mutual TLS on the hop.
+`GRPCClient("https://…", …)` remains for a deployment that terminates server TLS on
+accounts itself.
+
+Install `saas-sdk-python[revision-grpc]` for either gRPC client. Where an
 endpoint explicitly serves Connect, install `[revision-connect]` and select
 `ConnectClient`; selecting that compatibility transport is a composition choice,
 never an automatic fallback. `revision_request(claims)` needs neither networking

src/saas_sdk/authorization_revision.py

@@ -3,6 +3,14 @@
 The caller retains signature, lifetime, scope and resource policy. Every check
 refreshes its internal credential and contacts the authority; no decision is cached.
 Networking dependencies are optional extras so projection remains lightweight.
+
+Two gRPC transports reach the same RPC. ``MeshGRPCClient`` is the one a current
+host serves: the host's internal tier is cleartext HTTP/2 (h2c) multiplexed onto
+accounts' private ``rest`` endpoint, and in-cluster transport security belongs to
+the service mesh (module-saas-starter ``module/INTERNAL_TRANSPORT.md``). The
+caller dials the address Codefly resolved for that endpoint; the process adds no
+TLS because the mesh has already authenticated both ends. ``GRPCClient`` keeps
+the explicit server-TLS origin for a deployment that terminates TLS on accounts.
 """
 
 import asyncio
@@ -216,39 +224,46 @@ async def aclose(self) -> None:
         await self.client.aclose()
 
 
-class GRPCClient:
-    """TLS gRPC on the canonical internal RPC; no retries, fallback or authority cache."""
+_CHANNEL_OPTIONS = (
+    ("grpc.enable_retries", 0),
+    ("grpc.enable_http_proxy", 0),
+    ("grpc.max_receive_message_length", 8192),
+    ("grpc.max_send_message_length", 65536),
+)
 
-    def __init__(self, origin, credential, *, tls_context=None):
-        if grpc is None:
-            raise ImportError(
-                "Install saas-sdk-python[revision-grpc] for gRPC revision checks"
-            )
-        url = urlsplit(_https_origin(origin))
+
+def _mesh_address(value: str) -> str:
+    """A resolved ``host:port`` with nothing else: no scheme, path or credentials.
+
+    The address is what Codefly resolves for accounts' private ``rest`` endpoint,
+    so anything beyond a host and a nonzero port is a configuration error, not a
+    form to interpret.
+    """
+    if (
+        not isinstance(value, str)
+        or "://" in value
+        or any(c in value for c in "/?#@\\")
+        or any(c.isspace() or ord(c) < 32 for c in value)
+    ):
+        raise ValueError("Accounts requires a resolved host:port internal address")
+    url = urlsplit("//" + value)
+    try:
+        port = url.port
+    except ValueError:
+        port = None
+    if not url.hostname or not port:
+        raise ValueError("Accounts requires a resolved host:port internal address")
+    host = f"[{url.hostname}]" if ":" in url.hostname else url.hostname
+    return f"{host}:{port}"
+
+
+class _GRPCRevision:
+    """The canonical internal RPC over an already opened channel; no retries,
+    fallback or authority cache. Subclasses choose only the transport."""
+
+    def _bind(self, channel, credential):
         self.credential = credential
-        context = (
-            tls_context if tls_context is not None else ssl.create_default_context()
-        )
-        if not context.check_hostname or context.verify_mode != ssl.CERT_REQUIRED:
-            raise ValueError("Accounts TLS verification cannot be disabled")
-        # Use the deployment-selected trust roots, not gRPC environment overrides.
-        roots = b"".join(
-            ssl.DER_cert_to_PEM_cert(cert).encode("ascii")
-            for cert in context.get_ca_certs(binary_form=True)
-        )
-        if not roots:
-            raise ValueError("Accounts requires trusted TLS roots")
-        host = f"[{url.hostname}]" if ":" in url.hostname else url.hostname
-        self.channel = grpc.aio.secure_channel(
-            f"{host}:{url.port or 443}",
-            grpc.ssl_channel_credentials(root_certificates=roots),
-            options=(
-                ("grpc.enable_retries", 0),
-                ("grpc.enable_http_proxy", 0),
-                ("grpc.max_receive_message_length", 8192),
-                ("grpc.max_send_message_length", 65536),
-            ),
-        )
+        self.channel = channel
         self.rpc = self.channel.unary_unary(
             REVISION_PATH,
             request_serializer=pb.CheckAuthorizationRevisionRequest.SerializeToString,
@@ -293,3 +308,58 @@ async def _check(self, claims):
 
     async def aclose(self) -> None:
         await self.channel.close()
+
+
+class GRPCClient(_GRPCRevision):
+    """Server-TLS gRPC on the canonical internal RPC, for an explicit HTTPS origin."""
+
+    def __init__(self, origin, credential, *, tls_context=None):
+        if grpc is None:
+            raise ImportError(
+                "Install saas-sdk-python[revision-grpc] for gRPC revision checks"
+            )
+        url = urlsplit(_https_origin(origin))
+        context = (
+            tls_context if tls_context is not None else ssl.create_default_context()
+        )
+        if not context.check_hostname or context.verify_mode != ssl.CERT_REQUIRED:
+            raise ValueError("Accounts TLS verification cannot be disabled")
+        # Use the deployment-selected trust roots, not gRPC environment overrides.
+        roots = b"".join(
+            ssl.DER_cert_to_PEM_cert(cert).encode("ascii")
+            for cert in context.get_ca_certs(binary_form=True)
+        )
+        if not roots:
+            raise ValueError("Accounts requires trusted TLS roots")
+        host = f"[{url.hostname}]" if ":" in url.hostname else url.hostname
+        self._bind(
+            grpc.aio.secure_channel(
+                f"{host}:{url.port or 443}",
+                grpc.ssl_channel_credentials(root_certificates=roots),
+                options=_CHANNEL_OPTIONS,
+            ),
+            credential,
+        )
+
+
+class MeshGRPCClient(_GRPCRevision):
+    """h2c gRPC to the resolved address of accounts' private ``rest`` endpoint.
+
+    This is the transport a current host serves (``INTERNAL_TRANSPORT.md``):
+    the mesh authenticates both workloads and encrypts the hop, and the process
+    adds no TLS of its own. Pass the ``host:port`` Codefly resolved for the
+    endpoint, never a URL. Deploy it only where the mesh enforces mutual TLS on
+    the hop; outside one, the internal credential would cross the wire in the
+    clear.
+    """
+
+    def __init__(self, address, credential):
+        if grpc is None:
+            raise ImportError(
+                "Install saas-sdk-python[revision-grpc] for gRPC revision checks"
+            )
+        self._bind(
+            grpc.aio.insecure_channel(_mesh_address(address), options=_CHANNEL_OPTIONS),
+            credential,
+        )
+

src/saas_sdk/work_context.py

@@ -62,6 +62,7 @@
     "MAX_TTL",
     "WorkContextError",
     "WorkContextDenied",
+    "WorkContextUnavailable",
     "WorkContextMintError",
     "WorkScope",
     "WorkActor",
@@ -133,6 +134,13 @@ class WorkContextDenied(WorkContextError):
     told apart from invalid capabilities."""
 
 
+class WorkContextUnavailable(WorkContextError):
+    """The published key set could not be fetched or read, so no presented
+    token was judged. Distinct from :class:`WorkContextError` so a verifier
+    outage can be reported as the authority being unavailable rather than as
+    the caller presenting an invalid capability; both still fail closed."""
+
+
 @dataclass(frozen=True)
 class WorkScope:
     resource_kind: str
@@ -409,7 +417,10 @@ def _refresh_unknown(self, observed_generation: int) -> tuple[Verifier, frozense
     def _refresh_locked(self) -> tuple[Verifier, frozenset[str], int]:
         try:
             keys = self._fetch()
-            verifier = Verifier(keys, now=self._now, clock_skew=self._clock_skew)
+            try:
+                verifier = Verifier(keys, now=self._now, clock_skew=self._clock_skew)
+            except WorkContextError as error:
+                raise WorkContextUnavailable(str(error)) from None
         except WorkContextError as error:
             # Open a short backoff window so a failed refresh does not re-hit the
             # endpoint on the next request (avoids an outage-driven fetch storm).
@@ -431,17 +442,21 @@ def _fetch(self) -> dict[str, bytes]:
         try:
             response = self._opener.open(request, timeout=self._request_timeout)
         except urllib.error.HTTPError as error:
-            raise WorkContextError(f"Work Context JWKS returned HTTP {error.code}") from None
+            raise WorkContextUnavailable(f"Work Context JWKS returned HTTP {error.code}") from None
         except (urllib.error.URLError, OSError) as error:
-            raise WorkContextError(f"fetch Work Context JWKS: {error}") from None
+            raise WorkContextUnavailable(f"fetch Work Context JWKS: {error}") from None
         with response:
             content_type = response.headers.get("Content-Type")
             if content_type and content_type.split(";", 1)[0].strip().lower() != "application/json":
-                raise WorkContextError("Work Context JWKS is not application/json")
+                raise WorkContextUnavailable("Work Context JWKS is not application/json")
             payload = response.read(_JWKS_MAX_BYTES + 1)
         if len(payload) > _JWKS_MAX_BYTES:
-            raise WorkContextError(f"Work Context JWKS exceeds {_JWKS_MAX_BYTES} bytes")
-        return _parse_jwks(payload)
+            raise WorkContextUnavailable(f"Work Context JWKS exceeds {_JWKS_MAX_BYTES} bytes")
+        try:
+            return _parse_jwks(payload)
+        except WorkContextError as error:
+            # A malformed key set is the authority's failure, never the caller's.
+            raise WorkContextUnavailable(str(error)) from None
 
 
 def _parse_jwks(payload: bytes) -> dict[str, bytes]:

tests/test_authorization_revision.py

@@ -246,8 +246,8 @@ async def respond(request):
         await client.aclose()
 
 
-@pytest_asyncio.fixture
-async def grpc_endpoint(tmp_path):
+@pytest_asyncio.fixture(params=["tls", "mesh"])
+async def grpc_endpoint(request, tmp_path):
     key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
     name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])
     now = datetime.now(timezone.utc)
@@ -305,12 +305,19 @@ async def check(request, context):
             ),
         )
     )
-    port = server.add_secure_port(
-        "localhost:0", grpc.ssl_server_credentials(((key_bytes, cert_bytes),))
-    )
+    if request.param == "tls":
+        port = server.add_secure_port(
+            "localhost:0", grpc.ssl_server_credentials(((key_bytes, cert_bytes),))
+        )
+    else:
+        # The current host's transport: h2c, with the mesh securing the hop.
+        port = server.add_insecure_port("localhost:0")
     await server.start()
-    tls = ssl.create_default_context(cadata=cert_bytes.decode())
-    client = revision.GRPCClient(f"https://localhost:{port}", token, tls_context=tls)
+    if request.param == "tls":
+        tls = ssl.create_default_context(cadata=cert_bytes.decode())
+        client = revision.GRPCClient(f"https://localhost:{port}", token, tls_context=tls)
+    else:
+        client = revision.MeshGRPCClient(f"localhost:{port}", token)
     try:
         yield client, state
     finally:
@@ -443,3 +450,41 @@ async def respond(request):
         assert cancelled.is_set()
     finally:
         await client.aclose()
+
+
+@pytest.mark.parametrize(
+    "address",
+    [
+        "",
+        "accounts.invalid",
+        "accounts.invalid:0",
+        "accounts.invalid:65536",
+        "accounts.invalid:port",
+        "http://accounts.invalid:8080",
+        "https://accounts.invalid:8080",
+        "//accounts.invalid:8080",
+        "user@accounts.invalid:8080",
+        "accounts.invalid:8080/path",
+        "accounts.invalid:8080?x=1",
+        "accounts.invalid:8080#x",
+        "accounts.invalid :8080",
+        ":8080",
+    ],
+)
+async def test_mesh_address_is_a_resolved_host_and_port(address):
+    with pytest.raises(ValueError, match="host:port"):
+        revision.MeshGRPCClient(address, token)
+
+
+@pytest.mark.parametrize(
+    "address,target",
+    [
+        ("accounts.platform-saas.svc.cluster.local:8080", "accounts.platform-saas.svc.cluster.local:8080"),
+        ("127.0.0.1:8080", "127.0.0.1:8080"),
+        ("[::1]:8080", "[::1]:8080"),
+    ],
+)
+async def test_mesh_address_accepts_resolved_forms(address, target):
+    assert revision._mesh_address(address) == target
+    client = revision.MeshGRPCClient(address, token)
+    await client.aclose()