import json from datetime import datetime, timezone from pathlib import Path import pytest from audit_core.senders import SenderRegistry, WILDCARD ROOT = Path(__file__).resolve().parents[1] SCOPE_FILE = ROOT / "deploy" / "senders-scope.json" def test_declared_scope_keeps_user_engine_tenants_wildcard(): scope = json.loads(SCOPE_FILE.read_text()) user_engine = next(entry for entry in scope if entry["name"] == "user-engine") assert user_engine["tenants"] == ["*"] assert user_engine["sources"] == ["user-engine"] assert user_engine["may_write"] is True assert user_engine["may_read"] is False assert "tokens" not in user_engine assert "token" not in user_engine def test_scope_overlay_widens_narrow_secret_tenants(tmp_path): secret = json.dumps( [ { "name": "user-engine", "tokens": ["live-token"], "sources": ["user-engine"], "tenants": ["tenant:friendly:binky"], "may_write": True, "may_read": False, } ] ) env = { "AUDIT_CORE_SENDERS": secret, "AUDIT_CORE_SENDERS_SCOPE_PATH": str(SCOPE_FILE), } registry = SenderRegistry.from_env(env) identity = registry.authenticate("Bearer live-token") assert identity is not None assert WILDCARD in identity.tenants assert identity.permits_tenant("tenant:other:x") assert identity.tokens == ("live-token",) def test_scope_overlay_does_not_take_tokens_from_git(): overlay = json.dumps( [ { "name": "user-engine", "tokens": ["must-not-be-used"], "tenants": ["*"], } ] ) secret = json.dumps( [ { "name": "user-engine", "tokens": ["live-token"], "sources": ["user-engine"], "tenants": ["tenant:friendly:binky"], } ] ) registry = SenderRegistry.from_env( {"AUDIT_CORE_SENDERS": secret, "AUDIT_CORE_SENDERS_SCOPE": overlay} ) assert registry.authenticate("Bearer must-not-be-used") is None assert registry.authenticate("Bearer live-token") is not None def test_missing_overlay_leaves_secret_as_is(): secret = json.dumps( [ { "name": "user-engine", "tokens": ["live-token"], "sources": ["user-engine"], "tenants": ["tenant:friendly:binky"], } ] ) identity = SenderRegistry.from_env({"AUDIT_CORE_SENDERS": secret}).identities[0] assert identity.tenants == frozenset({"tenant:friendly:binky"}) def test_temporary_sender_is_rejected_at_and_after_expiry(): registry = SenderRegistry.from_env( { "AUDIT_CORE_SENDERS": json.dumps( [ { "name": "whitehat-a", "tokens": ["temporary"], "sources": ["whitehat-security"], "tenants": ["tenant:trial:whitehat-a"], "expires_at": "2026-08-22T18:15:00Z", } ] ) } ) before = datetime(2026, 8, 22, 18, 14, 59, tzinfo=timezone.utc) boundary = datetime(2026, 8, 22, 18, 15, 0, tzinfo=timezone.utc) assert registry.authenticate("Bearer temporary", now=before) is not None assert registry.authenticate("Bearer temporary", now=boundary) is None @pytest.mark.parametrize("expires_at", ["not-a-time", "2026-08-22T18:15:00"]) def test_temporary_sender_expiry_must_be_valid_and_timezone_aware(expires_at): sender = json.dumps( [ { "name": "whitehat-a", "tokens": ["temporary"], "sources": ["whitehat-security"], "expires_at": expires_at, } ] ) with pytest.raises(ValueError, match="expires_at"): SenderRegistry.from_env({"AUDIT_CORE_SENDERS": sender}) def test_scope_overlay_cannot_remove_secret_backed_expiry(): registry = SenderRegistry.from_env( { "AUDIT_CORE_SENDERS": json.dumps( [ { "name": "user-engine", "tokens": ["temporary"], "sources": ["user-engine"], "expires_at": "2026-08-22T18:15:00Z", } ] ), "AUDIT_CORE_SENDERS_SCOPE": json.dumps( [{"name": "user-engine", "tenants": ["*"]}] ), } ) assert registry.identities[0].expires_at == datetime( 2026, 8, 22, 18, 15, 0, tzinfo=timezone.utc ) def test_invalid_scope_overlay_is_a_startup_error(): secret = json.dumps( [{"name": "user-engine", "tokens": ["t"], "sources": ["user-engine"]}] ) with pytest.raises(ValueError, match="JSON list"): SenderRegistry.from_env( {"AUDIT_CORE_SENDERS": secret, "AUDIT_CORE_SENDERS_SCOPE": "{}"} )