FabledScribe/tests/test_retrieval_query_scrubbing.py

"""A logged query never carries a credential (#3925).

WHY THIS EXISTS

`pre_tool_rule` retrieves against the RAW COMMAND TEXT and `write_path_rule`
against the code being written, so whatever was on the command line or in the
buffer is what `record_retrieval` stores in `retrieval_logs.query`. A command
that exported a token therefore stored the token.

And storing it was not the worst of it. `near_miss_samples` is the readout the
threshold documentation tells you to open before moving a bar, so the value
came back OUT into an agent's context on the next tuning pass — which is
exactly how this was found, during #3853's threshold spike.

WHAT THIS PINS, IN BOTH DIRECTIONS, AND WHY THE SECOND HALF IS THE HARD ONE

A scrubber has two ways to fail and only one of them is obvious.

  1. It misses a secret. Caught by the redaction cases below.
  2. It eats the EVIDENCE. This is the failure that would do more damage,
     because it is silent: the whole worth of a near-miss sample is reading the
     query that was actually refused, and a scrubber that chewed up ordinary
     commands would turn the one instrument for tuning a bar into unreadable
     stubs while still looking like it worked. That is the #2663 shape — a
     surface that reads fine and has quietly stopped saying anything.

So the second block is not padding. Its cases are REAL queries taken from this
install's `near_miss_samples` during #3853, and they must survive byte for
byte. If a future pattern is added and one of them changes, the pattern is too
greedy — tighten it rather than editing the expectation.

The secret cases use FABRICATED values in the real formats. Nothing here is or
was a live credential.
"""
import pytest

from scribe.services.retrieval_telemetry import scrub_secrets

# Fabricated, in the shapes that actually occur. The first is the shape that
# was found stored: a shell assignment of a vendor-prefixed token.
_SECRETS = [
    ("vendor-prefixed token in a shell assignment",
     "TOK=flt_AAAABBBBCCCCDDDDEEEEFFFF\npython3 - <<'PY'",
     "flt_AAAABBBBCCCCDDDDEEEEFFFF"),
    ("a Scribe fmcp_ key in an auth header",
     "curl -H 'Authorization: Bearer fmcp_ZZZZYYYYXXXXWWWWVVVV' https://x",
     "fmcp_ZZZZYYYYXXXXWWWWVVVV"),
    ("a forge token in an export",
     "export GITHUB_TOKEN=ghp_1234567890abcdefghijABCDEF",
     "ghp_1234567890abcdefghijABCDEF"),
    ("a value assigned to a secret-named variable",
     'REGISTRY_PASSWORD="hunter2-correct-horse"',
     "hunter2-correct-horse"),
    ("an api_key in a query string",
     "curl 'https://api.example/v1/things?api_key=abcdef1234567890'",
     "abcdef1234567890"),
    ("a private key block",
     "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKC\n-----END RSA PRIVATE KEY-----",
     "MIIEowIBAAKC"),
]

# Real queries, from this install's near-miss samples during #3853.
_EVIDENCE = [
    "git push origin dev",
    'git pull --rebase origin dev 2>&1 | tail -3; echo "=== HEAD ==="; git log --oneline -2',
    "python3 - <<'PY'\nimport pathlib\np = pathlib.Path(\"web/src/routes/admin/tuning/tuning.test.ts\")",
    "docker compose up -d",
    'package library\n\nimport (\n\t"context"\n\t"fmt"\n)',
    "import {\n  fetchTransfers,\n  retryTransfer,\n} from './api'",
    'grep -rn "useState" src/components/ | head -20',
    # The word "token" in ordinary prose is not a token.
    "explain how the token bucket rate limiter works",
    "wc -l src/*.py && date",
]


@pytest.mark.parametrize(("label", "text", "secret"), _SECRETS,
                         ids=[c[0] for c in _SECRETS])
def test_a_credential_never_survives_into_the_query_column(label, text, secret):
    """The value goes; something visible stays in its place."""
    out = scrub_secrets(text)
    assert secret not in out, (
        f"{label}: the credential is still in the text that would be stored"
    )
    assert "[redacted" in out, (
        f"{label}: the span was removed without saying so. A silent deletion "
        "leaves a reader unable to tell a scrubbed query from a short one, "
        "which is the readout lying about itself rather than protecting you."
    )


@pytest.mark.parametrize("query", _EVIDENCE)
def test_an_ordinary_query_is_stored_exactly_as_it_was(query):
    """Evidence survives byte for byte.

    These came out of real `near_miss_samples`. A threshold is tuned by reading
    them, so a pattern greedy enough to touch one has destroyed the instrument
    it was meant to make safe — tighten the pattern, never this expectation.
    """
    assert scrub_secrets(query) == query


def test_empty_and_missing_queries_pass_through():
    """Some sources log no query at all; scrubbing must not invent one."""
    assert scrub_secrets(None) is None
    assert scrub_secrets("") == ""


def test_the_write_path_scrubs_rather_than_the_read_path():
    """The payload built for storage carries the redacted text (#3925).

    Pinned on `_build_payload` because that is the single seam every source
    reaches the column through. A per-caller scrub would be three places for
    one of them to be forgotten by whoever adds the fourth arm — and the one
    forgotten would be the one that stored a secret.
    """
    from scribe.services.retrieval_telemetry import _build_payload

    payload = _build_payload(
        user_id=1, source="pre_tool_rule",
        query="export API_TOKEN=ghp_1234567890abcdefghijABCDEF && git push",
        threshold=0.68, limit=5, project_id=0, is_task=None,
        results=[], duration_ms=1.0,
    )
    assert "ghp_1234567890abcdefghijABCDEF" not in payload["query"]
    assert "[redacted" in payload["query"]
    # The rest of the command survives, or the row stops being evidence.
    assert "git push" in payload["query"]