diff --git a/README.md b/README.md index 653afa7..95b0669 100644 --- a/README.md +++ b/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 diff --git a/src/saas_sdk/authorization_revision.py b/src/saas_sdk/authorization_revision.py index 1b58930..7a8fc9b 100644 --- a/src/saas_sdk/authorization_revision.py +++ b/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, + ) + diff --git a/src/saas_sdk/work_context.py b/src/saas_sdk/work_context.py index 4e02ec4..609d0d4 100644 --- a/src/saas_sdk/work_context.py +++ b/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]: diff --git a/tests/test_authorization_revision.py b/tests/test_authorization_revision.py index a711451..a3a8b49 100644 --- a/tests/test_authorization_revision.py +++ b/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() diff --git a/tests/test_work_context_jwks.py b/tests/test_work_context_jwks.py index 44bca23..1e795bd 100644 --- a/tests/test_work_context_jwks.py +++ b/tests/test_work_context_jwks.py @@ -178,14 +178,14 @@ def handler(): now_box["unix"] += 61 # cache expired; endpoint now down healthy["ok"] = False for _ in range(5): - with pytest.raises(wc.WorkContextError): + with pytest.raises(wc.WorkContextUnavailable): verifier.verify(token) # One refresh attempt for the whole backoff window, not one per call. assert state.requests == 2 # Once the backoff elapses the verifier tries again (still down => one more). now_box["unix"] += int(wc._JWKS_FAILED_REFRESH_BACKOFF.total_seconds()) - with pytest.raises(wc.WorkContextError): + with pytest.raises(wc.WorkContextUnavailable): verifier.verify(token) assert state.requests == 3 finally: @@ -217,7 +217,7 @@ def do_GET(self): src_url = f"http://{host}:{port}/v1/auth/.well-known/jwks.json" try: verifier = _verifier(src_url) - with pytest.raises(wc.WorkContextError): + with pytest.raises(wc.WorkContextUnavailable): verifier.verify(_token("key-1", _seed(1))) assert destination_hits["n"] == 0 finally: @@ -230,7 +230,7 @@ def test_refresh_fails_closed_at_boot(): httpd, url = _server(state) try: verifier = _verifier(url) - with pytest.raises(wc.WorkContextError, match="HTTP 503"): + with pytest.raises(wc.WorkContextUnavailable, match="HTTP 503"): verifier.refresh() finally: httpd.shutdown() @@ -265,18 +265,34 @@ def test_rejects_malformed_or_bounded_jwks(response): httpd, url = _server(state) try: verifier = _verifier(url) - with pytest.raises(wc.WorkContextError): + with pytest.raises(wc.WorkContextUnavailable): verifier.verify(_token("key-1", _seed(1))) finally: httpd.shutdown() +def test_invalid_token_against_a_served_key_set_is_not_unavailability(): + """An outage and a bad capability both fail closed, but only the outage is + the authority's: a caller holding a token the served keys reject must see an + invalid-capability error, never "unavailable" (which invites a retry).""" + keys = {"key-1": _public_bytes(_seed(1))} + state = _State(_ok(keys)) + httpd, url = _server(state) + try: + verifier = _verifier(url) + with pytest.raises(wc.WorkContextError) as caught: + verifier.verify(_token("key-1", _seed(2))) + assert not isinstance(caught.value, wc.WorkContextUnavailable) + finally: + httpd.shutdown() + + def test_rejects_oversized_jwks(): body = b"x" * (256 * 1024 + 1) state = _State(lambda: (200, "application/json", body)) httpd, url = _server(state) try: - with pytest.raises(wc.WorkContextError, match="exceeds"): + with pytest.raises(wc.WorkContextUnavailable, match="exceeds"): _verifier(url).verify(_token("key-1", _seed(1))) finally: httpd.shutdown()