667 lines
24 KiB
Python
667 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import copy
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
from aios_baseline.authorization import (
|
|
AuthorizationError,
|
|
AuthorizationService,
|
|
sign_owner_delegation_grant,
|
|
)
|
|
from aios_baseline.operations import OperationDefinition
|
|
|
|
|
|
class Clock:
|
|
def __init__(self, value: int = 2_000_000_000) -> None:
|
|
self.value = value
|
|
|
|
def __call__(self) -> int:
|
|
return self.value
|
|
|
|
|
|
def grant(
|
|
owner_key: Ed25519PrivateKey,
|
|
clock: Clock,
|
|
*,
|
|
grant_id: str = "grant-0001",
|
|
generation: int = 1,
|
|
tenant_id: str = "tenant-a",
|
|
subject_id: str = "owner-a",
|
|
task_ids: tuple[str, ...] = ("task-a",),
|
|
operations: tuple[str, ...] = ("deploy",),
|
|
expires_at: int | None = None,
|
|
max_effects: int | None = None,
|
|
destructive_ceiling: str = "NONE",
|
|
) -> dict:
|
|
return sign_owner_delegation_grant(
|
|
owner_key,
|
|
"owner-key-1",
|
|
grant_id=grant_id,
|
|
generation=generation,
|
|
tenant_id=tenant_id,
|
|
subject_id=subject_id,
|
|
mission_ids=("mission-a",),
|
|
task_ids=task_ids,
|
|
operations=operations,
|
|
component_ids=("component-a",),
|
|
environments=("staging",),
|
|
issued_at=clock(),
|
|
validity_mode="EXPIRES_AT" if expires_at is not None else "UNTIL_REVOKED",
|
|
expires_at=expires_at,
|
|
max_effects=max_effects,
|
|
destructive_ceiling=destructive_ceiling,
|
|
)
|
|
|
|
|
|
def service(tmp_path, owner_key: Ed25519PrivateKey, machine_key: Ed25519PrivateKey, clock: Clock, **kwargs):
|
|
return AuthorizationService(
|
|
tmp_path / "authorization.sqlite3",
|
|
{"owner-key-1": owner_key.public_key()},
|
|
machine_key,
|
|
"machine-key-1",
|
|
owner_key_scopes={
|
|
"owner-key-1": {
|
|
"tenant_ids": ["tenant-a"],
|
|
"subject_ids": ["owner-a"],
|
|
}
|
|
},
|
|
clock=clock,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
SCOPE = {
|
|
"tenant_id": "tenant-a",
|
|
"subject_id": "owner-a",
|
|
"mission_id": "mission-a",
|
|
"task_id": "task-a",
|
|
"operation": "deploy",
|
|
"component_id": "component-a",
|
|
"environment": "staging",
|
|
}
|
|
|
|
|
|
def test_signed_chain_authorizes_exact_narrowed_scope(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
signed_grant = grant(owner_key, clock)
|
|
|
|
registered = auth.register_grant(signed_grant)
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
assert registered == signed_grant
|
|
assert principal["schema"] == "aios.task-principal.v1"
|
|
assert principal["operational_epoch"]["schema"] == "aios.operational-epoch.v1"
|
|
assert principal["payload"]["root_grant_digest"] == auth.grant_digest(signed_grant)
|
|
assert principal["payload"]["root_grant_generation"] == 1
|
|
assert isinstance(principal["payload"]["credential_id"], str)
|
|
assert isinstance(principal["payload"]["nonce"], str)
|
|
assert auth.authorize(principal, **SCOPE) == signed_grant
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("target", "field", "replacement"),
|
|
[
|
|
("grant", "tenant_id", "tenant-b"),
|
|
("epoch", "task_id", "task-b"),
|
|
("principal", "operation", "delete"),
|
|
],
|
|
)
|
|
def test_any_signature_tampering_is_rejected(tmp_path, target, field, replacement):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
signed_grant = grant(owner_key, clock)
|
|
|
|
if target == "grant":
|
|
tampered = copy.deepcopy(signed_grant)
|
|
tampered["payload"][field] = replacement
|
|
with pytest.raises(AuthorizationError, match="signature"):
|
|
auth.register_grant(tampered)
|
|
return
|
|
|
|
auth.register_grant(signed_grant)
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
tampered = copy.deepcopy(principal)
|
|
if target == "epoch":
|
|
tampered["operational_epoch"]["payload"][field] = replacement
|
|
else:
|
|
tampered["payload"][field] = replacement
|
|
with pytest.raises(AuthorizationError, match="signature|parent"):
|
|
auth.authorize(tampered, **SCOPE)
|
|
|
|
|
|
def test_unknown_owner_key_and_unsigned_grant_are_rejected(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
unknown = grant(owner_key, clock)
|
|
unknown["signing"]["key_id"] = "attacker-key"
|
|
|
|
with pytest.raises(AuthorizationError, match="trusted owner key"):
|
|
auth.register_grant(unknown)
|
|
with pytest.raises(AuthorizationError, match="fields"):
|
|
auth.register_grant(grant(owner_key, clock)["payload"])
|
|
|
|
|
|
def test_fake_mapping_parent_and_parent_digest_substitution_are_rejected(tmp_path):
|
|
class FakeParent(dict):
|
|
pass
|
|
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
auth.register_grant(grant(owner_key, clock))
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
fake = copy.deepcopy(principal)
|
|
fake["operational_epoch"] = FakeParent(fake["operational_epoch"])
|
|
with pytest.raises(AuthorizationError, match="plain object"):
|
|
auth.authorize(fake, **SCOPE)
|
|
|
|
other = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
substituted = copy.deepcopy(principal)
|
|
substituted["operational_epoch"] = other["operational_epoch"]
|
|
with pytest.raises(AuthorizationError, match="parent|signature"):
|
|
auth.authorize(substituted, **SCOPE)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "wrong"),
|
|
[
|
|
("tenant_id", "tenant-b"),
|
|
("subject_id", "owner-b"),
|
|
("mission_id", "mission-b"),
|
|
("task_id", "task-b"),
|
|
("operation", "delete"),
|
|
("component_id", "component-b"),
|
|
("environment", "production"),
|
|
],
|
|
)
|
|
def test_authorize_denies_cross_scope_use(tmp_path, field, wrong):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
auth.register_grant(grant(owner_key, clock))
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
attempted = dict(SCOPE)
|
|
attempted[field] = wrong
|
|
|
|
with pytest.raises(AuthorizationError, match="scope"):
|
|
auth.authorize(principal, **attempted)
|
|
|
|
|
|
def test_principal_issuance_must_narrow_root_scope(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
auth.register_grant(grant(owner_key, clock, task_ids=("task-a", "task-b")))
|
|
|
|
denied = dict(SCOPE)
|
|
denied["task_id"] = "task-c"
|
|
with pytest.raises(AuthorizationError, match="grant scope"):
|
|
auth.issue_principal("grant-0001", ttl_seconds=30, **denied)
|
|
|
|
narrowed = dict(SCOPE)
|
|
narrowed["task_id"] = "task-b"
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **narrowed)
|
|
assert auth.authorize(principal, **narrowed)["payload"]["grant_id"] == "grant-0001"
|
|
|
|
|
|
def test_revocation_is_durable_and_checked_during_proof_lifetime(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
trusted_context = object()
|
|
verifier = lambda value: value is trusted_context
|
|
auth = service(
|
|
tmp_path,
|
|
owner_key,
|
|
machine_key,
|
|
clock,
|
|
control_context_verifier=verifier,
|
|
)
|
|
auth.register_grant(grant(owner_key, clock))
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=60, **SCOPE)
|
|
|
|
with pytest.raises(AuthorizationError, match="trusted control context"):
|
|
auth.revoke_grant("grant-0001", control_context=object())
|
|
assert auth.revoke_grant("grant-0001", control_context=trusted_context) == 2
|
|
with pytest.raises(AuthorizationError, match="generation|revoked"):
|
|
auth.authorize(principal, **SCOPE)
|
|
|
|
restarted = service(
|
|
tmp_path,
|
|
owner_key,
|
|
machine_key,
|
|
clock,
|
|
control_context_verifier=verifier,
|
|
)
|
|
with pytest.raises(AuthorizationError, match="generation|revoked"):
|
|
restarted.authorize(principal, **SCOPE)
|
|
|
|
|
|
def test_expired_proof_refreshes_without_owner_reapproval_and_survives_restart(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
signed_grant = grant(owner_key, clock)
|
|
auth.register_grant(signed_grant)
|
|
expired = auth.issue_principal("grant-0001", ttl_seconds=5, **SCOPE)
|
|
clock.value += 5
|
|
|
|
with pytest.raises(AuthorizationError, match="expired"):
|
|
auth.authorize(expired, **SCOPE)
|
|
|
|
restarted = service(tmp_path, owner_key, machine_key, clock)
|
|
refreshed = restarted.issue_principal("grant-0001", ttl_seconds=5, **SCOPE)
|
|
assert refreshed != expired
|
|
assert restarted.authorize(refreshed, **SCOPE) == signed_grant
|
|
|
|
|
|
def test_explicit_grant_expiry_caps_child_lease_and_cannot_be_refreshed(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
auth.register_grant(grant(owner_key, clock, expires_at=clock() + 7))
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
assert principal["payload"]["expires_at"] == clock() + 7
|
|
assert principal["operational_epoch"]["payload"]["expires_at"] == clock() + 7
|
|
clock.value += 7
|
|
with pytest.raises(AuthorizationError, match="expired"):
|
|
auth.authorize(principal, **SCOPE)
|
|
with pytest.raises(AuthorizationError, match="expired"):
|
|
auth.issue_principal("grant-0001", ttl_seconds=5, **SCOPE)
|
|
|
|
|
|
def test_higher_owner_signed_generation_reactivates_but_stale_proofs_stay_invalid(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
control = object()
|
|
auth = service(
|
|
tmp_path,
|
|
owner_key,
|
|
machine_key,
|
|
clock,
|
|
control_context_verifier=lambda value: value is control,
|
|
)
|
|
auth.register_grant(grant(owner_key, clock))
|
|
old = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
auth.revoke_grant("grant-0001", control_context=control)
|
|
|
|
new_grant = grant(owner_key, clock, generation=3)
|
|
auth.register_grant(new_grant)
|
|
fresh = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
with pytest.raises(AuthorizationError, match="generation"):
|
|
auth.authorize(old, **SCOPE)
|
|
assert auth.authorize(fresh, **SCOPE) == new_grant
|
|
|
|
|
|
def test_strict_schema_types_and_signature_encoding(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
|
|
malformed = grant(owner_key, clock)
|
|
malformed["payload"]["generation"] = True
|
|
with pytest.raises(AuthorizationError, match="generation"):
|
|
auth.register_grant(malformed)
|
|
|
|
malformed = grant(owner_key, clock)
|
|
malformed["signing"]["signature"] = base64.b64encode(b"short").decode("ascii")
|
|
with pytest.raises(AuthorizationError, match="signature"):
|
|
auth.register_grant(malformed)
|
|
|
|
with pytest.raises(AuthorizationError, match="ttl_seconds"):
|
|
auth.register_grant(grant(owner_key, clock))
|
|
auth.issue_principal("grant-0001", ttl_seconds=True, **SCOPE)
|
|
|
|
|
|
def test_grant_helper_requires_an_explicit_validity_mode():
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
arguments = {
|
|
"grant_id": "grant-0001",
|
|
"generation": 1,
|
|
"tenant_id": "tenant-a",
|
|
"subject_id": "owner-a",
|
|
"mission_ids": ("mission-a",),
|
|
"task_ids": ("task-a",),
|
|
"operations": ("deploy",),
|
|
"component_ids": ("component-a",),
|
|
"environments": ("staging",),
|
|
"issued_at": clock(),
|
|
}
|
|
with pytest.raises(TypeError):
|
|
sign_owner_delegation_grant(owner_key, "owner-key-1", **arguments)
|
|
with pytest.raises(AuthorizationError, match="expires_at"):
|
|
sign_owner_delegation_grant(
|
|
owner_key,
|
|
"owner-key-1",
|
|
validity_mode="EXPIRES_AT",
|
|
expires_at=None,
|
|
**arguments,
|
|
)
|
|
|
|
string_scope = dict(arguments)
|
|
string_scope["mission_ids"] = "abc"
|
|
with pytest.raises(AuthorizationError, match="mission_ids.*list or tuple"):
|
|
sign_owner_delegation_grant(
|
|
owner_key,
|
|
"owner-key-1",
|
|
validity_mode="UNTIL_REVOKED",
|
|
**string_scope,
|
|
)
|
|
|
|
|
|
def test_effect_budget_reserves_each_stable_effect_id_once_and_persists(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
auth.register_grant(grant(owner_key, clock, max_effects=1))
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
assert auth.reserve_effect("grant-0001", "effect-0001", principal, **SCOPE) is True
|
|
assert auth.reserve_effect("grant-0001", "effect-0001", principal, **SCOPE) is False
|
|
|
|
restarted = service(tmp_path, owner_key, machine_key, clock)
|
|
with pytest.raises(AuthorizationError, match="effect budget"):
|
|
restarted.reserve_effect("grant-0001", "effect-0002", principal, **SCOPE)
|
|
|
|
|
|
def test_unbounded_grant_leaves_mission_budget_to_caller(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
auth.register_grant(grant(owner_key, clock))
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
assert auth.reserve_effect("grant-0001", "effect-0001", principal, **SCOPE) is True
|
|
assert auth.reserve_effect("grant-0001", "effect-0002", principal, **SCOPE) is True
|
|
|
|
|
|
def test_old_machine_proof_can_be_verified_during_machine_key_rotation(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
old_machine_key = Ed25519PrivateKey.generate()
|
|
new_machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
original = service(tmp_path, owner_key, old_machine_key, clock)
|
|
signed_grant = grant(owner_key, clock)
|
|
original.register_grant(signed_grant)
|
|
old_principal = original.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
rotated = AuthorizationService(
|
|
tmp_path / "authorization.sqlite3",
|
|
{"owner-key-1": owner_key.public_key()},
|
|
new_machine_key,
|
|
"machine-key-2",
|
|
owner_key_scopes={
|
|
"owner-key-1": {
|
|
"tenant_ids": ["tenant-a"],
|
|
"subject_ids": ["owner-a"],
|
|
}
|
|
},
|
|
clock=clock,
|
|
trusted_machine_public_keys={
|
|
"machine-key-1": old_machine_key.public_key(),
|
|
"machine-key-2": new_machine_key.public_key(),
|
|
},
|
|
)
|
|
assert rotated.authorize(old_principal, **SCOPE) == signed_grant
|
|
|
|
|
|
def test_revocation_generation_cannot_reset_budget_or_revive_immutable_grant(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
control = object()
|
|
auth = service(
|
|
tmp_path,
|
|
owner_key,
|
|
machine_key,
|
|
clock,
|
|
control_context_verifier=lambda value: value is control,
|
|
)
|
|
signed_grant = grant(owner_key, clock, max_effects=1)
|
|
auth.register_grant(signed_grant)
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
assert auth.reserve_effect("grant-0001", "effect-0001", principal, **SCOPE) is True
|
|
|
|
assert auth.revoke_grant("grant-0001", control_context=control) == 2
|
|
with pytest.raises(AuthorizationError, match="stale"):
|
|
auth.register_grant(signed_grant)
|
|
with pytest.raises(AuthorizationError, match="revoked|generation"):
|
|
auth.reserve_effect("grant-0001", "effect-0002", principal, **SCOPE)
|
|
with pytest.raises(AuthorizationError, match="revoked"):
|
|
auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
|
|
def test_rejects_nan_empty_identity_and_untrusted_retired_machine_key(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
old_machine_key = Ed25519PrivateKey.generate()
|
|
new_machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
original = service(tmp_path, owner_key, old_machine_key, clock)
|
|
signed_grant = grant(owner_key, clock)
|
|
original.register_grant(signed_grant)
|
|
old_principal = original.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
malformed = grant(owner_key, clock)
|
|
malformed["payload"]["issued_at"] = float("nan")
|
|
with pytest.raises(AuthorizationError, match="issued_at"):
|
|
original.register_grant(malformed)
|
|
malformed = grant(owner_key, clock)
|
|
malformed["payload"]["tenant_id"] = ""
|
|
with pytest.raises(AuthorizationError, match="tenant_id"):
|
|
original.register_grant(malformed)
|
|
|
|
rotated_without_old_trust = AuthorizationService(
|
|
tmp_path / "authorization.sqlite3",
|
|
{"owner-key-1": owner_key.public_key()},
|
|
new_machine_key,
|
|
"machine-key-2",
|
|
owner_key_scopes={
|
|
"owner-key-1": {
|
|
"tenant_ids": ["tenant-a"],
|
|
"subject_ids": ["owner-a"],
|
|
}
|
|
},
|
|
clock=clock,
|
|
)
|
|
with pytest.raises(AuthorizationError, match="trusted machine key"):
|
|
rotated_without_old_trust.authorize(old_principal, **SCOPE)
|
|
|
|
|
|
def operation_definition(effect_class: str, *, reversible: bool = False) -> OperationDefinition:
|
|
return OperationDefinition.from_dict(
|
|
{
|
|
"operation_id": "deploy",
|
|
"version": 1,
|
|
"effect_class": effect_class,
|
|
"handler_ref": "trusted.deploy",
|
|
"input_schema": "aios.deploy.input.v1",
|
|
"required_permissions": ["deploy"],
|
|
"timeout_seconds": 10,
|
|
"readback": {"mode": "EXACT", "required": True},
|
|
"retry": {"mode": "AFTER_READBACK", "max_attempts": 1},
|
|
"reconcile": {"mode": "EXACT_READBACK"},
|
|
"rollback": {"mode": "OPERATION", "operation_id": "deploy.rollback"}
|
|
if reversible
|
|
else {"mode": "NONE"},
|
|
"owner_boundary": {"mode": "WITHIN_GRANT"},
|
|
}
|
|
)
|
|
|
|
|
|
def test_signed_destructive_ceiling_defaults_none_and_denies_mutating_definition(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
signed_grant = grant(owner_key, clock)
|
|
assert signed_grant["payload"]["destructive_ceiling"] == "NONE"
|
|
auth.register_grant(signed_grant)
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
assert auth.assert_effect_bounds(
|
|
principal, operation_definition("READ_ONLY"), **SCOPE
|
|
) == signed_grant
|
|
with pytest.raises(AuthorizationError, match="destructive ceiling"):
|
|
auth.assert_effect_bounds(
|
|
principal, operation_definition("IDEMPOTENT"), **SCOPE
|
|
)
|
|
|
|
|
|
def test_reversible_ceiling_requires_rollback_and_irreversible_ceiling_is_explicit(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
auth.register_grant(grant(owner_key, clock, destructive_ceiling="REVERSIBLE"))
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
|
|
with pytest.raises(AuthorizationError, match="rollback"):
|
|
auth.assert_effect_bounds(
|
|
principal, operation_definition("RECONCILABLE", reversible=False), **SCOPE
|
|
)
|
|
assert auth.assert_effect_bounds(
|
|
principal, operation_definition("RECONCILABLE", reversible=True), **SCOPE
|
|
)["payload"]["destructive_ceiling"] == "REVERSIBLE"
|
|
with pytest.raises(AuthorizationError, match="irreversible"):
|
|
auth.assert_effect_bounds(
|
|
principal, operation_definition("IRREVERSIBLE", reversible=False), **SCOPE
|
|
)
|
|
|
|
owner_key_2 = Ed25519PrivateKey.generate()
|
|
auth_2 = AuthorizationService(
|
|
tmp_path / "authorization-2.sqlite3",
|
|
{"owner-key-1": owner_key_2.public_key()},
|
|
machine_key,
|
|
"machine-key-1",
|
|
owner_key_scopes={
|
|
"owner-key-1": {
|
|
"tenant_ids": ["tenant-a"],
|
|
"subject_ids": ["owner-a"],
|
|
}
|
|
},
|
|
clock=clock,
|
|
)
|
|
auth_2.register_grant(grant(owner_key_2, clock, destructive_ceiling="IRREVERSIBLE"))
|
|
principal_2 = auth_2.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
assert auth_2.assert_effect_bounds(
|
|
principal_2, operation_definition("IRREVERSIBLE"), **SCOPE
|
|
)["payload"]["destructive_ceiling"] == "IRREVERSIBLE"
|
|
|
|
|
|
def test_destructive_ceiling_is_signed_and_strictly_enumerated(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
tampered = grant(owner_key, clock)
|
|
tampered["payload"]["destructive_ceiling"] = "IRREVERSIBLE"
|
|
with pytest.raises(AuthorizationError, match="signature"):
|
|
auth.register_grant(tampered)
|
|
with pytest.raises(AuthorizationError, match="destructive_ceiling"):
|
|
grant(owner_key, clock, destructive_ceiling="UNBOUNDED")
|
|
|
|
|
|
def test_owner_key_trust_is_explicitly_bound_to_tenant_and_subject(tmp_path):
|
|
owner_key = Ed25519PrivateKey.generate()
|
|
machine_key = Ed25519PrivateKey.generate()
|
|
clock = Clock()
|
|
|
|
with pytest.raises(TypeError, match="owner_key_scopes"):
|
|
AuthorizationService(
|
|
tmp_path / "missing-scopes.sqlite3",
|
|
{"owner-key-1": owner_key.public_key()},
|
|
machine_key,
|
|
"machine-key-1",
|
|
clock=clock,
|
|
)
|
|
|
|
auth = service(tmp_path, owner_key, machine_key, clock)
|
|
with pytest.raises(AuthorizationError, match="Owner key scope"):
|
|
auth.register_grant(grant(owner_key, clock, tenant_id="tenant-b"))
|
|
with pytest.raises(AuthorizationError, match="Owner key scope"):
|
|
auth.register_grant(grant(owner_key, clock, subject_id="owner-b"))
|
|
|
|
auth.register_grant(grant(owner_key, clock))
|
|
principal = auth.issue_principal("grant-0001", ttl_seconds=30, **SCOPE)
|
|
reopened_with_narrower_trust = AuthorizationService(
|
|
tmp_path / "authorization.sqlite3",
|
|
{"owner-key-1": owner_key.public_key()},
|
|
machine_key,
|
|
"machine-key-1",
|
|
owner_key_scopes={
|
|
"owner-key-1": {
|
|
"tenant_ids": ["tenant-a"],
|
|
"subject_ids": ["owner-b"],
|
|
}
|
|
},
|
|
clock=clock,
|
|
)
|
|
with pytest.raises(AuthorizationError, match="Owner key scope"):
|
|
reopened_with_narrower_trust.authorize(principal, **SCOPE)
|
|
|
|
globally_scoped = AuthorizationService(
|
|
tmp_path / "explicit-global.sqlite3",
|
|
{"owner-key-1": owner_key.public_key()},
|
|
machine_key,
|
|
"machine-key-1",
|
|
owner_key_scopes={
|
|
"owner-key-1": {"tenant_ids": ["*"], "subject_ids": ["*"]}
|
|
},
|
|
clock=clock,
|
|
)
|
|
assert globally_scoped.register_grant(
|
|
grant(owner_key, clock, tenant_id="tenant-b", subject_id="owner-b")
|
|
)["payload"]["tenant_id"] == "tenant-b"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"scopes",
|
|
[
|
|
{},
|
|
{"owner-key-1": {"tenant_ids": ["tenant-a"]}},
|
|
{
|
|
"owner-key-1": {
|
|
"tenant_ids": ["*", "tenant-a"],
|
|
"subject_ids": ["owner-a"],
|
|
}
|
|
},
|
|
{
|
|
"owner-key-1": {
|
|
"tenant_ids": ["tenant-a"],
|
|
"subject_ids": [],
|
|
}
|
|
},
|
|
],
|
|
)
|
|
def test_owner_key_scope_configuration_is_exact_and_fail_closed(tmp_path, scopes):
|
|
with pytest.raises(AuthorizationError, match="owner_key_scopes"):
|
|
AuthorizationService(
|
|
tmp_path / "authorization.sqlite3",
|
|
{"owner-key-1": Ed25519PrivateKey.generate().public_key()},
|
|
Ed25519PrivateKey.generate(),
|
|
"machine-key-1",
|
|
owner_key_scopes=scopes,
|
|
)
|