150 lines
4.4 KiB
Python
150 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import cloud.auth as auth_module
|
|
import pytest
|
|
from cloud.auth import (
|
|
BearerCredential,
|
|
ConfiguredBearerAuthProvider,
|
|
HostIdentityMismatchError,
|
|
HostPrincipalRequiredError,
|
|
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({"*"})
|
|
|
|
|
|
def test_host_credential_authenticates_as_one_bound_host() -> None:
|
|
provider = ConfiguredBearerAuthProvider(
|
|
[
|
|
BearerCredential(
|
|
principal_id="host-agent-a",
|
|
token="host-secret",
|
|
scopes=frozenset({"host:agent"}),
|
|
host_id="host-a",
|
|
)
|
|
]
|
|
)
|
|
|
|
principal = provider.authenticate(
|
|
_Request(headers={"authorization": "Bearer host-secret"})
|
|
)
|
|
|
|
assert principal is not None
|
|
assert principal.host_id == "host-a"
|
|
principal.require_host("host-a")
|
|
|
|
|
|
def test_host_principal_rejects_cross_host_operation() -> None:
|
|
provider = ConfiguredBearerAuthProvider(
|
|
[
|
|
BearerCredential(
|
|
principal_id="host-agent-a",
|
|
token="host-secret",
|
|
host_id="host-a",
|
|
)
|
|
]
|
|
)
|
|
principal = provider.authenticate(
|
|
_Request(headers={"authorization": "Bearer host-secret"})
|
|
)
|
|
|
|
assert principal is not None
|
|
with pytest.raises(HostIdentityMismatchError):
|
|
principal.require_host("host-b")
|
|
|
|
|
|
def test_public_principal_cannot_act_as_host() -> None:
|
|
provider = ConfiguredBearerAuthProvider(
|
|
[BearerCredential(principal_id="integrator", token="public-secret")]
|
|
)
|
|
principal = provider.authenticate(
|
|
_Request(headers={"authorization": "Bearer public-secret"})
|
|
)
|
|
|
|
assert principal is not None
|
|
with pytest.raises(HostPrincipalRequiredError):
|
|
principal.require_host("host-a")
|