llm-ingestion-pipeline-security/tests/test_contract.py

"""Tests for contract — the write-time quarantine asserters (BRIEF §5/§6, PLAN §93).

These are the differentiator: pure functions that RAISE (unlike detectors, which
return a Report). They are the write-time analogue of runtime least-privilege —
they harden the call the pipeline makes; they make no model call themselves.

Reference: claude-code-llm-wiki tools/wiki_ingest/enrich.py assert_quarantine —
generalized here into framework-agnostic, reusable pieces.
"""
from __future__ import annotations

import pytest

from llm_ingestion_guard.contract import (
    ContractViolation,
    assert_tool_less,
    assert_credential_allowlist,
    credential_env_names,
    scoped_env,
)


# --- ContractViolation shape -----------------------------------------------

def test_contract_violation_is_exception_with_code_and_details():
    exc = ContractViolation("boom", code="tool_present", details=("tools",))
    assert isinstance(exc, Exception)
    assert exc.code == "tool_present"
    assert exc.details == ("tools",)
    assert "boom" in str(exc)


# --- (a) tool-less assert ---------------------------------------------------

def test_tool_less_passes_when_no_tool_keys():
    # a bare messages request with no tool surface is tool-less.
    assert assert_tool_less({"model": "claude-opus-4-8", "messages": []}) is None


@pytest.mark.parametrize("request_kwargs", [
    {},
    {"tools": []},
    {"tools": None},
    {"tool_choice": None},
    {"functions": []},
    {"mcp_servers": []},
])
def test_tool_less_passes_on_absent_or_empty_tool_surface(request_kwargs):
    # empty/None tool surface is genuinely tool-less; only a *populated* one raises.
    assert assert_tool_less(request_kwargs) is None


@pytest.mark.parametrize("request_kwargs,offending", [
    ({"tools": [{"name": "run_shell"}]}, "tools"),
    ({"functions": [{"name": "exec"}]}, "functions"),          # OpenAI legacy shape
    ({"function_call": {"name": "exec"}}, "function_call"),      # OpenAI legacy shape
    ({"tool_choice": {"type": "tool", "name": "x"}}, "tool_choice"),
    ({"mcp_servers": [{"url": "https://x"}]}, "mcp_servers"),    # Anthropic MCP connector
])
def test_tool_less_raises_on_populated_tool_surface(request_kwargs, offending):
    with pytest.raises(ContractViolation) as excinfo:
        assert_tool_less(request_kwargs)
    exc = excinfo.value
    assert exc.code == "tool_present"
    assert offending in exc.details
    assert offending in str(exc)


def test_tool_less_reports_all_offending_keys_sorted():
    with pytest.raises(ContractViolation) as excinfo:
        assert_tool_less({"tools": [{"name": "a"}], "mcp_servers": [{"url": "u"}]})
    assert excinfo.value.details == ("mcp_servers", "tools")


# --- (b) per-stage credential allowlist ------------------------------------

def test_credential_env_names_detects_credential_shaped_keys():
    env = {
        "ANTHROPIC_API_KEY": "x",
        "FORGEJO_PUBLISH_TOKEN": "x",
        "DB_PASSWORD": "x",
        "MY_SECRET_VALUE": "x",
        "PATH": "/usr/bin",
        "HOME": "/home/u",
        "LANG": "en_US.UTF-8",
        "SECRETARY_NAME": "x",     # not a credential — must not false-positive
    }
    assert credential_env_names(env) == (
        "ANTHROPIC_API_KEY",
        "DB_PASSWORD",
        "FORGEJO_PUBLISH_TOKEN",
        "MY_SECRET_VALUE",
    )


def test_credential_allowlist_passes_when_only_allowlisted_credential_present():
    env = {"ANTHROPIC_API_KEY": "sk-ant-xxx", "PATH": "/usr/bin", "HOME": "/home/u"}
    assert assert_credential_allowlist(env, ("ANTHROPIC_API_KEY",)) is None


def test_credential_allowlist_passes_on_subset_holds_fewer_than_allowed():
    # least-privilege is "holds no MORE than the allowlist"; holding fewer is safe.
    env = {"PATH": "/usr/bin"}
    assert assert_credential_allowlist(env, ("ANTHROPIC_API_KEY", "OTHER_KEY")) is None


