audit-core/tests/test_senders.py

import json
from datetime import datetime, timezone
from pathlib import Path

import pytest

from audit_core.senders import SenderIdentity, 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": "{}"}
        )


# --- AUDIT-WP-0009-T03: §9.6 evidence kind ---------------------------------


def test_evidence_kind_defaults_to_attributive():
    """A source that has not declared must not be treated as load-bearing.

    The other default would have audit-core imply a completeness obligation no
    source ever accepted.
    """
    identity = SenderIdentity(
        name="quiet", tokens=("t",), sources=frozenset({"quiet"})
    )
    assert identity.evidence_kind == "attributive"
    assert identity.is_load_bearing is False


def test_load_bearing_is_an_explicit_declaration():
    identity = SenderIdentity(
        name="approval-engine",
        tokens=("t",),
        sources=frozenset({"approval-engine"}),
        evidence_kind="load-bearing",
    )
    assert identity.is_load_bearing is True


def test_unknown_evidence_kind_is_refused():
    with pytest.raises(ValueError, match="evidence_kind must be one of"):
        SenderIdentity(
            name="x", tokens=("t",), sources=frozenset({"x"}), evidence_kind="best-effort"
        )


def test_attributive_source_carries_its_declared_trade():
    identity = SenderIdentity(
        name="tenant-engine",
        tokens=("t",),
        sources=frozenset({"tenant-engine"}),
        completeness_trade="drain is non-blocking; emission is after commit",
    )
    declaration = identity.evidence_declaration()
    assert declaration["completeness_trade"].startswith("drain is non-blocking")
    # Neither kind licenses a completeness claim (§9.6).
    assert declaration["completeness_claimed"] is False


def test_load_bearing_source_may_not_declare_a_trade():
    """§9.6 requires atomicity of it, so there is no trade to record."""
    with pytest.raises(ValueError, match="may not declare a completeness_trade"):
        SenderIdentity(
            name="approval-engine",
            tokens=("t",),
            sources=frozenset({"approval-engine"}),
            evidence_kind="load-bearing",
            completeness_trade="emits after commit",
        )


def test_blank_trade_is_refused_rather_than_stored():
    with pytest.raises(ValueError, match="completeness_trade must say what"):
        SenderIdentity(
            name="x", tokens=("t",), sources=frozenset({"x"}), completeness_trade="   "
        )


def test_registration_schema_carries_evidence_kind():
    registry = SenderRegistry.from_env({
        "AUDIT_CORE_SENDERS": json.dumps([
            {
                "name": "approval-engine",
                "tokens": ["tok"],
                "sources": ["approval-engine"],
                "tenants": ["*"],
                "evidence_kind": "load-bearing",
            },
            {
                "name": "tenant-engine",
                "tokens": ["tok2"],
                "sources": ["tenant-engine"],
                "tenants": ["*"],
                "completeness_trade": "non-blocking drain",
            },
        ])
    })
    by_name = {i.name: i for i in registry.identities}
    assert by_name["approval-engine"].is_load_bearing is True
    assert by_name["tenant-engine"].is_load_bearing is False
    assert by_name["tenant-engine"].completeness_trade == "non-blocking drain"


def test_scope_overlay_may_raise_the_evidence_kind():
    registry = SenderRegistry.from_env({
        "AUDIT_CORE_SENDERS": json.dumps(
            [{"name": "approval-engine", "tokens": ["tok"], "sources": ["approval-engine"]}]
        ),
        "AUDIT_CORE_SENDERS_SCOPE": json.dumps(
            [{"name": "approval-engine", "evidence_kind": "load-bearing"}]
        ),
    })
    assert registry.identities[0].is_load_bearing is True


def test_scope_overlay_may_not_downgrade_a_load_bearing_source():
    """A ConfigMap refresh must not quietly drop atomicity obligations."""
    with pytest.raises(ValueError, match="may not downgrade"):
        SenderRegistry.from_env({
            "AUDIT_CORE_SENDERS": json.dumps([{
                "name": "approval-engine",
                "tokens": ["tok"],
                "sources": ["approval-engine"],
                "evidence_kind": "load-bearing",
            }]),
            "AUDIT_CORE_SENDERS_SCOPE": json.dumps(
                [{"name": "approval-engine", "evidence_kind": "attributive"}]
            ),
        })


# --- AUDIT-WP-0009-T09: approval-engine registration inputs ----------------

SCOPE_CONFIGMAP = ROOT / "deploy" / "senders-scope.yaml"


def test_declared_scope_and_configmap_stay_in_lockstep():
    """The header says to keep them in lockstep; nothing asserted it."""
    yaml = pytest.importorskip("yaml")
    embedded = yaml.safe_load(SCOPE_CONFIGMAP.read_text())["data"]["senders-scope.json"]
    assert json.loads(embedded) == json.loads(SCOPE_FILE.read_text())


