popoto.privacy.never_record

popoto.privacy.never_record

Never-record firewall — deterministic pre-storage privacy gate (#561).

What this guarantees

Content matching the enumerated guaranteed class below never reaches Redis: it is dropped in :meth:popoto.models.base.Model.save before the write filter, before pre_save(), and therefore before serialization, HSET, index writes, BM25 tokenization, embedding calls, and co-occurrence edges. Nothing about the dropped text is persisted — only a content-free tombstone carrying a random id and a reason code.

The guaranteed class:

off_the_record An explicit human marker. Voids the entire turn, not a guessed span. private_key_block PEM / OpenSSH / PuTTY private key headers. credential_prefix Vendor-prefixed API tokens (Anthropic, OpenAI, GitHub, AWS, Google, Slack, GitLab, npm, HuggingFace, DigitalOcean, SendGrid, Stripe). jwt Three-segment base64url JSON Web Tokens. credential_assignment password=/api_key:/token= and friends followed by a value. url_userinfo scheme://user:password@host. payment_card 13-19 digit runs that pass the Luhn checksum. government_id US SSN-shaped strings with a valid-area guard. high_entropy The backstop for unknown-prefix tokens: long, credential-charset, high-Shannon-entropy tokens.

What this does NOT guarantee

This is a deterministic pattern gate, not an oracle. Read the holes:

• Over-blocking is accepted and expected. Base64 blobs and long random identifiers in ordinary prose will be dropped. That is the price of the zero-false-negative goal on the class above.
• Canonical git SHAs and UUIDs are excluded from entropy scoring (:data:_STRUCTURAL_EXCLUSIONS). Developer memory content mentions commit SHAs constantly, and dropping every one of them would make the firewall worse than useless. The cost is a real, enumerated hole: a bare 40-hex-char secret with no vendor prefix is not caught by the entropy backstop. It is still caught if it appears as password=<sha> or similar.
• Digit-free tokens, and tokens with no long separator-free run, are excluded from entropy scoring (:func:_entropy_candidate). Entropy per character does not separate ExtractionProviderRegistry (3.87 bits) or text-embedding-3-small from a real secret, so scoring whole tokens voided 11.6% of this repo's own documentation paragraphs.

The cost is large and should not be read as a footnote. A secret escapes the backstop if it contains no digit, or if a separator breaks it so no unbroken run reaches NR_ENTROPY_MIN_TOKEN_LEN. The separator case dominates, because base64's own alphabet includes - and _. Measured over 200k random base64url tokens per length: ~49% escape at 20 characters, ~27% at 32, ~12% at 43 (compare the digit-free case alone at ~3% and ~0.4%). Treat high_entropy as opportunistic, not as a second guarantee. Only high_entropy is affected; every named format has its own detector, and that is where the guarantee lives. - A novel credential format with no known prefix, low entropy, and no assignment context can pass. The detector corpus is explicit and lives in one module so it can be extended; issue #561's module M9 (seeded audit) is the ongoing measurement of false negatives. - Semantically sensitive content is out of scope. The gate matches shape, not meaning. An LLM voter may be added later and may only ADD drops — the guarantee above rests solely on this deterministic core.

Why patterns are module constants, not Defaults entries

The numeric thresholds are tuning dials and live in :class:popoto.fields.constants.Defaults. The pattern corpus does not: it is a security-relevant list, and exposing it as a mutable registry would let a caller silently weaken the guarantee.

No Redis modules are used — tombstones are a plain HASH and a capped LIST, so the firewall works identically on Redis and Valkey.

NeverRecordVerdict dataclass

Result of a never-record scan.

Attributes:

Name	Type	Description
blocked	bool	True if the content must never be stored.
reason	Optional[str]	Stable reason code from :data:NEVER_RECORD_REASONS, or None.
detector	Optional[str]	Name of the specific detector that fired, or None.

Deliberately carries no matched text, no offset, and no length. This object's repr reaches log files, and any of those three would be a content side channel that defeats the whole module.

Source code in src/popoto/privacy/never_record.py

@dataclass(frozen=True)
class NeverRecordVerdict:
    """Result of a never-record scan.

    Attributes:
        blocked: True if the content must never be stored.
        reason: Stable reason code from :data:`NEVER_RECORD_REASONS`, or None.
        detector: Name of the specific detector that fired, or None.

    Deliberately carries no matched text, no offset, and no length. This
    object's repr reaches log files, and any of those three would be a
    content side channel that defeats the whole module.
    """

    blocked: bool
    reason: Optional[str] = None
    detector: Optional[str] = None

    def __bool__(self) -> bool:
        return self.blocked

NeverRecordMixin

Model mixin enforcing the never-record firewall on save().

Deliberately not a :class:~popoto.fields.write_filter.WriteFilterMixin subclass. The write filter is a scalar salience score with one compute_filter_score() slot per class, already claimed by salience scoring in the documented pattern; a firewall is a boolean content predicate with a tombstone side effect. Different shape, different slot.

Usage::

class Memory(NeverRecordMixin, Model):
    content = StringField()

Memory(content="my key is sk-ant-api03-AAAA...").save()  # -> False

Memory.never_record_counts()   # {"credential_prefix": 1}

:class:~popoto.recipes.default_memory.DefaultMemory carries this mixin already, so the batteries-included path and the Claude Code / Codex harness are gated with no adopter code change.

Key fields are excluded from the scan. AutoKeyField/KeyField values are generated UUID/hex shapes -- exactly what the entropy detector targets -- so scanning them would let a model block on its own identity. A key is never user content.

Source code in src/popoto/privacy/never_record.py

class NeverRecordMixin:
    """Model mixin enforcing the never-record firewall on ``save()``.

    Deliberately **not** a :class:`~popoto.fields.write_filter.WriteFilterMixin`
    subclass. The write filter is a scalar salience score with one
    ``compute_filter_score()`` slot per class, already claimed by salience
    scoring in the documented pattern; a firewall is a boolean content
    predicate with a tombstone side effect. Different shape, different slot.

    Usage::

        class Memory(NeverRecordMixin, Model):
            content = StringField()

        Memory(content="my key is sk-ant-api03-AAAA...").save()  # -> False

        Memory.never_record_counts()   # {"credential_prefix": 1}

    :class:`~popoto.recipes.default_memory.DefaultMemory` carries this mixin
    already, so the batteries-included path and the Claude Code / Codex
    harness are gated with no adopter code change.

    Key fields are excluded from the scan. ``AutoKeyField``/``KeyField``
    values are generated UUID/hex shapes -- exactly what the entropy detector
    targets -- so scanning them would let a model block on its own identity.
    A key is never user content.
    """

    # Export/import: the $NR: keys are audit telemetry about writes that were
    # refused, not state belonging to any record. Nothing to carry across a
    # round trip. Matches WriteFilterMixin's precedent.
    roundtrip_policy: str = "rebuild"

    #: Verdict from the most recent _check_never_record() on this instance:
    #: a blocking NeverRecordVerdict, or None when the scan came back clean.
    #: Content-free by construction, like every other verdict.
    _never_record_verdict: Optional[NeverRecordVerdict] = None

    if TYPE_CHECKING:
        # Supplied by Model, which every user of this mixin also
        # inherits from. Declared for the type checker only.
        _meta: Any

    def _never_record_scan_field_names(self) -> Iterator[str]:
        """Yield the field names whose values are exposed to the scan.

        Excludes key fields (see the class docstring). Mirrors the key-field
        iteration the ``KeyMutationError`` guard uses in ``Model.save()``.

        Split out from :meth:`_never_record_scan_values` so a subclass can
        narrow the surface by *name* without re-deriving the key-field rule --
        see ``JournalEntry._never_record_scan_values``, which drops the
        machine-generated ``target`` pointer.
        """
        meta = self._meta
        key_names = set(meta.key_field_names) | set(meta.auto_field_names)
        for field_name in meta.field_names:
            if field_name in key_names:
                continue
            yield field_name

    def _never_record_scan_values(self) -> Iterator[str]:
        """Yield the string field values this instance exposes to the scan.

        Excludes key fields (see the class docstring) and any non-string
        value.
        """
        for field_name in self._never_record_scan_field_names():
            value = getattr(self, field_name, None)
            if isinstance(value, str) and value.strip():
                yield value

    def _check_never_record(self) -> NeverRecordVerdict:
        """Evaluate the firewall and gate the save.

        Each field is scanned independently under both renderings. There is
        deliberately no cross-field concatenated pass: with whitespace
        removed, any join separator either vanishes -- letting two innocuous
        adjacent values merge into a shape neither contains -- or has to be a
        normalization-surviving sentinel, which is per-field scanning with
        extra steps. Issue #561 names only intra-value splitting as the
        adversarial case.

        Returns:
            NeverRecordVerdict: A clean verdict when the save may proceed.

        Raises:
            NeverRecordException: If any scanned value is blocked. The
                message carries only the reason code and detector name --
                never the matched text, an offset, or a length. That is
                load-bearing: ``MemoryService._record_failure`` writes
                ``f"{type(exc).__name__}: {exc}"`` into a plaintext log file,
                so a message quoting the match would defeat this module
                through a side channel.
        """
        for value in self._never_record_scan_values():
            verdict = scan_never_record(value)
            if verdict.blocked:
                # Recorded on the instance so a caller that only sees
                # save() -> False can tell a privacy drop from a rejected
                # write without re-scanning the content. The verdict is
                # content-free, so holding it costs nothing.
                self._never_record_verdict = verdict
                write_tombstone(type(self).__name__, verdict, model_class=type(self))
                raise NeverRecordException(
                    f"never-record: {verdict.reason} ({verdict.detector})"
                )
        self._never_record_verdict = None
        return _CLEAN

    @classmethod
    def never_record_counts(cls, redis_client: Any = None) -> dict[str, int]:
        """Return ``{reason_code: count}`` of drops for this model.

        Content-free by construction -- this is the auditable half of the
        guarantee.

        Args:
            redis_client: Optional client override, for tests.

        Returns:
            dict: Reason code to integer count. Empty if nothing dropped.
        """
        backend = _audit_backend(cls) if redis_client is None else None
        if backend is not None:
            return dict(backend.field_call(cls._meta.spec, "_never_record", "counts"))
        if redis_client is None:
            from ..redis_db import POPOTO_REDIS_DB

            redis_client = POPOTO_REDIS_DB
        raw = redis_client.hgetall(never_record_key(cls.__name__, "counts")) or {}
        counts: dict[str, int] = {}
        for key, value in raw.items():
            if isinstance(key, bytes):
                key = key.decode("utf-8")
            counts[key] = int(value)
        return counts

    @classmethod
    def never_record_log(
        cls, limit: int = 100, redis_client: Any = None
    ) -> list[dict[str, Any]]:
        """Return the most recent drop tombstones, newest first.

        Args:
            limit: Maximum entries to return.
            redis_client: Optional client override, for tests.

        Returns:
            list: Dicts with ``id``, ``reason``, ``detector``, ``at``. No
            entry contains any fragment of the dropped content.
        """
        backend = _audit_backend(cls) if redis_client is None else None
        if backend is not None:
            raw: Any = backend.field_call(cls._meta.spec, "_never_record", "log", limit)
        else:
            if redis_client is None:
                from ..redis_db import POPOTO_REDIS_DB

                redis_client = POPOTO_REDIS_DB
            raw = redis_client.lrange(
                never_record_key(cls.__name__, "drops"), 0, limit - 1
            )
        entries: list[dict[str, Any]] = []
        for item in raw or []:
            if isinstance(item, bytes):
                item = item.decode("utf-8")
            try:
                entries.append(json.loads(item))
            except (TypeError, ValueError):  # pragma: no cover - corrupt entry
                continue
        return entries

never_record_counts(redis_client=None) classmethod

Return {reason_code: count} of drops for this model.

Content-free by construction -- this is the auditable half of the guarantee.

Parameters:

Name	Type	Description	Default
redis_client	Any	Optional client override, for tests.	None

Returns:

Name	Type	Description
dict	dict[str, int]	Reason code to integer count. Empty if nothing dropped.

Source code in src/popoto/privacy/never_record.py

@classmethod
def never_record_counts(cls, redis_client: Any = None) -> dict[str, int]:
    """Return ``{reason_code: count}`` of drops for this model.

    Content-free by construction -- this is the auditable half of the
    guarantee.

    Args:
        redis_client: Optional client override, for tests.

    Returns:
        dict: Reason code to integer count. Empty if nothing dropped.
    """
    backend = _audit_backend(cls) if redis_client is None else None
    if backend is not None:
        return dict(backend.field_call(cls._meta.spec, "_never_record", "counts"))
    if redis_client is None:
        from ..redis_db import POPOTO_REDIS_DB

        redis_client = POPOTO_REDIS_DB
    raw = redis_client.hgetall(never_record_key(cls.__name__, "counts")) or {}
    counts: dict[str, int] = {}
    for key, value in raw.items():
        if isinstance(key, bytes):
            key = key.decode("utf-8")
        counts[key] = int(value)
    return counts

never_record_log(limit=100, redis_client=None) classmethod

Return the most recent drop tombstones, newest first.

Parameters:

Name	Type	Description	Default
limit	int	Maximum entries to return.	100
redis_client	Any	Optional client override, for tests.	None

Returns:

Name	Type	Description
list	list[dict[str, Any]]	Dicts with id, reason, detector, at. No
	list[dict[str, Any]]	entry contains any fragment of the dropped content.

Source code in src/popoto/privacy/never_record.py

@classmethod
def never_record_log(
    cls, limit: int = 100, redis_client: Any = None
) -> list[dict[str, Any]]:
    """Return the most recent drop tombstones, newest first.

    Args:
        limit: Maximum entries to return.
        redis_client: Optional client override, for tests.

    Returns:
        list: Dicts with ``id``, ``reason``, ``detector``, ``at``. No
        entry contains any fragment of the dropped content.
    """
    backend = _audit_backend(cls) if redis_client is None else None
    if backend is not None:
        raw: Any = backend.field_call(cls._meta.spec, "_never_record", "log", limit)
    else:
        if redis_client is None:
            from ..redis_db import POPOTO_REDIS_DB

            redis_client = POPOTO_REDIS_DB
        raw = redis_client.lrange(
            never_record_key(cls.__name__, "drops"), 0, limit - 1
        )
    entries: list[dict[str, Any]] = []
    for item in raw or []:
        if isinstance(item, bytes):
            item = item.decode("utf-8")
        try:
            entries.append(json.loads(item))
        except (TypeError, ValueError):  # pragma: no cover - corrupt entry
            continue
    return entries

scan_never_record(text)

Scan text for content that must never be recorded.

Pure and deterministic: no Redis, no model, no network, no clock. Safe to call from an extraction pipeline before an LLM sees the candidate.

Every regex detector marked whitespace-insensitive runs against two renderings — the raw text and a de-whitespaced rendering — so a credential split across whitespace or newlines is still caught. That is the adversarial case named in issue #561's acceptance criteria.

Parameters:

Name	Type	Description	Default
text	Any	The candidate content. Non-str input returns a clean verdict; callers pass field values of arbitrary type.	required

Returns:

Type	Description
NeverRecordVerdict	NeverRecordVerdict. Falsy when the content may be stored. Carries a
NeverRecordVerdict	reason code and detector name but never any fragment of the input.

Source code in src/popoto/privacy/never_record.py

def scan_never_record(text: Any) -> NeverRecordVerdict:
    """Scan ``text`` for content that must never be recorded.

    Pure and deterministic: no Redis, no model, no network, no clock. Safe to
    call from an extraction pipeline before an LLM sees the candidate.

    Every regex detector marked whitespace-insensitive runs against two
    renderings — the raw text and a de-whitespaced rendering — so a
    credential split across whitespace or newlines is still caught. That is
    the adversarial case named in issue #561's acceptance criteria.

    Args:
        text: The candidate content. Non-str input returns a clean verdict;
            callers pass field values of arbitrary type.

    Returns:
        NeverRecordVerdict. Falsy when the content may be stored. Carries a
        reason code and detector name but never any fragment of the input.
    """
    if not isinstance(text, str) or not text.strip():
        return _CLEAN

    stripped = _strip_whitespace(text)

    for reason, detector, pattern, whitespace_insensitive in _REGEX_DETECTORS:
        if pattern.search(text):
            return NeverRecordVerdict(blocked=True, reason=reason, detector=detector)
        if whitespace_insensitive and pattern.search(stripped):
            return NeverRecordVerdict(
                blocked=True, reason=reason, detector=f"{detector}_dewhitespaced"
            )

    for scanner in (
        _scan_credential_assignment,
        _scan_url_userinfo,
        _scan_payment_card,
    ):
        verdict = scanner(text)
        if verdict is not None:
            return verdict

    # Payment cards are the one non-regex detector worth re-running on the
    # de-whitespaced rendering: a card split across lines is a real shape.
    verdict = _scan_payment_card(stripped)
    if verdict is not None:
        return verdict

    # Entropy scoring runs on the RAW rendering only. Running it on the
    # de-whitespaced rendering was tried and is wrong: any prose paragraph
    # collapses into a single long token over the credential charset whose
    # entropy comfortably exceeds the threshold, so the firewall blocked
    # ordinary sentences like "Deploy uses blue-green with automatic
    # rollback". Whole-document entropy carries no signal about whether a
    # secret is present.
    #
    # The cost is a real, enumerated hole: an unknown-prefix secret split
    # across whitespace escapes the backstop. Split *known* formats are still
    # caught -- every vendor-prefix, JWT, and PEM pattern runs against the
    # de-whitespaced rendering above, which is the adversarial case #561's
    # acceptance criteria names.
    return _scan_high_entropy(text) or _CLEAN

never_record_key(class_name, kind)

Build a never-record Redis key.

The $NR: namespace is disjoint from $WF: (write filter) and $TOMB: (lifecycle tombstones, which deliberately archive the payload).

Parameters:

Name	Type	Description	Default
class_name	str	Model class name.	required
kind	str	"counts" or "drops".	required

Returns:

Name	Type	Description
str	str	Redis key like $NR:DefaultMemory:counts.

Source code in src/popoto/privacy/never_record.py

def never_record_key(class_name: str, kind: str) -> str:
    """Build a never-record Redis key.

    The ``$NR:`` namespace is disjoint from ``$WF:`` (write filter) and
    ``$TOMB:`` (lifecycle tombstones, which deliberately archive the payload).

    Args:
        class_name: Model class name.
        kind: ``"counts"`` or ``"drops"``.

    Returns:
        str: Redis key like ``$NR:DefaultMemory:counts``.
    """
    return f"$NR:{class_name}:{kind}"

write_tombstone(class_name, verdict, redis_client=None, *, model_class=None)

Record that a drop happened, without recording what was dropped.

Writes two plain Redis/Valkey structures (no modules):

• $NR:{class}:counts -- HASH, HINCRBY <reason> 1
• $NR:{class}:drops -- LIST, capped at :attr:Defaults.NR_TOMBSTONE_LOG_MAX

The entry id is :func:uuid.uuid4, deliberately not a hash or fingerprint of the content. A content-derived id would be a confirmation oracle: anyone holding a candidate secret could hash it and check the log for a match. This is the precise point where the never-record tombstone diverges from :class:popoto.recipes.memory_lifecycle.Tombstone, which stores an ExistenceFilter fingerprint on purpose so writes can be matched against it.

No field of the entry contains, derives from, or is length-correlated with the dropped text.

Parameters:

Name	Type	Description	Default
class_name	str	Model class name, for the key namespace.	required
verdict	NeverRecordVerdict	The blocking verdict. Only its reason/detector are stored.	required
redis_client	Any	Optional client override, for tests.	None
model_class	Any	The refusing model (#759 M4). On a non-Redis backend the entry goes to that backend's audit tables instead; omitted (or a Redis-bound model), it goes to Redis as before.	None

Returns:

Name	Type	Description
str	str	The tombstone entry id.

Source code in src/popoto/privacy/never_record.py

def write_tombstone(
    class_name: str,
    verdict: NeverRecordVerdict,
    redis_client: Any = None,
    *,
    model_class: Any = None,
) -> str:
    """Record that a drop happened, without recording what was dropped.

    Writes two plain Redis/Valkey structures (no modules):

    - ``$NR:{class}:counts`` -- HASH, ``HINCRBY <reason> 1``
    - ``$NR:{class}:drops``  -- LIST, capped at
      :attr:`Defaults.NR_TOMBSTONE_LOG_MAX`

    The entry id is :func:`uuid.uuid4`, deliberately **not** a hash or
    fingerprint of the content. A content-derived id would be a confirmation
    oracle: anyone holding a candidate secret could hash it and check the log
    for a match. This is the precise point where the never-record tombstone
    diverges from :class:`popoto.recipes.memory_lifecycle.Tombstone`, which
    stores an ExistenceFilter fingerprint on purpose so writes can be matched
    against it.

    No field of the entry contains, derives from, or is length-correlated
    with the dropped text.

    Args:
        class_name: Model class name, for the key namespace.
        verdict: The blocking verdict. Only its reason/detector are stored.
        redis_client: Optional client override, for tests.
        model_class: The refusing model (#759 M4). On a non-Redis backend
            the entry goes to that backend's audit tables instead; omitted
            (or a Redis-bound model), it goes to Redis as before.

    Returns:
        str: The tombstone entry id.
    """
    entry_id = str(uuid.uuid4())
    entry = json.dumps(
        {
            "id": entry_id,
            "reason": verdict.reason,
            "detector": verdict.detector,
            "at": round(time.time(), 3),
        },
        sort_keys=True,
    )

    backend = _audit_backend(model_class) if redis_client is None else None
    if backend is not None:
        # Best-effort, as the Redis pipeline below is.
        try:
            backend.field_call(
                model_class._meta.spec,
                "_never_record",
                "drop",
                verdict.reason or "unknown",
                entry,
                int(Defaults.NR_TOMBSTONE_LOG_MAX),
            )
        except Exception:  # pragma: no cover - exercised only on an outage
            pass
        return entry_id

    if redis_client is None:
        from ..redis_db import POPOTO_REDIS_DB

        redis_client = POPOTO_REDIS_DB

    counts_key = never_record_key(class_name, "counts")
    drops_key = never_record_key(class_name, "drops")

    # Best-effort: a firewall that raises when its audit log is unwritable
    # would fail *open* under Redis trouble, which is the one direction this
    # module must never fail in. The drop itself is enforced by the caller.
    try:
        pipe = redis_client.pipeline()
        pipe.hincrby(counts_key, verdict.reason or "unknown", 1)
        pipe.lpush(drops_key, entry)
        pipe.ltrim(drops_key, 0, Defaults.NR_TOMBSTONE_LOG_MAX - 1)
        pipe.execute()
    except Exception:  # pragma: no cover - exercised only on Redis failure
        pass

    return entry_id