feat(cloud-auth): verify scoped bearer tokens

This commit is contained in:
2026-07-12 17:45:31 +08:00
parent 4cb116f527
commit b916b394c7
4 changed files with 186 additions and 27 deletions
+93
View File
@@ -0,0 +1,93 @@
from __future__ import annotations
from dataclasses import dataclass
import cloud.auth as auth_module
from cloud.auth import (
BearerCredential,
ConfiguredBearerAuthProvider,
NullAuthProvider,
)
@dataclass
class _Request:
headers: dict[str, str]
def test_configured_bearer_auth_returns_identity_and_scopes() -> None:
provider = ConfiguredBearerAuthProvider(
[
BearerCredential(
principal_id="integrator-a",
token="secret-a",
scopes=frozenset({"tasks:submit", "tasks:read"}),
)
]
)
principal = provider.authenticate(
_Request(headers={"authorization": "Bearer secret-a"})
)
assert principal is not None
assert principal.id == "integrator-a"
assert principal.scopes == frozenset({"tasks:submit", "tasks:read"})
def test_missing_malformed_or_invalid_bearer_token_is_rejected() -> None:
provider = ConfiguredBearerAuthProvider(
[BearerCredential(principal_id="integrator", token="valid-token")]
)
assert provider.authenticate(_Request(headers={})) is None
assert (
provider.authenticate(_Request(headers={"authorization": "Basic value"}))
is None
)
assert (
provider.authenticate(_Request(headers={"authorization": "Bearer invalid"}))
is None
)
def test_bearer_verification_compares_every_configured_digest(monkeypatch) -> None:
comparisons: list[tuple[bytes, bytes]] = []
original_compare_digest = auth_module.compare_digest
def recording_compare_digest(left: bytes, right: bytes) -> bool:
comparisons.append((left, right))
return original_compare_digest(left, right)
monkeypatch.setattr(auth_module, "compare_digest", recording_compare_digest)
provider = ConfiguredBearerAuthProvider(
[
BearerCredential(principal_id="first", token="match"),
BearerCredential(principal_id="second", token="other"),
]
)
principal = provider.authenticate(
_Request(headers={"authorization": "Bearer match"})
)
assert principal is not None
assert principal.id == "first"
assert len(comparisons) == 2
assert all(len(left) == len(right) == 32 for left, right in comparisons)
def test_credential_representations_do_not_expose_token() -> None:
credential = BearerCredential(principal_id="integrator", token="top-secret")
provider = ConfiguredBearerAuthProvider([credential])
assert "top-secret" not in repr(credential)
assert "top-secret" not in repr(provider.__dict__)
def test_null_auth_provider_is_explicitly_unrestricted() -> None:
principal = NullAuthProvider().authenticate(_Request(headers={}))
assert principal is not None
assert principal.id == "anonymous"
assert principal.scopes == frozenset({"*"})