def test_approval_engine_is_declared_load_bearing():
    scope = json.loads(SCOPE_FILE.read_text())
    entry = next(e for e in scope if e["name"] == "approval-engine")
    assert entry["evidence_kind"] == "load-bearing"
    assert entry["sources"] == ["approval-engine"]
    # A source does not gain a read surface by emitting.
    assert entry["may_read"] is False
    # §9.6 permits no completeness trade for a load-bearing source.
    assert "completeness_trade" not in entry
    assert "tokens" not in entry and "token" not in entry


def test_a_scope_entry_without_a_token_admits_nothing():
    """What makes it safe to land approval-engine's scope ahead of its credential."""
    registry = SenderRegistry.from_env({
        "AUDIT_CORE_SENDERS": json.dumps(
            [{"name": "user-engine", "tokens": ["live"], "sources": ["user-engine"]}]
        ),
        "AUDIT_CORE_SENDERS_SCOPE_PATH": str(SCOPE_FILE),
    })
    assert [i.name for i in registry.identities] == ["user-engine"]
    assert registry.authenticate("Bearer live").name == "user-engine"


@pytest.mark.parametrize("tenant", ["platform", "tenant:coulomb", "tenant:Platform", "tenant:platform ", "tenant:customer"])
def test_approval_scope_restricts_a_stale_wildcard_registration(tenant):
    registry = SenderRegistry.from_env({
        "AUDIT_CORE_SENDERS": json.dumps([{
            "name": "approval-engine", "tokens": ["fixture-only"],
            "sources": ["*"], "tenants": ["*"],
            "may_read": True, "secret_policy": "redact",
        }]),
        "AUDIT_CORE_SENDERS_SCOPE_PATH": str(SCOPE_FILE),
    })
    identity = registry.authenticate("Bearer fixture-only")
    assert identity.permits_tenant("tenant:platform")
    assert not identity.permits_tenant(tenant)
    assert identity.permits_source("approval-engine")
    assert not identity.permits_source("user-engine")
    assert identity.may_write and not identity.may_read
    assert identity.secret_policy == "redact"
    assert identity.is_load_bearing


def test_user_engine_evidence_kind_is_not_asserted_on_its_behalf():
    """Undeclared means attributive by default, not a claim audit-core made."""
    scope = json.loads(SCOPE_FILE.read_text())
    entry = next(e for e in scope if e["name"] == "user-engine")
    assert "evidence_kind" not in entry


# --- AUDIT-WP-0009-T11: informed-decision registration inputs --------------


def test_informed_decision_is_declared_load_bearing():
    scope = json.loads(SCOPE_FILE.read_text())
    entry = next(e for e in scope if e["name"] == "informed-decision")
    assert entry["evidence_kind"] == "load-bearing"
    assert entry["sources"] == ["informed-decision"]
    assert entry["tenants"] == ["tenant:platform"]
    assert entry["may_read"] is False
    # §9.6 permits no completeness trade for a load-bearing source.
    assert "completeness_trade" not in entry
    assert "tokens" not in entry and "token" not in entry


def test_informed_decision_scope_entry_admits_nothing_without_a_token():
    """Safe to land ahead of the credential, asserted rather than read."""
    registry = SenderRegistry.from_env({
        "AUDIT_CORE_SENDERS": json.dumps(
            [{"name": "user-engine", "tokens": ["live"], "sources": ["user-engine"]}]
        ),
        "AUDIT_CORE_SENDERS_SCOPE_PATH": str(SCOPE_FILE),
    })
    assert "informed-decision" not in [i.name for i in registry.identities]


@pytest.mark.parametrize(
    "tenant", ["platform", "tenant:coulomb", "tenant:Platform", "tenant:platform "]
)
def test_informed_decision_scope_restricts_a_stale_wildcard_registration(tenant):
    registry = SenderRegistry.from_env({
        "AUDIT_CORE_SENDERS": json.dumps([{
            "name": "informed-decision", "tokens": ["fixture-only"],
            "sources": ["*"], "tenants": ["*"],
            "may_read": True, "secret_policy": "redact",
        }]),
        "AUDIT_CORE_SENDERS_SCOPE_PATH": str(SCOPE_FILE),
    })
    identity = registry.authenticate("Bearer fixture-only")
    assert identity.permits_tenant("tenant:platform")
    assert not identity.permits_tenant(tenant)
    assert identity.permits_source("informed-decision")
    assert not identity.permits_source("approval-engine")
    assert identity.may_write and not identity.may_read
    assert identity.is_load_bearing


def test_a_load_bearing_source_cannot_be_given_a_completeness_trade():
    """informed-decision declares no trade and may not acquire one by overlay."""
    with pytest.raises(ValueError, match="completeness_trade"):
        SenderIdentity(
            name="informed-decision",
            tokens=("t",),
            sources=frozenset({"informed-decision"}),
            evidence_kind="load-bearing",
            completeness_trade="emits after commit",
        )