From 1d2d5c66322370f0c51ed2fb94bc5f11b3844b95 Mon Sep 17 00:00:00 2001 From: Luka Zhupa Date: Mon, 14 Sep 2026 15:41:37 -0400 Subject: [PATCH] Add current authorization revision protocol clients --- .github/workflows/ci.yml | 2 +- README.md | 46 +- pyproject.toml | 4 +- src/saas_sdk/authorization_revision.py | 295 ++++++++++++ .../authorization_revision_projection.json | 115 +++++ tests/test_authorization_revision.py | 445 ++++++++++++++++++ 6 files changed, 904 insertions(+), 3 deletions(-) create mode 100644 src/saas_sdk/authorization_revision.py create mode 100644 tests/fixtures/authorization_revision_projection.json create mode 100644 tests/test_authorization_revision.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e1c9f88..712444f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -17,7 +17,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - run: python -m pip install --upgrade pip - - run: pip install -e ".[test]" + - run: pip install -e ".[test,revision-connect,revision-grpc]" # Runs the Work Context wire-golden parity test, which pins the token # byte-for-byte to sdk-go's TestWorkContextWireGolden. - run: python -m pytest -q diff --git a/README.md b/README.md index 24016c1..653afa7 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ Everything ships under a single top-level package, `saas_sdk`, so installing it never claims a generic name (`saas`, `buf`, `datasource`) in the consumer's import namespace. -Three layers: +Packages: - **`saas_sdk._gen`** — the generated protobuf message bindings (`*_pb2`), from the accounts public proto at the ref recorded in `SOURCE.txt`. Only message @@ -117,6 +117,50 @@ raising `WorkContextError` (or `WorkContextDenied` from `require_scope`) on any failure. Replay policy is reported on the claims; enforcing single-use consumption still requires a durable replay store the caller owns. +## Current authorization revision + +`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 +signature/lifetime verification, action/resource policy and public error mapping. + +```python +from saas_sdk.authorization_revision import GRPCClient, RevisionDenied, RevisionUnavailable + +client = GRPCClient("https://accounts.internal:50051", 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 +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 +extra and returns the canonical generated protobuf request. + +Each check loads the current internal service credential through an async callback +and makes one query. It does not cache authorization, retry, inherit environment +proxies, follow redirects or forward caller bearer tokens, cookies or Work Context +headers. One three-second budget covers credential loading, connection and the +complete response; caller cancellation propagates. The callback and any injected +fixture transport must cooperate with cancellation. `tls_context` selects verified +trust roots; TLS verification cannot be disabled. Connect's explicit `transport` +injection exists for fixtures and is trusted to implement its own transport +security; production should use the default transport. + +Only gRPC `PermissionDenied`/`FailedPrecondition`, or their coherent Connect +status/code pairs, raise `RevisionDenied`. Credential, transport and malformed +response failures raise `RevisionUnavailable`. Responses must be the canonical +empty message, and error strings omit remote details and credentials. Neither +error grants authority; consumers explicitly map them onto their public API. + +The projection fixture at `tests/fixtures/authorization_revision_projection.json` +is shared byte-for-byte with the Go SDK's +`authorizationrevision/testdata/projection.json`. Owner/actor scope projection is +therefore verified independently of transport and consumer policy. + ## Versioning This SDK's release version tracks the **saas-starter module version** diff --git a/pyproject.toml b/pyproject.toml index df5c6c1..1cffa4b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,9 @@ requires-python = ">=3.10" dependencies = ["protobuf>=5.29.3", "cryptography>=41"] [project.optional-dependencies] -test = ["pytest>=8"] +test = ["pytest>=8", "pytest-asyncio>=0.24"] +revision-connect = ["httpx>=0.27,<1"] +revision-grpc = ["grpcio>=1.66"] [tool.setuptools.packages.find] where = ["src"] diff --git a/src/saas_sdk/authorization_revision.py b/src/saas_sdk/authorization_revision.py new file mode 100644 index 0000000..1b58930 --- /dev/null +++ b/src/saas_sdk/authorization_revision.py @@ -0,0 +1,295 @@ +"""Accounts current-revision protocol for already verified Work Context claims. + +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. +""" + +import asyncio +import json +import ssl +from collections.abc import Awaitable, Callable +from urllib.parse import urlsplit + +from google.protobuf.json_format import MessageToDict + +from ._gen import work_contexts_pb2 as pb +from .work_context import WorkContext + +try: + import grpc +except ImportError: + grpc = None +try: + import httpx +except ImportError: + httpx = None + +_SERVICE = pb.DESCRIPTOR.services_by_name["WorkContextService"] +_METHOD = _SERVICE.methods_by_name["CheckAuthorizationRevision"] +REVISION_PATH = "/" + _SERVICE.full_name + "/" + _METHOD.name +TIMEOUT = 3.0 + + +class RevisionDenied(Exception): + """Accounts reported PermissionDenied or a stale/revoked revision.""" + + +class RevisionUnavailable(Exception): + """Internal credential, transport or response-contract failure; no authority granted.""" + + +def _decode(data): + def pairs(items): + value = {} + for key, item in items: + if key in value: + raise ValueError("Duplicate JSON key") + value[key] = item + return value + + def invalid(value): + raise ValueError("Non-finite JSON value") + + return json.loads(data, object_pairs_hook=pairs, parse_constant=invalid) + + +async def _internal_credential(provider: Callable[[], Awaitable[str]]) -> str: + try: + token = await provider() + except Exception: + raise RevisionUnavailable("Current authority is unavailable") from None + if ( + not isinstance(token, str) + or not 1 <= len(token) <= 8192 + or any(ord(c) < 33 or ord(c) > 126 for c in token) + ): + raise RevisionUnavailable("Current authority is unavailable") + return token + + +def _https_origin(value: str) -> str: + url = urlsplit(value) + if ( + url.scheme != "https" + or not url.hostname + or url.username is not None + or url.password is not None + or url.path not in ("", "/") + or url.query + or url.fragment + or "\\" in value + or any(c.isspace() or ord(c) < 32 for c in value) + or url.port == 0 + ): + raise ValueError("Accounts requires an explicit HTTPS internal origin") + return value.rstrip("/") + + +def revision_request(claims: WorkContext) -> pb.CheckAuthorizationRevisionRequest: + """Copy owner and every actor scope; the caller owns verification and policy.""" + if ( + not isinstance(claims, WorkContext) + or not claims.tenant_id + or not claims.owner_principal_id + ): + raise RevisionUnavailable("Current authority is unavailable") + + def subject(principal, scopes): + if not principal: + raise RevisionUnavailable("Current authority is unavailable") + return pb.WorkContextRevisionSubject( + principal_id=principal, + scopes=[ + pb.WorkContextScope( + resource_kind=s.resource_kind, + actions=s.actions, + resource_ids=s.resource_ids, + ) + for s in scopes + ], + ) + + return pb.CheckAuthorizationRevisionRequest( + org_id=claims.tenant_id, + owner_principal_id=claims.owner_principal_id, + authorization_revision=claims.authorization_revision, + subjects=[ + subject(claims.owner_principal_id, claims.authority_scopes), + *(subject(a.principal_id, a.granted_scopes) for a in claims.actor_chain), + ], + ) + + +class ConnectClient: + """Connect compatibility client; select only when the authority endpoint serves Connect.""" + + def __init__(self, origin, credential, *, transport=None, tls_context=None): + if httpx is None: + raise ImportError( + "Install saas-sdk-python[revision-connect] for Connect revision checks" + ) + self.endpoint = _https_origin(origin) + REVISION_PATH + self.credential = credential + if tls_context is not None and ( + not tls_context.check_hostname + or tls_context.verify_mode != ssl.CERT_REQUIRED + ): + raise ValueError("Accounts TLS verification cannot be disabled") + self.client = httpx.AsyncClient( + transport=transport + if transport is not None + else httpx.AsyncHTTPTransport( + retries=0, trust_env=False, verify=tls_context if tls_context else True + ), + trust_env=False, + follow_redirects=False, + ) + + async def check(self, claims: WorkContext) -> None: + try: + await asyncio.wait_for(self._check(claims), timeout=TIMEOUT) + except asyncio.TimeoutError: + raise RevisionUnavailable("Current authority is unavailable") from None + + async def _check(self, claims): + response = None + try: + # Includes credential refresh, pool wait, connection and the complete body. + token = await _internal_credential(self.credential) + body = MessageToDict( + revision_request(claims), preserving_proto_field_name=True + ) + # Construct directly: never inherit caller headers, client cookies or auth. + request = httpx.Request( + "POST", + self.endpoint, + headers={ + "content-type": "application/json", + "accept": "application/json", + "accept-encoding": "identity", + "connect-protocol-version": "1", + "x-codefly-internal-token": token, + }, + content=json.dumps(body).encode(), + extensions={ + "timeout": dict.fromkeys(("connect", "read", "write", "pool"), 3) + }, + ) + response = await self.client.send(request, stream=True, auth=None) + if response.status_code not in (200, 400, 403): + raise ValueError("Accounts unavailable") + if ( + response.headers.get("content-type", "").split(";", 1)[0] + != "application/json" + ): + raise ValueError("Invalid Accounts response type") + if response.headers.get("content-encoding", "identity") != "identity": + raise ValueError("Compressed Accounts responses are not supported") + data = bytearray() + async for part in response.aiter_bytes(): + data.extend(part) + if len(data) > 8192: + raise ValueError("Accounts response exceeds bound") + payload = _decode(data) + if response.status_code != 200: + # Connect maps gRPC FailedPrecondition to 400, not HTTP 412. + # An internal 401 is a service credential outage, never caller denial. + if isinstance(payload, dict) and ( + response.status_code, + payload.get("code"), + ) in ( + (400, "failed_precondition"), + (403, "permission_denied"), + ): + raise RevisionDenied("Current authority was not confirmed") + raise ValueError("Invalid Accounts error response") + if payload != {}: + raise ValueError("Expected canonical empty revision response") + except (httpx.HTTPError, OSError, ValueError, TimeoutError): + raise RevisionUnavailable("Current authority is unavailable") from None + finally: + if response is not None: + await response.aclose() + + async def aclose(self) -> None: + await self.client.aclose() + + +class GRPCClient: + """TLS gRPC on the canonical internal RPC; no retries, fallback or authority cache.""" + + 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)) + 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.rpc = self.channel.unary_unary( + REVISION_PATH, + request_serializer=pb.CheckAuthorizationRevisionRequest.SerializeToString, + # The canonical response is google.protobuf.Empty. Reject unknown fields + # rather than silently accepting a different response contract. + response_deserializer=lambda data: data, + ) + + async def check(self, claims: WorkContext) -> None: + try: + await asyncio.wait_for(self._check(claims), timeout=TIMEOUT) + except asyncio.TimeoutError: + raise RevisionUnavailable("Current authority is unavailable") from None + + 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) + 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),), + timeout=remaining, + wait_for_ready=False, + ) + if response != b"": + raise ValueError("Expected canonical empty revision response") + except grpc.aio.AioRpcError as error: + if error.code() in ( + grpc.StatusCode.PERMISSION_DENIED, + grpc.StatusCode.FAILED_PRECONDITION, + ): + raise RevisionDenied("Current authority was not confirmed") from None + # Internal credential failures are service outages, never caller denials. + # Do not expose provider details, trailers, metadata or credentials. + raise RevisionUnavailable("Current authority is unavailable") from None + except (OSError, ValueError, TimeoutError): + raise RevisionUnavailable("Current authority is unavailable") from None + + async def aclose(self) -> None: + await self.channel.close() diff --git a/tests/fixtures/authorization_revision_projection.json b/tests/fixtures/authorization_revision_projection.json new file mode 100644 index 0000000..9894100 --- /dev/null +++ b/tests/fixtures/authorization_revision_projection.json @@ -0,0 +1,115 @@ +{ + "claims": { + "tenant_id": "tenant-1", + "owner_principal_id": "owner-1", + "authorization_revision": "42", + "authority_scopes": [ + { + "resource_kind": "example.items", + "actions": [ + "read", + "write" + ], + "resource_ids": [ + "item-2", + "item-1" + ] + }, + { + "resource_kind": "example.status", + "actions": [ + "read" + ] + } + ], + "actor_chain": [ + { + "principal_id": "actor-1", + "granted_scopes": [ + { + "resource_kind": "example.items", + "actions": [ + "read", + "write" + ], + "resource_ids": [ + "item-1" + ] + } + ] + }, + { + "principal_id": "actor-2", + "granted_scopes": [ + { + "resource_kind": "example.items", + "actions": [ + "read" + ], + "resource_ids": [ + "item-1" + ] + } + ] + } + ] + }, + "request": { + "org_id": "tenant-1", + "owner_principal_id": "owner-1", + "authorization_revision": "42", + "subjects": [ + { + "principal_id": "owner-1", + "scopes": [ + { + "resource_kind": "example.items", + "actions": [ + "read", + "write" + ], + "resource_ids": [ + "item-2", + "item-1" + ] + }, + { + "resource_kind": "example.status", + "actions": [ + "read" + ] + } + ] + }, + { + "principal_id": "actor-1", + "scopes": [ + { + "resource_kind": "example.items", + "actions": [ + "read", + "write" + ], + "resource_ids": [ + "item-1" + ] + } + ] + }, + { + "principal_id": "actor-2", + "scopes": [ + { + "resource_kind": "example.items", + "actions": [ + "read" + ], + "resource_ids": [ + "item-1" + ] + } + ] + } + ] + } +} diff --git a/tests/test_authorization_revision.py b/tests/test_authorization_revision.py new file mode 100644 index 0000000..a711451 --- /dev/null +++ b/tests/test_authorization_revision.py @@ -0,0 +1,445 @@ +"""Owner protocol conformance, without consumer permissions or provider calls.""" + +import asyncio +import json +import ssl +from dataclasses import replace +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import grpc +import httpx +import pytest +import pytest_asyncio +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID +from google.protobuf.json_format import ParseDict + +from saas_sdk import authorization_revision as revision +from saas_sdk import work_context as wc +from saas_sdk._gen import work_contexts_pb2 as pb + +pytestmark = pytest.mark.asyncio + + +def projection_fixture(): + return json.loads( + ( + Path(__file__).parent / "fixtures/authorization_revision_projection.json" + ).read_text() + ) + + +def claims(): + value = projection_fixture()["claims"] + + def scopes(values): + return tuple( + wc.WorkScope( + s["resource_kind"], + tuple(s["actions"]), + tuple(s.get("resource_ids", ())), + ) + for s in values + ) + + return wc.WorkContext( + typ=wc.WORK_CONTEXT_TYPE, + algorithm="Ed25519", + key_id="key", + issuer="issuer", + audience="audience", + not_before_unix=1, + issued_at_unix=1, + expires_at_unix=2, + nonce="nonce", + authorization_revision=int(value["authorization_revision"]), + replay_policy="idempotent", + tenant_id=value["tenant_id"], + owner_principal_id=value["owner_principal_id"], + task_id="task", + session_id="session", + parent_session_id=None, + authority_scopes=scopes(value["authority_scopes"]), + actor_chain=tuple( + wc.WorkActor( + a["principal_id"], "service", "delegation", scopes(a["granted_scopes"]) + ) + for a in value["actor_chain"] + ), + attribution_team_ids=(), + workspace_id=None, + project_id=None, + ) + + +async def token(): + return "internal-fixture" + + +async def test_shared_projection(): + value = claims() + request = revision.revision_request(value) + assert request == ParseDict( + projection_fixture()["request"], pb.CheckAuthorizationRevisionRequest() + ) + request.subjects[0].scopes[0].actions[0] = "mutated" + assert value.authority_scopes[0].actions == ("read", "write") + assert ( + revision.REVISION_PATH + == "/saas.accounts.v1.WorkContextService/CheckAuthorizationRevision" + ) + + +@pytest.mark.parametrize( + "status,body,error", + [ + (200, b"{}", None), + (200, b"null", revision.RevisionUnavailable), + (200, b"[]", revision.RevisionUnavailable), + (200, b"true", revision.RevisionUnavailable), + (200, b"{}{}", revision.RevisionUnavailable), + (200, b'{"x":1}', revision.RevisionUnavailable), + (200, b'{"x":1,"x":2}', revision.RevisionUnavailable), + (200, b'{"x":NaN}', revision.RevisionUnavailable), + (200, b" " * 8193, revision.RevisionUnavailable), + (400, b'{"code":"failed_precondition"}', revision.RevisionDenied), + ( + 403, + b'{"code":"permission_denied","message":"private-detail"}', + revision.RevisionDenied, + ), + (400, b'{"code":"permission_denied"}', revision.RevisionUnavailable), + (403, b'{"code":"failed_precondition"}', revision.RevisionUnavailable), + (401, b'{"code":"permission_denied"}', revision.RevisionUnavailable), + (307, b'{"code":"permission_denied"}', revision.RevisionUnavailable), + (503, b'{"code":"permission_denied"}', revision.RevisionUnavailable), + ( + 403, + b'{"code":"permission_denied","code":"permission_denied"}', + revision.RevisionUnavailable, + ), + ], +) +async def test_connect_exact_empty_and_coherent_errors(status, body, error): + calls = [] + + async def respond(request): + calls.append(request) + return httpx.Response( + status, + content=body, + headers={ + "content-type": "application/json", + "location": "https://redirect.invalid", + }, + ) + + client = revision.ConnectClient( + "https://accounts.invalid", token, transport=httpx.MockTransport(respond) + ) + try: + if error is None: + await client.check(claims()) + else: + with pytest.raises(error) as caught: + await client.check(claims()) + assert "private-detail" not in str(caught.value) + assert len(calls) == 1 + finally: + await client.aclose() + + +async def test_connect_refresh_revocation_and_no_ambient_credentials(): + current = [42, "internal-one"] + calls = [] + credential_calls = [] + + async def credential(): + credential_calls.append(current[1]) + return current[1] + + async def respond(request): + calls.append(request) + assert request.headers["x-codefly-internal-token"] == current[1] + assert not {"cookie", "authorization", "x-codefly-work-context"}.intersection( + request.headers + ) + body = json.loads(request.content) + if int(body["authorization_revision"]) != current[0]: + return httpx.Response(400, json={"code": "failed_precondition"}) + return httpx.Response(200, json={}) + + client = revision.ConnectClient( + "https://accounts.invalid", credential, transport=httpx.MockTransport(respond) + ) + client.client.cookies.set("private", "caller") + client.client.headers["authorization"] = "Bearer caller" + try: + await client.check(claims()) + current[:] = [43, "internal-two"] + with pytest.raises(revision.RevisionDenied): + await client.check(claims()) + await client.check(replace(claims(), authorization_revision=43)) + assert len(calls) == len(credential_calls) == 3 + finally: + await client.aclose() + + +@pytest.mark.parametrize("value", [None, "", "with space", "bad\nvalue", "x" * 8193]) +async def test_invalid_credential_does_not_dispatch(value): + calls = [] + + async def credential(): + return value + + async def respond(request): + calls.append(request) + return httpx.Response(200, json={}) + + client = revision.ConnectClient( + "https://accounts.invalid", credential, transport=httpx.MockTransport(respond) + ) + try: + with pytest.raises(revision.RevisionUnavailable): + await client.check(claims()) + assert calls == [] + finally: + await client.aclose() + + +async def test_connect_total_budget_includes_credential_and_stream(monkeypatch): + monkeypatch.setattr(revision, "TIMEOUT", 0.08) + cancelled = asyncio.Event() + closed = asyncio.Event() + + class Body(httpx.AsyncByteStream): + async def __aiter__(self): + try: + await asyncio.sleep(10) + yield b"{}" + finally: + cancelled.set() + + async def aclose(self): + closed.set() + + async def credential(): + await asyncio.sleep(0.03) + return "internal-fixture" + + async def respond(request): + return httpx.Response( + 200, stream=Body(), headers={"content-type": "application/json"} + ) + + client = revision.ConnectClient( + "https://accounts.invalid", credential, transport=httpx.MockTransport(respond) + ) + try: + with pytest.raises(revision.RevisionUnavailable): + await client.check(claims()) + assert cancelled.is_set() and closed.is_set() + finally: + await client.aclose() + + +@pytest_asyncio.fixture +async def grpc_endpoint(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) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(hours=1)) + .add_extension(x509.SubjectAlternativeName([x509.DNSName("localhost")]), False) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), True) + .sign(key, hashes.SHA256()) + ) + cert_bytes = cert.public_bytes(serialization.Encoding.PEM) + key_bytes = key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + state = { + "calls": [], + "status": None, + "response": b"", + "delay": 0, + "cancelled": asyncio.Event(), + "started": asyncio.Event(), + } + + async def check(request, context): + state["calls"].append((request, dict(context.invocation_metadata()))) + state["started"].set() + try: + if state["delay"]: + await asyncio.sleep(state["delay"]) + if state["status"]: + await context.abort(state["status"], "private-detail") + return state["response"] + except asyncio.CancelledError: + state["cancelled"].set() + raise + + server = grpc.aio.server() + handler = grpc.unary_unary_rpc_method_handler( + check, + request_deserializer=pb.CheckAuthorizationRevisionRequest.FromString, + response_serializer=lambda value: value, + ) + server.add_generic_rpc_handlers( + ( + grpc.method_handlers_generic_handler( + "saas.accounts.v1.WorkContextService", + {"CheckAuthorizationRevision": handler}, + ), + ) + ) + port = server.add_secure_port( + "localhost:0", grpc.ssl_server_credentials(((key_bytes, cert_bytes),)) + ) + await server.start() + tls = ssl.create_default_context(cadata=cert_bytes.decode()) + client = revision.GRPCClient(f"https://localhost:{port}", token, tls_context=tls) + try: + yield client, state + finally: + await client.aclose() + await server.stop(None) + + +async def test_grpc_canonical_wire_credential_refresh_and_current_denial(grpc_endpoint): + client, state = grpc_endpoint + await client.check(claims()) + + async def rotated(): + return "internal-rotated" + + client.credential = rotated + state["status"] = grpc.StatusCode.FAILED_PRECONDITION + with pytest.raises(revision.RevisionDenied): + await client.check(claims()) + assert len(state["calls"]) == 2 + assert [m["x-codefly-internal-token"] for _, m in state["calls"]] == [ + "internal-fixture", + "internal-rotated", + ] + for request, metadata in state["calls"]: + assert request == revision.revision_request(claims()) + assert not {"authorization", "cookie", "x-codefly-work-context"}.intersection( + metadata + ) + + +@pytest.mark.parametrize( + "status,error", + [ + (grpc.StatusCode.PERMISSION_DENIED, revision.RevisionDenied), + (grpc.StatusCode.FAILED_PRECONDITION, revision.RevisionDenied), + (grpc.StatusCode.UNAUTHENTICATED, revision.RevisionUnavailable), + (grpc.StatusCode.UNAVAILABLE, revision.RevisionUnavailable), + ], +) +async def test_grpc_typed_errors_no_retry(grpc_endpoint, status, error): + client, state = grpc_endpoint + state["status"] = status + with pytest.raises(error) as caught: + await client.check(claims()) + assert "private-detail" not in str(caught.value) + assert len(state["calls"]) == 1 + + +@pytest.mark.parametrize("body", [b"\x08\x01", b"x" * 8193]) +async def test_grpc_exact_empty_response(grpc_endpoint, body): + client, state = grpc_endpoint + state["response"] = body + with pytest.raises(revision.RevisionUnavailable): + await client.check(claims()) + assert len(state["calls"]) == 1 + + +async def test_grpc_caller_cancellation(grpc_endpoint): + client, state = grpc_endpoint + state["delay"] = 10 + task = asyncio.create_task(client.check(claims())) + await asyncio.wait_for(state["started"].wait(), 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + await asyncio.wait_for(state["cancelled"].wait(), 1) + assert len(state["calls"]) == 1 + + +async def test_grpc_total_budget_includes_credential(grpc_endpoint, monkeypatch): + monkeypatch.setattr(revision, "TIMEOUT", 0.1) + client, state = grpc_endpoint + + async def delayed(): + await asyncio.sleep(0.03) + return "internal-fixture" + + client.credential = delayed + state["delay"] = 10 + with pytest.raises(revision.RevisionUnavailable): + await client.check(claims()) + await asyncio.wait_for(state["cancelled"].wait(), 1) + assert len(state["calls"]) == 1 + + +@pytest.mark.parametrize( + "origin", + [ + "http://accounts.invalid", + "https://user@accounts.invalid", + "https://accounts.invalid/path", + "https://accounts.invalid?other=x", + "https://accounts.invalid#other", + "https://accounts.invalid:0", + ], +) +async def test_explicit_https_origin_required(origin): + for kind in (revision.GRPCClient, revision.ConnectClient): + with pytest.raises(ValueError): + kind(origin, token) + + +async def test_tls_verification_cannot_be_disabled(): + tls = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + tls.check_hostname = False + for kind in (revision.GRPCClient, revision.ConnectClient): + with pytest.raises(ValueError, match="cannot be disabled"): + kind("https://accounts.invalid", token, tls_context=tls) + + +async def test_credential_cancellation_prevents_dispatch(monkeypatch): + monkeypatch.setattr(revision, "TIMEOUT", 0.03) + cancelled = asyncio.Event() + + async def credential(): + try: + await asyncio.sleep(10) + finally: + cancelled.set() + + async def respond(request): + pytest.fail("credential failure dispatched") + + client = revision.ConnectClient( + "https://accounts.invalid", credential, transport=httpx.MockTransport(respond) + ) + try: + with pytest.raises(revision.RevisionUnavailable): + await client.check(claims()) + assert cancelled.is_set() + finally: + await client.aclose()