Function _make_fail_closed_bearer_provider

python/openshell/sandbox.py:883–918

Per-RPC provider that re-reads `oidc_token.json` but does NOT refresh. Raises `SandboxError` when the token is missing or expired. Available as an opt-out from the default `_OidcRefresher` for callers (e.g. tests) that want to assert expiry behavior or that don't want the SDK to make

(
    gateway_dir: pathlib.Path,
    cluster_name: str,
)

def _make_fail_closed_bearer_provider(
    gateway_dir: pathlib.Path,
    cluster_name: str,
) -> Callable[[], str]:
    """Per-RPC provider that re-reads `oidc_token.json` but does NOT refresh.

    Raises `SandboxError` when the token is missing or expired. Available as
    an opt-out from the default `_OidcRefresher` for callers (e.g. tests)
    that want to assert expiry behavior or that don't want the SDK to make
    outbound HTTP calls to the IdP.
    """

    def provider() -> str:
        bundle = _read_oidc_token_bundle(gateway_dir)
        if bundle is None:
            raise SandboxError(
                f"OIDC token for gateway '{cluster_name}' is missing or "
                f"unreadable. Re-authenticate with: openshell gateway login"
            )
        access_token = bundle.get("access_token")
        if not isinstance(access_token, str) or not access_token:
            raise SandboxError(
                f"OIDC token for gateway '{cluster_name}' has no access "
                f"token. Re-authenticate with: openshell gateway login"
            )
        expires_at = bundle.get("expires_at")
        if isinstance(expires_at, int):
            now = int(time.time())
            if now + _OIDC_TOKEN_EXPIRY_GRACE_SECONDS >= expires_at:
                raise SandboxError(
                    f"OIDC token for gateway '{cluster_name}' has expired. "
                    f"Re-authenticate with: openshell gateway login"
                )
        return access_token

    return provider


class _InvalidGrantError(SandboxError):