def test_credential_allowlist_raises_on_off_allowlist_credential():
    env = {"ANTHROPIC_API_KEY": "x", "FORGEJO_PUBLISH_TOKEN": "x"}
    with pytest.raises(ContractViolation) as excinfo:
        assert_credential_allowlist(env, ("ANTHROPIC_API_KEY",))
    exc = excinfo.value
    assert exc.code == "credential_leak"
    assert exc.details == ("FORGEJO_PUBLISH_TOKEN",)
    assert "FORGEJO_PUBLISH_TOKEN" in str(exc)


def test_credential_allowlist_never_leaks_the_secret_value_in_the_message():
    # minimal-alert (BRIEF §6.8): the value must never appear in the raised message.
    secret = "ghp_" + "a" * 36
    env = {"PUBLISH_TOKEN": secret}
    with pytest.raises(ContractViolation) as excinfo:
        assert_credential_allowlist(env, ())
    assert secret not in str(excinfo.value)
    assert secret not in "".join(excinfo.value.details)


def test_credential_allowlist_reports_multiple_leaks_sorted():
    env = {"AWS_SECRET_ACCESS_KEY": "x", "GITHUB_TOKEN": "x", "ANTHROPIC_API_KEY": "x"}
    with pytest.raises(ContractViolation) as excinfo:
        assert_credential_allowlist(env, ("ANTHROPIC_API_KEY",))
    assert excinfo.value.details == ("AWS_SECRET_ACCESS_KEY", "GITHUB_TOKEN")


def test_credential_allowlist_catches_bare_provider_key():
    # M2: a bare `<PROVIDER>_KEY` (e.g. STRIPE_KEY) is credential-shaped and must
    # be caught off-allowlist. A silent miss would fail *open* — the key leaks
    # into a stage that was never granted it. Fail-loud > fail-silent here.
    env = {"ANTHROPIC_API_KEY": "x", "STRIPE_KEY": "x"}
    with pytest.raises(ContractViolation) as excinfo:
        assert_credential_allowlist(env, ("ANTHROPIC_API_KEY",))
    assert excinfo.value.details == ("STRIPE_KEY",)
    # scoped_env then drops it, so the assert passes by construction.
    scoped = scoped_env(env, ("ANTHROPIC_API_KEY",))
    assert "STRIPE_KEY" not in scoped
    assert assert_credential_allowlist(scoped, ("ANTHROPIC_API_KEY",)) is None


def test_bare_key_rule_is_broad_by_design_dynamodb_keys_are_loud_fps():
    # Deliberate tradeoff (M2): the bare `_KEY` rule also flags non-secret keys
    # like DynamoDB's PARTITION_KEY / SORT_KEY. This is a LOUD false positive
    # (raises, one-line allowlist fix) chosen over a silent fail-open on some new
    # provider's `<X>_KEY`. Fail-loud > fail-silent for an isolation control.
    assert credential_env_names({"PARTITION_KEY": "pk", "SORT_KEY": "sk"}) == (
        "PARTITION_KEY", "SORT_KEY",
    )
    env = {"PARTITION_KEY": "pk", "SORT_KEY": "sk"}
    assert assert_credential_allowlist(env, ("PARTITION_KEY", "SORT_KEY")) is None


# --- (c) capability isolation (env scoping) --------------------------------

def test_scoped_env_strips_off_allowlist_credentials_keeps_the_rest():
    env = {
        "ANTHROPIC_API_KEY": "keep",
        "FORGEJO_PUBLISH_TOKEN": "drop",
        "PATH": "/usr/bin",
        "HOME": "/home/u",
    }
    scoped = scoped_env(env, ("ANTHROPIC_API_KEY",))
    assert scoped == {"ANTHROPIC_API_KEY": "keep", "PATH": "/usr/bin", "HOME": "/home/u"}


def test_scoped_env_does_not_mutate_input():
    env = {"ANTHROPIC_API_KEY": "x", "FORGEJO_PUBLISH_TOKEN": "y"}
    original = dict(env)
    scoped_env(env, ("ANTHROPIC_API_KEY",))
    assert env == original


def test_scoped_env_output_always_satisfies_the_allowlist_assert():
    # the isolation guarantee: scoping then asserting always passes, by construction.
    env = {
        "ANTHROPIC_API_KEY": "x",
        "FORGEJO_PUBLISH_TOKEN": "x",
        "AWS_SECRET_ACCESS_KEY": "x",
        "PATH": "/usr/bin",
    }
    allowed = ("ANTHROPIC_API_KEY",)
    assert assert_credential_allowlist(scoped_env(env, allowed), allowed) is None