feat(contract): tool-less + credential-allowlist + env-scoping quarantine asserters (TDD) [skip-docs]

Module 9 of the build order — the differentiator. Where the other modules
detect and report, these are pure functions that ENFORCE an invariant and
raise: the write-time analogue of runtime least-privilege. They harden the
call a pipeline makes around its quarantined transform; the library makes no
model call itself.

Three asserts, matching the reusable-contract checklist (BRIEF §6 steps 3-4):
- assert_tool_less(request): the model request carries no tool surface —
  tools/functions/tool_choice/function_call/mcp_servers, populated. Covers
  both Anthropic and OpenAI request shapes; empty/None is genuinely tool-less.
- assert_credential_allowlist(env, allowed): the process env holds no
  credential beyond this stage's allowlist (subset = least-privilege; the
  enrichment stage sees only the model key, never the publish credential).
- scoped_env(env, allowed): the isolation primitive — strips off-allowlist
  credentials, keeps PATH etc.; the assert then passes by construction.

Generalized from claude-code-llm-wiki tools/wiki_ingest/enrich.py
assert_quarantine (a pipeline-specific gate) into framework-agnostic pieces.
Two deliberate safety choices inherited from the reference: credential
detection is name-based (env VALUES are never read — value-scanning is the
output module's job); and a raised ContractViolation names only the offending
KEY, never a value, so it is safe to route to an alert channel (minimal-alert,
BRIEF §6 step 8). CREDENTIAL_NAME_RE ported verbatim from the proven reference.

23 new tests; 161 green total.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HyRCQMocjZ6SmSQ6JidJ2k

src/llm_ingestion_guard/contract.py

@@ -0,0 +1,133 @@
+"""contract — the write-time quarantine asserters (BRIEF §5/§6, PLAN §93).
+
+The *differentiator*. Where the other modules detect and report, these functions
+enforce an invariant and **raise**: they are the write-time analogue of runtime
+least-privilege. They harden the call a pipeline makes around the quarantined,
+tool-less transform — the library itself makes no model call.
+
+Three asserts, matching the reusable-contract checklist (BRIEF §6, steps 3-4):
+
+* **(a) tool-less transform** — :func:`assert_tool_less` verifies the model
+  request carries zero tools/functions/MCP servers. A successful injection then
+  has nothing to act with (checklist step 3).
+* **(b) per-stage credential allowlist** — :func:`assert_credential_allowlist`
+  verifies the process env holds no credential beyond the ones this stage is
+  entitled to: the enrichment stage sees only the model key, never the publish
+  credential (checklist step 4).
+* **(c) capability isolation** — :func:`scoped_env` produces a correctly scoped
+  env so a hijacked stage cannot even read a credential it was never granted;
+  the assert then passes by construction.
+
+Reference: ``claude-code-llm-wiki`` ``tools/wiki_ingest/enrich.py``
+``assert_quarantine`` — a pipeline-specific quarantine gate, generalized here
+into framework-agnostic, reusable pieces.
+
+Two safety choices are deliberate and inherited from the reference:
+
+* Credential detection is **name-based**; env *values* are never read into the
+  assert. Value-scanning is :mod:`output`'s job for emitted text, not the env.
+* A raised :class:`ContractViolation` names only the offending *key* — never a
+  credential value — so the exception is safe to route to an alert channel
+  (minimal-alert payloads, BRIEF §6, step 8).
+"""
+from __future__ import annotations
+
+import re
+from typing import Iterable, Mapping
+
+
+class ContractViolation(Exception):
+    """A write-time security-contract invariant was violated.
+
+    Carries a machine-readable :attr:`code` and a :attr:`details` tuple of the
+    offending identifiers (tool-surface keys or env credential names) so a caller
+    can build a minimal, content-free alert payload. The message and details
+    never contain a credential value.
+    """
+
+    def __init__(self, message: str, *, code: str, details: tuple[str, ...] = ()):
+        super().__init__(message)
+        self.code = code
+        self.details = details
+
+
+# --- (a) tool-less transform -----------------------------------------------
+
+# Populated tool surface across Anthropic + OpenAI request shapes. An empty
+# list / None is genuinely tool-less; only a *populated* key is a violation.
+_TOOL_KEYS = ("tools", "functions", "tool_choice", "function_call", "mcp_servers")
+
+
+def assert_tool_less(request: Mapping[str, object]) -> None:
+    """Assert the model request carries no tool surface (checklist step 3).
+
+    Raises :class:`ContractViolation` (code ``"tool_present"``) if any of
+    ``tools``, ``functions``, ``tool_choice``, ``function_call`` or
+    ``mcp_servers`` is present *and populated*. ``request`` is the request
+    kwargs mapping — framework-agnostic, no SDK type required.
+    """
+    offending = tuple(sorted(k for k in _TOOL_KEYS if request.get(k)))
+    if offending:
+        raise ContractViolation(
+            "model request carries a tool surface: %s" % ", ".join(offending),
+            code="tool_present",
+            details=offending,
+        )
+
+
+# --- (b) per-stage credential allowlist ------------------------------------
+
+# Credential-shaped env-var names, matched on the underscore-delimited word
+# boundary so ``SECRETARY_NAME`` does not false-positive. Ported verbatim from
+# the reference pipeline's proven ``CREDENTIAL_NAME_RE``.
+CREDENTIAL_NAME_RE = re.compile(
+    r"(^|_)(API_?KEY|APIKEY|TOKEN|SECRET|PASSWORD|PASSWD|CREDENTIALS?)(_|$)"
+)
+
+
+def credential_env_names(env: Mapping[str, str]) -> tuple[str, ...]:
+    """Return the sorted env-var names that look like credentials.
+
+    Name-based only: values are never read. ``PATH``/``HOME``/``LANG`` and other
+    non-credential vars are ignored.
+    """
+    return tuple(sorted(
+        name for name in env if CREDENTIAL_NAME_RE.search(name.upper())
+    ))
+
+
+def assert_credential_allowlist(env: Mapping[str, str], allowed: Iterable[str]) -> None:
+    """Assert the env holds no credential beyond ``allowed`` (checklist step 4).
+
+    Least-privilege: the credential set must be a *subset* of ``allowed`` —
+    holding fewer than allowed is safe; holding even one more raises
+    :class:`ContractViolation` (code ``"credential_leak"``). The raised message
+    names only the offending keys, never their values.
+    """
+    allowlist = set(allowed)
+    leaked = tuple(
+        name for name in credential_env_names(env) if name not in allowlist
+    )
+    if leaked:
+        raise ContractViolation(
+            "env carries off-allowlist credential(s): %s" % ", ".join(leaked),
+            code="credential_leak",
+            details=leaked,
+        )
+
+
+# --- (c) capability isolation (env scoping) --------------------------------
+
+
+def scoped_env(env: Mapping[str, str], allowed: Iterable[str]) -> dict[str, str]:
+    """Return a copy of ``env`` with off-allowlist credentials removed.
+
+    Non-credential vars (``PATH`` etc.) pass through so the scoped env still
+    works for a subprocess; only credential-shaped keys outside ``allowed`` are
+    dropped. The input is not mutated. By construction,
+    ``assert_credential_allowlist(scoped_env(env, a), a)`` always passes — a
+    hijacked stage cannot read a credential it was never granted.
+    """
+    allowlist = set(allowed)
+    leaked = set(credential_env_names(env)) - allowlist
+    return {name: value for name, value in env.items() if name not in leaked}

tests/test_contract.py

@@ -0,0 +1,163 @@
+"""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")
+
+
+# --- (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