From bb9ebec516a54ddf7f805a8c29437808f58aa4c6 Mon Sep 17 00:00:00 2001 From: Nikhil Arora Date: Sat, 3 Oct 2026 20:43:05 +0530 Subject: [PATCH] Add privacy policy and tamper-evident audit logging --- CHANGELOG.md | 37 +++ MIGRATING.md | 21 +- README.md | 106 ++++++++- logquill/__init__.py | 21 +- logquill/audit.py | 109 +++++++++ logquill/cli.py | 60 +++++ logquill/logger.py | 27 ++- logquill/plugins/pii_redact_plugin.py | 59 ++++- logquill/plugins/redact_plugin.py | 17 +- logquill/plugins/tamper_evident_plugin.py | 155 ++++++++++++- logquill/privacy.py | 147 ++++++++++++ schema/golden_records.json | 35 +++ schema/record.schema.json | 18 +- tests/test_audit.py | 138 ++++++++++++ tests/test_cli_verify.py | 134 +++++++++++ tests/test_otlp_transport.py | 11 +- tests/test_plugins/test_pii_redact_plugin.py | 100 +++++++++ tests/test_plugins/test_redact_plugin.py | 42 ++++ .../test_tamper_evident_plugin.py | 210 +++++++++++++++++- tests/test_privacy.py | 201 +++++++++++++++++ tests/test_state_diff.py | 5 +- 21 files changed, 1616 insertions(+), 37 deletions(-) create mode 100644 logquill/audit.py create mode 100644 logquill/privacy.py create mode 100644 tests/test_audit.py create mode 100644 tests/test_cli_verify.py create mode 100644 tests/test_privacy.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d4f343..53cf2c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,43 @@ All notable changes to this project are documented in this file. ## Unreleased +- Privacy, content policy, and an audit trail: + - Every `Logger` now has a `content_policy` (`"off"` by default \| `"hash"` + \| `"truncate"` \| `"full"`), governing the content fields an LLM call or + tool call can carry (`meta.input_messages`/`output_messages`/ + `system_instructions`/`tool_arguments`/`tool_result`/`state_diff`). + Applied before any plugin or transport ever sees the record, so a seeded + secret inside a prompt never reaches a transport unless you opt in. + `.child()` inherits the parent's policy. + - Field classification (`logquill.privacy.FIELD_CLASSES`: `secret` \| + `pii` \| `content`) lets a redaction rule target a class instead of + listing keys by name: `RedactPlugin(classes=["secret", "content"])`. + `RedactPlugin`'s existing default key set is unchanged, now sourced from + the same registry. + - `PIIRedactPlugin` gained patterns for well-known API-key/secret shapes — + OpenAI/Anthropic keys, AWS access key IDs, GitHub/Slack tokens, JWTs, + full PEM private-key blocks, generic `Bearer` tokens — alongside its + existing email/SSN/credit-card/phone patterns. Also fixed: overlapping + matches between two patterns (e.g. a phone-shaped digit run inside a + Slack token) used to redact in a sequence of independent substitutions, + which could leave a value only half-redacted; matches are now resolved + across all patterns in one pass before anything is replaced. + - `TamperEvidentPlugin` gained `.head_hash`/`.sign_head(key)` and the + module-level `verify_chain_detailed`/`verify_signed_chain`/`sign_head`/ + `verify_head_signature`. Hash-chaining alone can't catch a log file + being truncated — the remaining chain is perfectly self-consistent, just + shorter — so signing the chain's current head and checking it + separately closes that gap. `logquill verify ` (optionally + `--sign-key`/`--signature`) wraps this from the command line, reporting + where and why a chain broke. + - `AuditLogger` bundles `RedactPlugin`, `PIIRedactPlugin`, and + `TamperEvidentPlugin` with content capture off by default — the + technical controls an audit trail commonly needs, in one call. Its + `.child()` shares the parent's hash chain rather than starting a second, + interleaved one. Same scope-honesty note as the rest of the plugin + pipeline: none of this is a compliance guarantee on its own. + - The record contract gained `meta.tool_arguments`/`meta.tool_result`, + and documents which fields the content-capture policy governs. - Local-first trace viewer: - `logquill trace --file logs.jsonl` reconstructs and prints one agent run's span tree, annotated with each span's own duration and the diff --git a/MIGRATING.md b/MIGRATING.md index c8c424b..9d3457f 100644 --- a/MIGRATING.md +++ b/MIGRATING.md @@ -62,7 +62,9 @@ has them unless you use those. | `meta.mcp.server`, `meta.mcp.tool`, `meta.mcp.run_id` | `meta` | string | | `meta.tool`, `meta.tool_call_id`, `meta.provider`, `meta.agent_name`, `meta.agent_id`, `meta.response_model` | `meta` | string | | `meta.operation` | `meta` | `chat`, `text_completion` or `invoke_agent` | -| `meta.input_messages`, `meta.output_messages` | `meta` | array (opt-in prompt/completion content) | +| `meta.input_messages`, `meta.output_messages` | `meta` | array | +| `meta.system_instructions` | `meta` | array | +| `meta.tool_arguments`, `meta.tool_result` | `meta` | any JSON value | `llm` is its own block, not part of `meta`, because cost and latency dashboards need stable names for it. A record that isn't an LLM call has no @@ -85,3 +87,20 @@ and `logquill` on npm both test against. - If a transport of yours rebuilds a record from its five 1.x fields, copy the record instead (`{**record, "meta": new_meta}`) so `schema_version` and `llm` aren't dropped. + +## 6. Content capture is off by default + +The fields above that can carry a prompt, a completion, or a tool's +arguments/result — `input_messages`, `output_messages`, +`system_instructions`, `tool_arguments`, `tool_result`, and `state_diff` — +are now governed by a per-logger content-capture policy, `Logger( +content_policy="off" | "hash" | "truncate" | "full")`, **off by default**. +If you were relying on these fields reaching a transport as given, add +`content_policy="full"` explicitly. `AuditLogger` (new in this release) also +defaults to `"off"`. + +`RedactPlugin` also gained `classes=` (e.g. `RedactPlugin(classes= +["secret", "content"])`), redacting by field classification instead of only +by exact key name — see `logquill.privacy.FIELD_CLASSES`. Existing +`RedactPlugin()` calls with no arguments are unaffected: the default key set +is unchanged, just now sourced from that same classification. diff --git a/README.md b/README.md index db18b3e..1152242 100644 --- a/README.md +++ b/README.md @@ -535,10 +535,25 @@ logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-67 ``` Detection is regex-based by default — fast, dependency-free, matched on shape -rather than meaning. For fuzzier ML-based detection instead, pass -`use_presidio=True` (`pip install logquill[presidio]`) to route values -through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real, -opt-in dependency, never a default one. +rather than meaning. Beyond PII, the same scan also catches well-known +API-key/secret shapes wherever they appear — OpenAI/Anthropic keys, AWS +access key IDs, GitHub/Slack tokens, JWTs, PEM private-key blocks (the whole +block, not just the `BEGIN` line), and generic `Bearer` tokens — each a +provider's own fixed prefix or structure, chosen to keep false positives low. +For fuzzier ML-based detection instead, pass `use_presidio=True` (`pip +install logquill[presidio]`) to route values through Microsoft Presidio's +analyzer/anonymizer; Presidio stays a real, opt-in dependency, never a +default one. + +`RedactPlugin` can also target a **field classification** +(`logquill.privacy.FIELD_CLASSES`: `secret` \| `pii` \| `content`) instead of +listing key names one by one: + +```python +from logquill import RedactPlugin + +RedactPlugin(keys=(), classes=["secret"]) # exactly today's default key set, by class +``` ### Tamper-evident logs @@ -559,6 +574,89 @@ records[1]["message"] = "tampered" # simulate an edited log line assert TamperEvidentPlugin.verify_chain(records) is False ``` +`logquill verify ` wraps the same check for a JSONL file on disk, +reporting *where* and *why* it broke: + +```bash +logquill verify app.log +# logquill verify: OK — 142 record(s) verified in app.log +# logquill verify: chain head is 9f1c...e03a +``` + +Hash-chaining alone can't catch a file being **truncated** — delete the last +few lines and what remains is still perfectly self-consistent, just shorter. +Signing the chain's current head, and keeping that signature somewhere other +than the log file itself, catches that too: + +```python +from logquill import Logger, TamperEvidentPlugin, verify_signed_chain + +tamper = TamperEvidentPlugin() +logger = Logger("app", plugins=[tamper]) +records = [logger.info(f"step {i}") for i in range(3)] + +signature = tamper.sign_head(b"a-key-you-keep-somewhere-else") # not in this file + +assert verify_signed_chain(records, key=b"a-key-you-keep-somewhere-else", signature=signature).ok + +truncated = records[:-1] # an attacker deletes the last line +result = verify_signed_chain(truncated, key=b"a-key-you-keep-somewhere-else", signature=signature) +assert result.ok is False # caught — the chain is consistent but its head moved +``` + +```bash +logquill verify app.log --sign-key --signature +``` + +### Content capture policy + +Prompt/completion text, tool call arguments/results, and a span's captured +state (`meta.input_messages`, `output_messages`, `system_instructions`, +`tool_arguments`, `tool_result`, `state_diff` — the `content`-classed fields +in `logquill.privacy.FIELD_CLASSES`) are governed by a per-logger policy, +**off by default** — applied before any plugin or transport ever sees the +value, not something a plugin has to remember to attach: + +```python +from logquill import Logger + +logger = Logger("app.agent") # content_policy="off" by default + +record = logger.llm_call("chat", model="m", input_messages=["the user's actual prompt"]) +assert "the user's actual prompt" not in str(record["meta"]["input_messages"]) +``` + +`"hash"` replaces the value with a stable digest (the same content always +hashes the same, so two records can be confirmed to share one without either +ever reaching a transport); `"truncate"` keeps the first 200 characters; +`"full"` passes it through: + +```python +from logquill import Logger + +Logger("app.agent", content_policy="full") # see it all +Logger("app.agent", content_policy="hash") # correlate without exposing it +``` + +### `AuditLogger`: a preset for audit trails + +`AuditLogger` bundles `RedactPlugin`, `PIIRedactPlugin`, and +`TamperEvidentPlugin`, with content capture off by default — the technical +controls an audit trail commonly needs, in one call. (Scope-honesty note, +same as the rest of the plugin pipeline: this doesn't make a deployment +"compliant" with anything by itself.) + +```python +from logquill import AuditLogger + +audit_log = AuditLogger("audit") # pass transports=[FileTransport(...)] etc. as usual +record = audit_log.info("user login", user_id=42, password="hunter2") +assert record["meta"]["password"] == "***" +# .child() shares the same hash chain, not a second interleaved one + +signature = audit_log.sign_head(b"a-key-you-keep-somewhere-else") +``` + ### Alerting on errors `AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL diff --git a/logquill/__init__.py b/logquill/__init__.py index 09cf5a3..05872a3 100644 --- a/logquill/__init__.py +++ b/logquill/__init__.py @@ -1,4 +1,5 @@ from logquill.adapters.base import LogQuillAdapter +from logquill.audit import AuditLogger from logquill.config import load_config, logger_from_env, logger_from_file from logquill.context import bind_context, current_context from logquill.exceptions import format_exc_info @@ -20,8 +21,16 @@ from logquill.plugins.run_plugin import RunPlugin from logquill.plugins.sampling_plugin import SamplingPlugin from logquill.plugins.slack_alert_plugin import SlackAlertPlugin -from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin +from logquill.plugins.tamper_evident_plugin import ( + TamperEvidentPlugin, + VerificationResult, + sign_head, + verify_chain_detailed, + verify_head_signature, + verify_signed_chain, +) from logquill.plugins.trace_context_plugin import TraceContextPlugin +from logquill.privacy import CONTENT_FIELDS, FIELD_CLASSES, ContentCapturePolicy, FieldClass from logquill.records import SCHEMA_VERSION, LLMBlock, LogRecord, parse_record from logquill.serverless import with_azure_function, with_cloud_function, with_lambda from logquill.toggle import disable, enable, is_enabled @@ -60,18 +69,23 @@ "AppriseAlertPlugin", "AppInsightsTransport", "AsyncWorker", + "AuditLogger", "BaseQueueTransport", "BaseSQLTransport", "BatchingTransport", "CloudLoggingTransport", "CloudWatchTransport", "CollectingTransport", + "CONTENT_FIELDS", "ConsoleTransport", + "ContentCapturePolicy", "ContextPlugin", "DatadogTransport", "DynamoDBTransport", "ElasticsearchTransport", "EmailAlertPlugin", + "FIELD_CLASSES", + "FieldClass", "FileTransport", "Formatter", "FunctionPlugin", @@ -114,6 +128,7 @@ "TextFormatter", "TraceContextPlugin", "Transport", + "VerificationResult", "bind_context", "current_context", "disable", @@ -127,6 +142,10 @@ "parse_level", "parse_logfmt", "parse_record", + "sign_head", + "verify_chain_detailed", + "verify_head_signature", + "verify_signed_chain", "with_azure_function", "with_cloud_function", "with_lambda", diff --git a/logquill/audit.py b/logquill/audit.py new file mode 100644 index 0000000..05f5176 --- /dev/null +++ b/logquill/audit.py @@ -0,0 +1,109 @@ +"""`AuditLogger` — a `Logger` preset bundling LogQuill's privacy and +integrity controls for an audit trail: content capture off, secret and PII +redaction on, and hash-chained records so the result can be checked for +tampering afterward. + +Scope honesty, same as the rest of the plugin pipeline: none of this makes +a deployment "compliant" with anything on its own — it's a technical +control that commonly supports compliance work (SOC 2, HIPAA, PCI-DSS audit +trails all expect *something* like this), never a substitute for the +process and review compliance actually requires. +""" + +from __future__ import annotations + +from typing import Any + +from logquill.levels import Level +from logquill.logger import Logger +from logquill.plugins.context_plugin import ContextPlugin +from logquill.plugins.pii_redact_plugin import PIIRedactPlugin +from logquill.plugins.plugin import MiddlewareFunc, Plugin +from logquill.plugins.redact_plugin import RedactPlugin +from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin +from logquill.privacy import ContentCapturePolicy +from logquill.records import LogRecord # noqa: F401 — resolves MiddlewareFunc's forward reference +from logquill.transports.transport import Transport + + +class AuditLogger(Logger): + """A `Logger` preset for an audit trail. Equivalent to: + + Logger( + name, + content_policy="off", + plugins=[RedactPlugin(), PIIRedactPlugin(), TamperEvidentPlugin(), *extra_plugins], + ) + + plus `.sign_head(key)` and `.head_hash`, forwarded to the + `TamperEvidentPlugin` instance this attaches — see that class, and + `logquill.plugins.tamper_evident_plugin.verify_signed_chain`, for what + to do with a signature and how `logquill verify` checks one. + + `content_policy` defaults to `"off"` here too — pass `"hash"` if an + audit trail should at least be able to confirm two entries shared a + prompt without storing it. `extra_plugins` run after the three above, + in the order given, the same as any other `Logger`. + """ + + def __init__( + self, + name: str, + *, + level: int | str | Level = Level.INFO, + transports: list[Transport] | None = None, + content_policy: ContentCapturePolicy | str = "off", + extra_plugins: list[Plugin | MiddlewareFunc] | None = None, + **kwargs: Any, + ) -> None: + """`**kwargs` are forwarded to `Logger.__init__` (`async_dispatch`, + `flush_at_exit`, etc.).""" + self._tamper = TamperEvidentPlugin() + plugins: list[Plugin | MiddlewareFunc] = [ + RedactPlugin(), + PIIRedactPlugin(), + self._tamper, + *(extra_plugins or []), + ] + super().__init__( + name, + level=level, + transports=transports, + plugins=plugins, + content_policy=content_policy, + **kwargs, + ) + + def child(self, name: str, /, **fixed_meta: Any) -> AuditLogger: + """Like `Logger.child()`, but returns another `AuditLogger` with + the *same* redaction plugins and, importantly, the **same** + `TamperEvidentPlugin` instance as this one — a child shares this + logger's transports already (the same output stream), so it + continues this logger's hash chain rather than starting a second, + interleaved one that would make the combined file unverifiable as + a single chain.""" + child_logger = AuditLogger( + f"{self.name}.{name}", + level=self._level, + transports=self.transports, + content_policy=self.content_policy, + ) + child_logger._tamper = self._tamper + child_logger.plugins = [RedactPlugin(), PIIRedactPlugin(), self._tamper] + if fixed_meta: + child_logger.use(ContextPlugin(**fixed_meta)) + child_logger._worker = self._worker + return child_logger + + @property + def head_hash(self) -> str: + """This logger's current hash-chain head — see + `TamperEvidentPlugin.head_hash`.""" + return self._tamper.head_hash + + def sign_head(self, key: bytes) -> str: + """Signs `self.head_hash` with `key` — see + `TamperEvidentPlugin.sign_head`/the module-level `sign_head` for + what to do with the result and why it has to be kept somewhere + other than this logger's own output.""" + return self._tamper.sign_head(key) diff --git a/logquill/cli.py b/logquill/cli.py index 179029e..8138656 100644 --- a/logquill/cli.py +++ b/logquill/cli.py @@ -4,6 +4,7 @@ import argparse import contextlib +import io import json import sys import time @@ -12,6 +13,7 @@ from logquill.formatters import format_text from logquill.levels import Level, parse_level +from logquill.plugins.tamper_evident_plugin import verify_chain_detailed, verify_signed_chain from logquill.trace_tree import build_trace, render_tree from logquill.webui import JSONLSource, RecordSource, SQLiteSource, TraceViewerServer @@ -126,6 +128,25 @@ def build_parser() -> argparse.ArgumentParser: metavar="N", help="At most this many existing lines seed the initial view (default: 5000).", ) + + verify_parser = subparsers.add_parser( + "verify", + help="Check a TamperEvidentPlugin-written JSONL log file's hash chain.", + ) + verify_parser.add_argument("file", help="Path to a LogQuill JSONL log file.") + verify_parser.add_argument( + "--sign-key", + metavar="HEX", + default=None, + help="Hex-encoded key, with --signature: also check the chain's head against a " + "signature taken with TamperEvidentPlugin.sign_head/AuditLogger.sign_head.", + ) + verify_parser.add_argument( + "--signature", + metavar="HEX", + default=None, + help="Hex-encoded signature to check against, with --sign-key.", + ) return parser @@ -439,6 +460,42 @@ def redraw() -> None: return 0 +def _run_verify(args: argparse.Namespace, *, out: IO[str]) -> int: + path = Path(args.file) + if not path.exists(): + out.write(f"logquill verify: no such file: {args.file}\n") + return 1 + + if bool(args.sign_key) != bool(args.signature): + out.write("logquill verify: --sign-key and --signature must be given together\n") + return 1 + + records = _iter_records(path, warn_stream=io.StringIO()) + if args.sign_key: + try: + key = bytes.fromhex(args.sign_key) + except ValueError: + out.write("logquill verify: --sign-key must be hex-encoded\n") + return 1 + result = verify_signed_chain(records, key=key, signature=args.signature) + else: + result = verify_chain_detailed(records) + + if result.ok: + out.write( + f"logquill verify: OK — {result.records_checked} record(s) verified in {args.file}\n" + ) + if result.head_hash is not None: + out.write(f"logquill verify: chain head is {result.head_hash}\n") + return 0 + + out.write( + f"logquill verify: FAILED — {result.reason} (at record {result.broken_at} of " + f"{result.records_checked} checked) in {args.file}\n" + ) + return 1 + + def main(argv: Sequence[str] | None = None) -> int: """The `logquill` console-script entry point: parses `argv` (defaulting to `sys.argv`) and dispatches to the matching subcommand, returning the @@ -455,6 +512,9 @@ def main(argv: Sequence[str] | None = None) -> int: if args.command == "dev": return _run_dev(args, out=sys.stdout, warn_stream=sys.stderr) + if args.command == "verify": + return _run_verify(args, out=sys.stdout) + if args.command != "tail": parser.error(f"Unknown command: {args.command}") diff --git a/logquill/logger.py b/logquill/logger.py index 08e74e5..c04c14a 100644 --- a/logquill/logger.py +++ b/logquill/logger.py @@ -11,6 +11,7 @@ from logquill.opt import OptLogger, caller_info, resolve_lazy from logquill.plugins.context_plugin import ContextPlugin from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin +from logquill.privacy import ContentCapturePolicy, apply_content_policy, parse_content_policy from logquill.records import LLMBlock, LogRecord, build_llm_block, create_record from logquill.retry import RetryTracker from logquill.span import SpanContext, current_span_id @@ -40,6 +41,7 @@ def __init__( max_queue_size: int = 10_000, backpressure: BackpressurePolicy = "drop_oldest", flush_at_exit: bool = True, + content_policy: ContentCapturePolicy | str = "off", ) -> None: """`async_dispatch=True` moves per-record transport writes (and the `after_log` plugin hooks that follow them) onto a background thread, @@ -61,9 +63,24 @@ def __init__( `sys.exit()`, an unhandled exception) but not when the process is killed outright (`SIGKILL`, `os._exit()`, or `SIGTERM` with no handler installed) — handle `SIGTERM` and call `close()` for that. + + `content_policy` governs the content fields — an LLM call's prompt/ + completion, a tool call's arguments/result, a span's captured state + (see `logquill.privacy.CONTENT_FIELDS`) — **`"off"` by default**: + every call site, this logger's own plugins, and every transport + never see the raw value unless you choose otherwise. `"hash"` + replaces it with a stable digest (same content, same hash, so two + records can be confirmed to share one without exposing it); + `"truncate"` keeps the first `logquill.privacy.TRUNCATE_CHARS` + characters; `"full"` passes it through. Applied before any plugin + sees the record, so no plugin — not even a custom one — can + override it; see `logquill.privacy` for the full explanation and + `AuditLogger` for a profile that bundles this with the other + privacy/integrity controls. """ self.name = name self._level = parse_level(level) + self.content_policy: ContentCapturePolicy = parse_content_policy(content_policy) self.transports: list[Transport] = list(transports) if transports else [] self.plugins: list[Plugin] = [] for plugin in plugins or []: @@ -109,8 +126,14 @@ def child(self, name: str, /, **fixed_meta: Any) -> Logger: a child can `.use(RunPlugin())` without attaching it to the parent's pipeline too. Any `fixed_meta` given is injected into every record the child produces, via an internal `ContextPlugin`. + Inherits this logger's `content_policy`. """ - child_logger = Logger(f"{self.name}.{name}", level=self._level, transports=self.transports) + child_logger = Logger( + f"{self.name}.{name}", + level=self._level, + transports=self.transports, + content_policy=self.content_policy, + ) if fixed_meta: child_logger.use(ContextPlugin(**fixed_meta)) # Share the parent's worker (if any) rather than spinning up a second @@ -283,6 +306,8 @@ def _log( if parent_span_id is not None: record["meta"].setdefault("parent_span_id", parent_span_id) + apply_content_policy(record["meta"], self.content_policy) + for plugin in self.plugins: try: result = plugin.before_log(record) diff --git a/logquill/plugins/pii_redact_plugin.py b/logquill/plugins/pii_redact_plugin.py index d97bd04..5c806fe 100644 --- a/logquill/plugins/pii_redact_plugin.py +++ b/logquill/plugins/pii_redact_plugin.py @@ -11,11 +11,32 @@ #: positives (a random 9-digit number) and false negatives (anything that #: doesn't look like these shapes) are expected. Override via `patterns=` #: for anything stricter. +#: +#: The API-key/secret entries below aren't "PII" in the usual sense — +#: they're here because they're exactly the other thing a stray value in an +#: LLM prompt, a tool argument, or a log line can leak, and the same +#: scan-every-value-regardless-of-key mechanism applies. Each is a +#: well-known provider's own fixed prefix/shape, chosen specifically to +#: keep the false-positive rate low (unlike a generic "looks random" check, +#: which would flag plenty of non-secrets). DEFAULT_PII_PATTERNS: dict[str, re.Pattern[str]] = { "email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"), "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), "credit_card": re.compile(r"\b(?:\d[ -]?){13,16}\b"), "phone": re.compile(r"\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"), + "openai_api_key": re.compile(r"\bsk-(?:proj-)?[A-Za-z0-9_-]{20,}\b"), + "anthropic_api_key": re.compile(r"\bsk-ant-[A-Za-z0-9_-]{20,}\b"), + "aws_access_key_id": re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"), + "github_token": re.compile(r"\bgh[pousr]_[A-Za-z0-9]{36,}\b"), + "slack_token": re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b"), + "jwt": re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b"), + "private_key_block": re.compile( + r"-----BEGIN (?:RSA |EC |OPENSSH |DSA |)PRIVATE KEY-----" + r".*?" + r"-----END (?:RSA |EC |OPENSSH |DSA |)PRIVATE KEY-----", + re.DOTALL, + ), + "bearer_token": re.compile(r"\bBearer\s+[A-Za-z0-9\-_.]{20,}\b"), } _MAX_DEPTH = 50 @@ -115,9 +136,41 @@ def _redact_value(self, value: Any, seen: set[int], depth: int) -> Any: def _redact_text(self, text: str) -> str: if self.use_presidio: return self._redact_with_presidio(text) - for pattern in self.patterns.values(): - text = pattern.sub(self.replacement, text) - return text + + # Found independently per pattern, over the *original* text — not + # applied as a sequence of in-place substitutions. Doing it + # in-place would let an earlier pattern's replacement text corrupt a + # later pattern's match (e.g. a phone-shaped digit run inside a + # Slack token: redacting the digits first leaves the token only + # half-redacted, worse than catching it correctly or not at all). + spans = [ + (match.start(), match.end()) + for pattern in self.patterns.values() + for match in pattern.finditer(text) + ] + if not spans: + return text + + # Leftmost first, longest first on a tie, then take non-overlapping + # spans in that order — the standard "resolve overlapping matches" + # sweep, so two patterns matching the same stretch of text redact it + # once, as the longer/more specific match, not twice or partially. + spans.sort(key=lambda span: (span[0], -span[1])) + chosen: list[tuple[int, int]] = [] + last_end = -1 + for start, end in spans: + if start >= last_end: + chosen.append((start, end)) + last_end = end + + pieces = [] + cursor = 0 + for start, end in chosen: + pieces.append(text[cursor:start]) + pieces.append(self.replacement) + cursor = end + pieces.append(text[cursor:]) + return "".join(pieces) def _redact_with_presidio(self, text: str) -> str: results = self._analyzer.analyze( diff --git a/logquill/plugins/redact_plugin.py b/logquill/plugins/redact_plugin.py index 66cc9df..f2b4d4b 100644 --- a/logquill/plugins/redact_plugin.py +++ b/logquill/plugins/redact_plugin.py @@ -3,9 +3,13 @@ from collections.abc import Iterable from logquill.plugins.plugin import Plugin +from logquill.privacy import FieldClass, keys_in_class from logquill.records import LogRecord -DEFAULT_REDACTED_KEYS = frozenset({"password", "token", "secret", "api_key", "authorization"}) +#: The `secret`-classed keys in `logquill.privacy.FIELD_CLASSES` — kept as its +#: own name since it's also `RedactPlugin`'s default, and existing code may +#: already import it directly. +DEFAULT_REDACTED_KEYS = frozenset(keys_in_class("secret")) class RedactPlugin(Plugin): @@ -15,10 +19,19 @@ def __init__( self, keys: Iterable[str] = DEFAULT_REDACTED_KEYS, replacement: str = "***", + *, + classes: Iterable[FieldClass] | None = None, ) -> None: """`keys` defaults to `DEFAULT_REDACTED_KEYS`; matching is - case-insensitive, so callers don't need to worry about casing.""" + case-insensitive, so callers don't need to worry about casing. + `classes`, if given, adds every key `logquill.privacy.FIELD_CLASSES` + tags with one of the given classes — `RedactPlugin(keys=(), + classes=["secret", "content"])` redacts both the usual credential + keys and every content field, by class rather than listing each one. + """ self.keys = {key.lower() for key in keys} + for field_class in classes or (): + self.keys |= {key.lower() for key in keys_in_class(field_class)} self.replacement = replacement def before_log(self, record: LogRecord) -> LogRecord | None: diff --git a/logquill/plugins/tamper_evident_plugin.py b/logquill/plugins/tamper_evident_plugin.py index bc4e391..49d6c89 100644 --- a/logquill/plugins/tamper_evident_plugin.py +++ b/logquill/plugins/tamper_evident_plugin.py @@ -1,8 +1,10 @@ from __future__ import annotations import hashlib +import hmac import json from collections.abc import Iterable, Mapping +from dataclasses import dataclass from typing import Any from logquill.plugins.plugin import Plugin @@ -11,6 +13,122 @@ GENESIS_HASH = "0" * 64 +@dataclass(frozen=True) +class VerificationResult: + """What `verify_chain_detailed`/`verify_signed_chain` found. + + `ok=False` means stop reading `broken_at` as "trustworthy up to + here" — it's exactly where corruption was *detected*, not necessarily + where it was *introduced* (a deleted line, for instance, is detected at + the record right after the gap). + """ + + ok: bool + records_checked: int + broken_at: int | None + reason: str | None + head_hash: str | None + + +def verify_chain_detailed( + records: Iterable[Mapping[str, Any]], *, genesis_hash: str = GENESIS_HASH +) -> VerificationResult: + """Like `TamperEvidentPlugin.verify_chain`, but reports where and why, + instead of a bare `bool` — what `logquill verify` uses to print an + actionable result. `records_checked` counts every record looked at, + including the one the chain broke on; `head_hash` is the chain's hash + as of the last record verified (the full chain's head, on success). + """ + prev_hash = genesis_hash + checked = 0 + for record in records: + checked += 1 + meta = record.get("meta", {}) + stored_hash = meta.get("hash") + stored_prev_hash = meta.get("prev_hash") + if stored_hash is None: + return VerificationResult( + False, + checked, + checked, + "this record has no meta.hash — it was never chained, or that field was removed", + prev_hash if checked > 1 else None, + ) + if stored_prev_hash != prev_hash: + return VerificationResult( + False, + checked, + checked, + "meta.prev_hash doesn't match the previous record's hash — a line before " + "this one was edited, removed, or reordered", + prev_hash, + ) + if _compute_hash(record, prev_hash) != stored_hash: + return VerificationResult( + False, + checked, + checked, + "this record's content doesn't match its own stored hash — this line was " + "edited after being chained", + prev_hash, + ) + prev_hash = stored_hash + return VerificationResult(True, checked, None, None, prev_hash if checked else genesis_hash) + + +def sign_head(head_hash: str, key: bytes) -> str: + """HMAC-SHA256 of `head_hash` under `key`, hex-encoded — call this with + `TamperEvidentPlugin.head_hash` (or `.sign_head(key)`, which does exactly + this) at any point you want to be able to prove the chain's state later, + e.g. right before `logger.close()`. + + **Store the result somewhere the log file itself doesn't live** — a + secrets vault, a separate append-only ledger, a console the tamperer + doesn't control. A signature stored inside the same file it attests to + protects nothing: whoever can edit the log can edit the signature next + to it too. What this buys you: internal hash-chaining alone can't + detect the file being truncated and the last N lines dropped (the + remaining chain is perfectly self-consistent, just shorter) — comparing + the chain's current head against an independently-held signature can. + """ + return hmac.new(key, head_hash.encode("utf-8"), hashlib.sha256).hexdigest() + + +def verify_head_signature(head_hash: str, key: bytes, signature: str) -> bool: + """Whether `signature` is what `sign_head(head_hash, key)` would + produce — using `hmac.compare_digest`, so this doesn't leak timing + information about how much of `signature` matched.""" + return hmac.compare_digest(sign_head(head_hash, key), signature) + + +def verify_signed_chain( + records: Iterable[Mapping[str, Any]], + *, + key: bytes, + signature: str, + genesis_hash: str = GENESIS_HASH, +) -> VerificationResult: + """`verify_chain_detailed`, plus confirming the chain's resulting head + matches an independently-held `signature` — see `sign_head` for why + that catches what hash-chaining alone can't: the file being truncated + (or wholesale replaced) after the signature was taken. + """ + result = verify_chain_detailed(records, genesis_hash=genesis_hash) + if not result.ok: + return result + if result.head_hash is None or not verify_head_signature(result.head_hash, key, signature): + return VerificationResult( + False, + result.records_checked, + result.records_checked, + "the chain is internally consistent, but its head doesn't match the given " + "signature — the file may have been truncated, or replaced outright, after " + "the signature was taken", + result.head_hash, + ) + return result + + class TamperEvidentPlugin(Plugin): """Hash-chains every record so tampering with a written log can be detected after the fact. @@ -26,7 +144,16 @@ class TamperEvidentPlugin(Plugin): Verify a previously-written log with `TamperEvidentPlugin.verify_chain`, which re-derives each record's hash from its content and confirms it matches both the stored `meta.hash` and the chain built from the - records before it, in order. + records before it, in order (`logquill verify ` wraps this). + `verify_chain_detailed` reports *where* and *why* a chain broke instead + of a bare `bool`. + + Hash-chaining alone can't catch a file being truncated — a tamperer who + deletes the last N lines leaves a perfectly self-consistent, just + shorter, chain. `sign_head(key)` (or the module-level `sign_head`) + signs the chain's current head so that can be caught too: see + `verify_signed_chain` and `sign_head`'s own docstring for where to keep + that signature. """ def __init__(self, *, genesis_hash: str = GENESIS_HASH) -> None: @@ -45,6 +172,19 @@ def before_log(self, record: LogRecord) -> LogRecord | None: self._last_hash = digest return record + @property + def head_hash(self) -> str: + """This instance's current chain head — the `hash` of the last + record it processed, or `genesis_hash` if it hasn't processed one + yet. Pass to `sign_head`/`.sign_head()` to make the chain's current + state provable later.""" + return self._last_hash + + def sign_head(self, key: bytes) -> str: + """`sign_head(self.head_hash, key)` — see the module-level + `sign_head` for what to do with the result and why.""" + return sign_head(self._last_hash, key) + @staticmethod def verify_chain( records: Iterable[Mapping[str, Any]], *, genesis_hash: str = GENESIS_HASH @@ -52,18 +192,9 @@ def verify_chain( """Return `True` iff every record's hash matches its content plus the previous record's hash, in the given order. Returns `False` at the first break in the chain (an edited, removed, or reordered record). + See `verify_chain_detailed` for *where* and *why* it broke. """ - prev_hash = genesis_hash - for record in records: - meta = record.get("meta", {}) - stored_hash = meta.get("hash") - stored_prev_hash = meta.get("prev_hash") - if stored_hash is None or stored_prev_hash != prev_hash: - return False - if _compute_hash(record, prev_hash) != stored_hash: - return False - prev_hash = stored_hash - return True + return verify_chain_detailed(records, genesis_hash=genesis_hash).ok def _compute_hash(record: Mapping[str, Any], prev_hash: str) -> str: diff --git a/logquill/privacy.py b/logquill/privacy.py new file mode 100644 index 0000000..cc46746 --- /dev/null +++ b/logquill/privacy.py @@ -0,0 +1,147 @@ +"""Field classification and the content-capture policy: the two pieces of +LogQuill's privacy controls that aren't specific to one plugin. + +**Field classification** lets a redaction rule say "every `secret`-classed +field" instead of listing key names one by one — `RedactPlugin(classes= +["secret"])` redacts everything in `FIELD_CLASSES` tagged `"secret"`, which +is also what its own default (no arguments) now resolves to, so nothing +about `RedactPlugin`'s existing behavior changed by adding this. + +**Content capture** governs the content fields specifically — an LLM call's +prompt/completion (`meta.input_messages`/`output_messages`), a tool call's +arguments/result (`meta.tool_arguments`/`tool_result`), a span's captured +state (`meta.state_diff`), and a model's system prompt +(`meta.system_instructions`). Unlike a `secret` (always redact) or `pii` +(scan for a pattern), this content is sometimes exactly what you want to +see while debugging and sometimes exactly what you must never write to a +transport — so it's governed by an explicit, per-logger policy +(`Logger(content_policy=...)`), **off by default**, applied unconditionally +inside `Logger._log()` for every record regardless of call site (span +closes, the stdlib bridge, a framework adapter) rather than being a plugin +someone has to remember to attach. + +Same scope-honesty note as the rest of the plugin pipeline: none of this +makes a deployment "compliant" with anything — it's a technical control +that commonly supports compliance work, not a guarantee of it. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from typing import Any, Literal + +_logger = logging.getLogger("logquill") + +FieldClass = Literal["secret", "pii", "content"] + +#: Which class a well-known `meta` key belongs to. `secret`: always-redact +#: credentials. `pii`: personal data `PIIRedactPlugin` also scans for by +#: pattern regardless of key. `content`: prompt/completion/tool-call data +#: governed by `ContentCapturePolicy` instead of a flat redact-or-don't rule. +FIELD_CLASSES: dict[str, FieldClass] = { + "password": "secret", + "token": "secret", + "secret": "secret", + "api_key": "secret", + "authorization": "secret", + "email": "pii", + "ssn": "pii", + "phone": "pii", + "input_messages": "content", + "output_messages": "content", + "system_instructions": "content", + "tool_arguments": "content", + "tool_result": "content", + "state_diff": "content", +} + +#: The `meta` keys `Logger(content_policy=...)` governs — see the module +#: docstring. Kept as its own tuple (rather than deriving it from +#: `FIELD_CLASSES` on every call) since `Logger._log` checks it on every +#: record. +CONTENT_FIELDS: tuple[str, ...] = tuple( + key for key, cls in FIELD_CLASSES.items() if cls == "content" +) + + +def keys_in_class( + field_class: FieldClass, registry: dict[str, FieldClass] = FIELD_CLASSES +) -> set[str]: + """Every key `registry` classifies as `field_class`.""" + return {key for key, cls in registry.items() if cls == field_class} + + +ContentCapturePolicy = Literal["off", "hash", "truncate", "full"] +_VALID_POLICIES = ("off", "hash", "truncate", "full") + +#: How much of a `truncate`-policy value's JSON text survives. +TRUNCATE_CHARS = 200 + +_OFF_PLACEHOLDER = "" + + +def parse_content_policy(policy: ContentCapturePolicy | str) -> ContentCapturePolicy: + """Validates `policy`, raising `ValueError` naming the valid choices if + it isn't one of them. Accepts a plain string so it can come from config + the same way a `Level` does.""" + if policy not in _VALID_POLICIES: + raise ValueError(f"content_policy must be one of {_VALID_POLICIES}, got {policy!r}") + return policy # type: ignore[return-value] + + +def _to_text(value: Any) -> str: + try: + return json.dumps(value, sort_keys=True, default=str) + except (TypeError, ValueError, RecursionError): + return str(value) + + +def _apply_one(value: Any, policy: ContentCapturePolicy) -> Any: + if policy == "full": + return value + if policy == "off": + return _OFF_PLACEHOLDER + text = _to_text(value) + if policy == "hash": + digest = hashlib.sha256(text.encode("utf-8", errors="surrogatepass")).hexdigest() + return f"sha256:{digest}" + # truncate + if len(text) <= TRUNCATE_CHARS: + return text + return text[:TRUNCATE_CHARS] + f"...(truncated, {len(text)} chars total)" + + +def apply_content_policy( + meta: dict[str, Any], + policy: ContentCapturePolicy, + *, + fields: tuple[str, ...] = CONTENT_FIELDS, +) -> None: + """Applies `policy` to every key in `fields` that's present in `meta`, + in place. `"full"` is a no-op (by far the common case — most records + carry no content field at all — so this returns immediately rather than + walking `fields` against an empty intersection). + + - `"off"` (default): the value is replaced with a fixed placeholder + string. Never the raw content, and never silently dropped either — + a reader can tell capture was configured off rather than wondering + why the field is simply missing. + - `"hash"`: replaced with `"sha256:"` of the value's JSON text. + The same content always hashes the same, so two records can be + confirmed to share a prompt/completion without either ever being + written anywhere. + - `"truncate"`: the value's JSON text, cut to `TRUNCATE_CHARS` + characters with a marker noting the original length. + - `"full"`: passed through unchanged. + + A value that can't be JSON-serialized (a circular reference, a raw + object) falls back to `str()` for `"hash"`/`"truncate"` rather than + raising — a log call must not fail over this. + """ + if policy == "full": + return + for field in fields: + if field in meta: + meta[field] = _apply_one(meta[field], policy) diff --git a/schema/golden_records.json b/schema/golden_records.json index e43c265..52d4e94 100644 --- a/schema/golden_records.json +++ b/schema/golden_records.json @@ -413,6 +413,26 @@ } } } + }, + { + "name": "tool call with arguments and result", + "record": { + "schema_version": "2.0", + "timestamp": "2026-01-01T00:00:00.000Z", + "level": "INFO", + "logger": "app.agent", + "message": "search", + "meta": { + "kind": "action", + "tool": "search", + "tool_arguments": { + "q": "x" + }, + "tool_result": { + "ok": true + } + } + } } ], "invalid": [ @@ -931,6 +951,21 @@ } }, "why": "mcp.run_id is a string" + }, + { + "name": "tool_arguments at the top level, not meta", + "record": { + "schema_version": "2.0", + "timestamp": "2026-01-01T00:00:00.000Z", + "level": "INFO", + "logger": "app.agent", + "message": "search", + "meta": {}, + "tool_arguments": { + "q": "x" + } + }, + "why": "tool_arguments lives in meta, not the top level" } ], "legacy": [ diff --git a/schema/record.schema.json b/schema/record.schema.json index 339b7f5..99a61f9 100644 --- a/schema/record.schema.json +++ b/schema/record.schema.json @@ -92,7 +92,7 @@ }, "state_diff": { "type": "object", - "description": "State captured before and after a span." + "description": "State captured before and after a span, governed by Logger(content_policy=...) the same as input_messages." }, "mcp": { "$ref": "#/$defs/mcp" @@ -103,11 +103,13 @@ }, "hash": { "type": "string", - "pattern": "^[0-9a-f]{64}$" + "pattern": "^[0-9a-f]{64}$", + "description": "SHA-256 hex digest over this record's content plus prev_hash." }, "prev_hash": { "type": "string", - "pattern": "^[0-9a-f]{64}$" + "pattern": "^[0-9a-f]{64}$", + "description": "The previous record's hash in a TamperEvidentPlugin chain." }, "tool": { "type": "string", @@ -142,11 +144,17 @@ }, "input_messages": { "type": "array", - "description": "Opt-in prompt content; only exported to OpenTelemetry when content capture is on." + "description": "Prompt content governed by Logger(content_policy=...) — off by default; only exported to OpenTelemetry when OTLPTransport's capture_content is on." }, "output_messages": { "type": "array", - "description": "Opt-in completion content." + "description": "Completion content, governed the same way as input_messages." + }, + "tool_arguments": { + "description": "A tool call's arguments — any JSON value, governed by Logger(content_policy=...) the same as input_messages." + }, + "tool_result": { + "description": "A tool call's result — any JSON value, governed by Logger(content_policy=...) the same as input_messages." } } }, diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..b50f31e --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import pytest + +from logquill import AuditLogger +from logquill.plugins.pii_redact_plugin import PIIRedactPlugin +from logquill.plugins.redact_plugin import RedactPlugin +from logquill.plugins.tamper_evident_plugin import ( + TamperEvidentPlugin, + verify_chain_detailed, + verify_signed_chain, +) +from logquill.transports.transport import CollectingTransport + + +def _audit_logger(**kwargs: object) -> tuple[AuditLogger, CollectingTransport]: + sink = CollectingTransport() + return AuditLogger("audit", transports=[sink], **kwargs), sink # type: ignore[arg-type] + + +def test_content_policy_defaults_to_off() -> None: + logger, _ = _audit_logger() + + assert logger.content_policy == "off" + + +def test_bundles_redact_pii_and_tamper_evident_plugins() -> None: + logger, _ = _audit_logger() + + kinds = [type(plugin) for plugin in logger.plugins] + + assert RedactPlugin in kinds + assert PIIRedactPlugin in kinds + assert TamperEvidentPlugin in kinds + + +def test_a_seeded_secret_and_pii_never_reach_the_transport() -> None: + logger, sink = _audit_logger() + + logger.info("login", password="hunter2", note="contact a@example.com") + + assert "hunter2" not in str(sink.records[0]) + assert "a@example.com" not in str(sink.records[0]) + + +def test_a_seeded_secret_inside_a_prompt_never_reaches_the_transport() -> None: + logger, sink = _audit_logger() + + logger.llm_call("chat", model="m", input_messages=["the user's actual secret prompt"]) + + assert "the user's actual secret prompt" not in str(sink.records[0]) + + +def test_records_are_hash_chained() -> None: + logger, sink = _audit_logger() + + logger.info("one") + logger.info("two") + + assert verify_chain_detailed(sink.records).ok is True + + +def test_head_hash_and_sign_head_are_exposed() -> None: + logger, sink = _audit_logger() + logger.info("one") + + signature = logger.sign_head(b"key") + + assert logger.head_hash == sink.records[-1]["meta"]["hash"] + result = verify_signed_chain(sink.records, key=b"key", signature=signature) + assert result.ok is True + + +def test_extra_plugins_run_after_the_bundled_ones() -> None: + from logquill.plugins.plugin import Plugin + + seen = [] + + class Spy(Plugin): + def before_log(self, record): # type: ignore[no-untyped-def] + seen.append(record["meta"].get("password")) # already redacted by now + return record + + logger, _ = _audit_logger(extra_plugins=[Spy()]) + + logger.info("x", password="hunter2") + + assert seen == ["***"] + + +def test_content_policy_can_be_overridden() -> None: + logger, sink = _audit_logger(content_policy="full") + + logger.llm_call("chat", model="m", input_messages=["raw prompt"]) + + assert sink.records[0]["meta"]["input_messages"] == ["raw prompt"] + + +def test_an_invalid_content_policy_still_raises() -> None: + with pytest.raises(ValueError, match="content_policy"): + AuditLogger("audit", content_policy="plaintext") + + +def test_child_continues_the_same_hash_chain_as_the_parent() -> None: + logger, sink = _audit_logger() + child = logger.child("sub") + + logger.info("from parent") + child.info("from child") + + assert isinstance(child, AuditLogger) + assert verify_chain_detailed(sink.records).ok is True + assert sink.records[1]["meta"]["prev_hash"] == sink.records[0]["meta"]["hash"] + + +def test_child_also_redacts_secrets_and_pii() -> None: + logger, sink = _audit_logger() + child = logger.child("sub") + + child.info("x", password="hunter2") + + assert "hunter2" not in str(sink.records[-1]) + + +def test_child_inherits_content_policy() -> None: + logger, _ = _audit_logger(content_policy="hash") + + child = logger.child("sub") + + assert child.content_policy == "hash" + + +def test_kwargs_are_forwarded_to_logger_init() -> None: + logger, _ = _audit_logger(flush_at_exit=False) + + from logquill import shutdown + + assert logger not in shutdown._loggers diff --git a/tests/test_cli_verify.py b/tests/test_cli_verify.py new file mode 100644 index 0000000..dae7738 --- /dev/null +++ b/tests/test_cli_verify.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +import io +from pathlib import Path + +from logquill import AuditLogger +from logquill.cli import _run_verify, build_parser, main +from logquill.transports.file_transport import FileTransport + + +def _write_an_audit_log(path: Path) -> AuditLogger: + logger = AuditLogger("audit", transports=[FileTransport(path)]) + logger.info("login", user_id=1) + logger.info("logout", user_id=1) + return logger + + +def _verify(file: str, *extra_args: str) -> tuple[str, int]: + args = build_parser().parse_args(["verify", file, *extra_args]) + out = io.StringIO() + exit_code = _run_verify(args, out=out) + return out.getvalue(), exit_code + + +def test_verify_passes_on_an_untampered_file(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + _write_an_audit_log(path).close() + + output, exit_code = _verify(str(path)) + + assert exit_code == 0 + assert "OK" in output + assert "2 record(s)" in output + assert "chain head is" in output + + +def test_verify_fails_with_an_actionable_message_on_a_tampered_file(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + _write_an_audit_log(path).close() + lines = path.read_text().splitlines() + lines[0] = lines[0].replace("login", "EDITED") + path.write_text("\n".join(lines) + "\n") + + output, exit_code = _verify(str(path)) + + assert exit_code == 1 + assert "FAILED" in output + assert "edited after being chained" in output + assert "record 1" in output + + +def test_verify_reports_a_helpful_error_for_a_missing_file(tmp_path: Path) -> None: + output, exit_code = _verify(str(tmp_path / "missing.log")) + + assert exit_code == 1 + assert "no such file" in output + + +def test_verify_with_a_matching_signature_passes(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + logger = _write_an_audit_log(path) + signature = logger.sign_head(bytes.fromhex("6b31")) + logger.close() + + output, exit_code = _verify(str(path), "--sign-key", "6b31", "--signature", signature) + + assert exit_code == 0 + assert "OK" in output + + +def test_verify_with_a_wrong_signature_fails(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + _write_an_audit_log(path).close() + + output, exit_code = _verify(str(path), "--sign-key", "6b31", "--signature", "a" * 64) + + assert exit_code == 1 + assert "FAILED" in output + assert "signature" in output + + +def test_verify_catches_a_truncated_file_with_the_right_signature(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + logger = _write_an_audit_log(path) + signature = logger.sign_head(bytes.fromhex("6b31")) + logger.close() + lines = path.read_text().splitlines() + path.write_text(lines[0] + "\n") # drop the last (signed) line + + output, exit_code = _verify(str(path), "--sign-key", "6b31", "--signature", signature) + + assert exit_code == 1 + assert "truncated" in output + + +def test_sign_key_and_signature_must_be_given_together(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + _write_an_audit_log(path).close() + + output, exit_code = _verify(str(path), "--sign-key", "6b31") + + assert exit_code == 1 + assert "together" in output + + +def test_a_non_hex_sign_key_is_rejected(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + _write_an_audit_log(path).close() + + output, exit_code = _verify(str(path), "--sign-key", "not-hex!", "--signature", "a" * 64) + + assert exit_code == 1 + assert "hex" in output + + +def test_verify_via_main_entry_point(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + _write_an_audit_log(path).close() + + assert main(["verify", str(path)]) == 0 + + +def test_verify_on_a_file_with_no_hash_at_all(tmp_path: Path) -> None: + from logquill import Logger + + path = tmp_path / "plain.log" + plain = Logger("app", transports=[FileTransport(path)], flush_at_exit=False) + plain.info("hello") # never chained — no TamperEvidentPlugin attached + plain.close() + + output, exit_code = _verify(str(path)) + + assert exit_code == 1 + assert "never chained" in output diff --git a/tests/test_otlp_transport.py b/tests/test_otlp_transport.py index 2ec9af3..a8a58c1 100644 --- a/tests/test_otlp_transport.py +++ b/tests/test_otlp_transport.py @@ -24,7 +24,16 @@ def _setup(**kwargs: Any) -> tuple[Logger, OTLPTransport, InMemorySpanExporter]: exporter = InMemorySpanExporter() transport = OTLPTransport(span_exporter=exporter, processor="simple", **kwargs) - logger = Logger("app.agent", level="TRACE", transports=[transport], plugins=[RunPlugin()]) + # content_policy="full": these tests are about what OTLPTransport does + # with whatever is in meta, not the Logger-level privacy policy that + # governs content fields by default (see tests/test_privacy.py). + logger = Logger( + "app.agent", + level="TRACE", + transports=[transport], + plugins=[RunPlugin()], + content_policy="full", + ) return logger, transport, exporter diff --git a/tests/test_plugins/test_pii_redact_plugin.py b/tests/test_plugins/test_pii_redact_plugin.py index b18fae0..ec68688 100644 --- a/tests/test_plugins/test_pii_redact_plugin.py +++ b/tests/test_plugins/test_pii_redact_plugin.py @@ -142,3 +142,103 @@ def test_use_presidio_routes_text_through_the_analyzer_and_anonymizer( fake_anonymizer_engine.return_value.anonymize.assert_called_once_with( text="jane@example.com", analyzer_results="fake-analysis" ) + + +# Deliberately low-entropy, dictionary-word placeholders ("FAKE" repeated to +# the length each pattern requires) rather than realistic-looking random +# characters — these exist only to exercise each pattern's shape, but a +# convincing-looking fake still matches real secret-scanning services' +# *own* shape-based detectors just as well as it matches ours (GitHub's push +# protection flagged an earlier, more realistic-looking version of this +# exact test as a live Slack token). Keeping these obviously inert avoids +# that false positive without weakening what's actually under test. +@pytest.mark.parametrize( + "name, text, leaked_fragment", + [ + ("openai", "key is sk-proj-FAKEFAKEFAKEFAKEFAKE", "FAKEFAKEFAKEFAKEFAKE"), + ("anthropic", "key: sk-ant-FAKEFAKEFAKEFAKEFAKE", "FAKEFAKEFAKEFAKEFAKE"), + ("aws", "id AKIAFAKEFAKEFAKEFAKE in use", "FAKEFAKEFAKEFAKE"), + ( + "github", + "token ghp_FAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKE", + "FAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKE", + ), + ("slack", "webhook xoxb-FAKEFAKEFAKE ok", "FAKEFAKEFAKE"), + ( + "jwt", + "Authorization: eyJFAKEFAKEFAKE.FAKEFAKEFAKEFAKE.FAKEFAKEFAKEFAKE", + "FAKEFAKEFAKEFAKE", + ), + ( + "bearer", + "Authorization: Bearer FAKEFAKEFAKEFAKEFAKE", + "FAKEFAKEFAKEFAKEFAKE", + ), + ], +) +def test_redacts_known_secret_shapes(name: str, text: str, leaked_fragment: str) -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info("log line", note=text) + + assert record is not None + assert leaked_fragment not in record["meta"]["note"], f"{name}: secret material leaked" + assert "***" in record["meta"]["note"] + + +def test_redacts_a_full_pem_private_key_block_body_and_all() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + pem = ( + "preamble -----BEGIN RSA PRIVATE KEY-----\n" + "MIIBogIBAAJ...\nMoreBase64KeyMaterialHere==\n" + "-----END RSA PRIVATE KEY-----\n trailer" + ) + + record = logger.info("log line", note=pem) + + assert record is not None + assert "MIIBogIBAAJ" not in record["meta"]["note"] + assert "MoreBase64KeyMaterialHere" not in record["meta"]["note"] + assert "preamble" in record["meta"]["note"] and "trailer" in record["meta"]["note"] + + +def test_an_earlier_patterns_replacement_does_not_corrupt_a_later_overlapping_match() -> None: + # a phone-shaped digit run embedded inside a Slack-token-shaped value: an + # in-place, sequential substitution would redact the digits first and + # leave the token only half-redacted — this must come out fully redacted + # instead. (Low-entropy "FAKE" padding, not a realistic token — see the + # parametrized test above for why.) + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info("log line", note="call xoxb-5551234567FAKE or 415-555-0199") + + assert record is not None + assert "xoxb" not in record["meta"]["note"] + assert "5551234567FAKE" not in record["meta"]["note"] + assert "415-555-0199" not in record["meta"]["note"] + assert record["meta"]["note"] == "call *** or ***" + + +def test_two_patterns_matching_the_same_span_redact_it_exactly_once() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + # a 16-digit run matches both `credit_card` and part of a longer secret + # shape isn't realistic here, so instead assert the simpler invariant: + # overlapping matches never produce doubled-up replacement text. + record = logger.info("log line", note="4242 4242 4242 4242 4242 4242 4242 4242") + + assert record is not None + assert record["meta"]["note"].count("***") <= 2 + + +def test_ordinary_text_is_never_falsely_flagged_as_a_secret() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info( + "log line", note="just a normal sentence about sk8boarding and github repositories" + ) + + assert record is not None + assert ( + record["meta"]["note"] == "just a normal sentence about sk8boarding and github repositories" + ) diff --git a/tests/test_plugins/test_redact_plugin.py b/tests/test_plugins/test_redact_plugin.py index cce9d4c..39c02ca 100644 --- a/tests/test_plugins/test_redact_plugin.py +++ b/tests/test_plugins/test_redact_plugin.py @@ -29,3 +29,45 @@ def test_custom_keys_and_replacement() -> None: assert record is not None assert record["meta"]["ssn"] == "[REDACTED]" assert record["meta"]["password"] == "not redacted here" + + +def test_classes_adds_every_key_tagged_with_that_class() -> None: + from logquill.privacy import keys_in_class + + plugin = RedactPlugin(keys=(), classes=["secret"]) + + assert plugin.keys == {key.lower() for key in keys_in_class("secret")} + + +def test_classes_combines_with_explicit_keys() -> None: + logger = Logger("app.test", plugins=[RedactPlugin(keys=["my_custom_key"], classes=["secret"])]) + + record = logger.info("x", my_custom_key="x", password="y", other="z") + + assert record is not None + assert record["meta"] == {"my_custom_key": "***", "password": "***", "other": "z"} + + +def test_content_classed_redaction_strips_prompt_fields_too() -> None: + # content_policy="full": isolates RedactPlugin's own classes=["content"] + # redaction from Logger's own content-capture policy, which would + # otherwise mask meta.input_messages first regardless of this plugin. + logger = Logger( + "app.test", + plugins=[RedactPlugin(keys=(), classes=["content"])], + content_policy="full", + ) + + record = logger.info("x", input_messages=["raw"], other=1) + + assert record is not None + assert record["meta"] == {"input_messages": "***", "other": 1} + + +def test_default_redacted_keys_still_matches_the_original_fixed_set() -> None: + from logquill.plugins.redact_plugin import DEFAULT_REDACTED_KEYS + + assert ( + frozenset({"password", "token", "secret", "api_key", "authorization"}) + == DEFAULT_REDACTED_KEYS + ) diff --git a/tests/test_plugins/test_tamper_evident_plugin.py b/tests/test_plugins/test_tamper_evident_plugin.py index 18ebfdd..e5fff1c 100644 --- a/tests/test_plugins/test_tamper_evident_plugin.py +++ b/tests/test_plugins/test_tamper_evident_plugin.py @@ -4,10 +4,12 @@ from logquill.logger import Logger from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin +from logquill.transports.transport import CollectingTransport def test_each_record_gets_a_hash_and_prev_hash() -> None: - logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) record = logger.info("hello") @@ -17,7 +19,8 @@ def test_each_record_gets_a_hash_and_prev_hash() -> None: def test_chain_links_consecutive_records() -> None: - logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) first = logger.info("one") second = logger.info("two") @@ -27,14 +30,16 @@ def test_chain_links_consecutive_records() -> None: def test_verify_chain_passes_on_an_untampered_log() -> None: - logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) records = [logger.info(f"event {i}", n=i) for i in range(5)] assert TamperEvidentPlugin.verify_chain(records) is True # type: ignore[arg-type] def test_verify_chain_detects_an_edited_message() -> None: - logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) records = [logger.info(f"event {i}", n=i) for i in range(5)] tampered = [copy.deepcopy(r) for r in records] tampered[2]["message"] = "edited after the fact" # type: ignore[index] @@ -43,7 +48,8 @@ def test_verify_chain_detects_an_edited_message() -> None: def test_verify_chain_detects_a_removed_record() -> None: - logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) records = [logger.info(f"event {i}", n=i) for i in range(5)] tampered = records[:2] + records[3:] # remove index 2 @@ -51,7 +57,8 @@ def test_verify_chain_detects_a_removed_record() -> None: def test_verify_chain_detects_reordered_records() -> None: - logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) records = [logger.info(f"event {i}", n=i) for i in range(3)] reordered = [records[1], records[0], records[2]] @@ -60,3 +67,194 @@ def test_verify_chain_detects_reordered_records() -> None: def test_verify_chain_on_empty_input_is_true() -> None: assert TamperEvidentPlugin.verify_chain([]) is True + + +# --- head_hash / sign_head --------------------------------------------------- + + +def test_head_hash_starts_at_genesis_and_tracks_the_last_record() -> None: + plugin = TamperEvidentPlugin() + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[plugin]) + + assert plugin.head_hash == "0" * 64 + + record = logger.info("one") + + assert record is not None + assert plugin.head_hash == record["meta"]["hash"] + + +def test_sign_head_is_deterministic_for_the_same_key_and_state() -> None: + plugin = TamperEvidentPlugin() + sink = CollectingTransport() + Logger("app.test", transports=[sink], plugins=[plugin]).info("one") + + assert plugin.sign_head(b"key1") == plugin.sign_head(b"key1") + assert plugin.sign_head(b"key1") != plugin.sign_head(b"key2") + + +def test_sign_head_changes_as_the_chain_grows() -> None: + plugin = TamperEvidentPlugin() + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[plugin]) + logger.info("one") + sig_after_one = plugin.sign_head(b"key") + logger.info("two") + + assert plugin.sign_head(b"key") != sig_after_one + + +# --- verify_chain_detailed --------------------------------------------------- + + +def test_verify_chain_detailed_reports_ok_with_the_head_hash() -> None: + from logquill.plugins.tamper_evident_plugin import verify_chain_detailed + + plugin = TamperEvidentPlugin() + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[plugin]) + logger.info("one") + logger.info("two") + + result = verify_chain_detailed(sink.records) + + assert result.ok is True + assert result.records_checked == 2 + assert result.broken_at is None + assert result.head_hash == plugin.head_hash + + +def test_verify_chain_detailed_reports_where_an_edit_broke_the_chain() -> None: + from logquill.plugins.tamper_evident_plugin import verify_chain_detailed + + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) + logger.info("one") + logger.info("two") + logger.info("three") + tampered = copy.deepcopy(sink.records) + tampered[1]["message"] = "EDITED" + + result = verify_chain_detailed(tampered) + + assert result.ok is False + assert result.broken_at == 2 + assert "edited after being chained" in result.reason + + +def test_verify_chain_detailed_reports_a_removed_line() -> None: + from logquill.plugins.tamper_evident_plugin import verify_chain_detailed + + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) + logger.info("one") + logger.info("two") + logger.info("three") + records = sink.records + with_a_line_removed = [records[0], records[2]] + + result = verify_chain_detailed(with_a_line_removed) + + assert result.ok is False + assert result.broken_at == 2 + assert "removed" in result.reason or "edited" in result.reason + + +def test_verify_chain_detailed_on_an_empty_log() -> None: + from logquill.plugins.tamper_evident_plugin import verify_chain_detailed + + result = verify_chain_detailed([]) + + assert result.ok is True + assert result.records_checked == 0 + assert result.head_hash == TamperEvidentPlugin().head_hash # genesis hash + + +def test_verify_chain_still_returns_a_bare_bool_for_existing_callers() -> None: + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[TamperEvidentPlugin()]) + logger.info("one") + + assert TamperEvidentPlugin.verify_chain(sink.records) is True + + +# --- verify_signed_chain / sign_head / verify_head_signature ----------------- + + +def test_verify_signed_chain_passes_with_the_right_key_and_signature() -> None: + from logquill.plugins.tamper_evident_plugin import verify_signed_chain + + plugin = TamperEvidentPlugin() + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[plugin]) + logger.info("one") + logger.info("two") + signature = plugin.sign_head(b"shared-key") + + result = verify_signed_chain(sink.records, key=b"shared-key", signature=signature) + + assert result.ok is True + + +def test_verify_signed_chain_fails_on_the_wrong_key() -> None: + from logquill.plugins.tamper_evident_plugin import verify_signed_chain + + plugin = TamperEvidentPlugin() + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[plugin]) + logger.info("one") + signature = plugin.sign_head(b"real-key") + + result = verify_signed_chain(sink.records, key=b"wrong-key", signature=signature) + + assert result.ok is False + assert "signature" in result.reason + + +def test_verify_signed_chain_catches_a_deleted_trailing_line() -> None: + """The exit criterion: hash-chaining alone can't catch a truncated file + — the remaining chain is perfectly self-consistent, just shorter. A + signature taken before the deletion catches it.""" + from logquill.plugins.tamper_evident_plugin import verify_chain_detailed, verify_signed_chain + + plugin = TamperEvidentPlugin() + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[plugin]) + logger.info("one") + logger.info("two") + logger.info("three") + signature = plugin.sign_head(b"key") # signed once all three are in + + truncated = sink.records[:-1] # an attacker then deletes "three" from the file + + assert verify_chain_detailed(truncated).ok is True # internally consistent! + result = verify_signed_chain(truncated, key=b"key", signature=signature) + assert result.ok is False + assert "truncated" in result.reason + + +def test_verify_signed_chain_passes_against_the_full_un_truncated_chain() -> None: + from logquill.plugins.tamper_evident_plugin import verify_signed_chain + + plugin = TamperEvidentPlugin() + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink], plugins=[plugin]) + logger.info("one") + logger.info("two") + logger.info("three") + signature = plugin.sign_head(b"key") + + result = verify_signed_chain(sink.records, key=b"key", signature=signature) + + assert result.ok is True + + +def test_verify_head_signature_uses_constant_time_comparison() -> None: + from logquill.plugins.tamper_evident_plugin import sign_head, verify_head_signature + + signature = sign_head("a" * 64, b"key") + + assert verify_head_signature("a" * 64, b"key", signature) is True + assert verify_head_signature("a" * 64, b"key", "wrong") is False + assert verify_head_signature("a" * 64, b"other-key", signature) is False diff --git a/tests/test_privacy.py b/tests/test_privacy.py new file mode 100644 index 0000000..3b0df6a --- /dev/null +++ b/tests/test_privacy.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +import pytest + +from logquill import Logger +from logquill.privacy import ( + CONTENT_FIELDS, + FIELD_CLASSES, + apply_content_policy, + keys_in_class, + parse_content_policy, +) +from logquill.transports.transport import CollectingTransport + + +def test_keys_in_class_returns_every_key_tagged_with_that_class() -> None: + secrets = keys_in_class("secret") + + assert secrets == {"password", "token", "secret", "api_key", "authorization"} + assert all(FIELD_CLASSES[key] == "secret" for key in secrets) + + +def test_content_fields_matches_the_registrys_content_class() -> None: + assert set(CONTENT_FIELDS) == keys_in_class("content") + assert "input_messages" in CONTENT_FIELDS + assert "state_diff" in CONTENT_FIELDS + + +def test_parse_content_policy_accepts_the_four_valid_choices() -> None: + for policy in ("off", "hash", "truncate", "full"): + assert parse_content_policy(policy) == policy + + +def test_parse_content_policy_rejects_anything_else() -> None: + with pytest.raises(ValueError, match="off.*hash.*truncate.*full"): + parse_content_policy("plaintext") + + +# --- apply_content_policy --------------------------------------------------- + + +def test_full_policy_is_a_true_no_op() -> None: + meta = {"input_messages": ["raw"], "other": 1} + + apply_content_policy(meta, "full") + + assert meta == {"input_messages": ["raw"], "other": 1} + + +def test_off_policy_replaces_content_fields_with_a_placeholder() -> None: + meta = {"input_messages": ["raw prompt"], "other": 1} + + apply_content_policy(meta, "off") + + assert "raw prompt" not in str(meta["input_messages"]) + assert "content capture is off" in meta["input_messages"] + assert meta["other"] == 1 + + +def test_off_policy_only_touches_classified_content_fields() -> None: + meta = {"password": "hunter2", "note": "whatever"} + + apply_content_policy(meta, "off") + + assert meta == {"password": "hunter2", "note": "whatever"} + + +def test_hash_policy_is_stable_and_never_shows_the_raw_value() -> None: + meta_a = {"input_messages": [{"role": "user", "content": "secret prompt"}]} + meta_b = {"input_messages": [{"role": "user", "content": "secret prompt"}]} + + apply_content_policy(meta_a, "hash") + apply_content_policy(meta_b, "hash") + + assert meta_a["input_messages"].startswith("sha256:") + assert "secret prompt" not in meta_a["input_messages"] + assert meta_a["input_messages"] == meta_b["input_messages"] # same content, same hash + + +def test_hash_policy_differs_for_different_content() -> None: + meta_a = {"input_messages": "one"} + meta_b = {"input_messages": "two"} + + apply_content_policy(meta_a, "hash") + apply_content_policy(meta_b, "hash") + + assert meta_a["input_messages"] != meta_b["input_messages"] + + +def test_truncate_policy_keeps_short_values_whole() -> None: + meta = {"tool_arguments": {"q": "short"}} + + apply_content_policy(meta, "truncate") + + assert meta["tool_arguments"] == '{"q": "short"}' or "short" in meta["tool_arguments"] + + +def test_truncate_policy_cuts_long_values_with_a_marker() -> None: + meta = {"output_messages": "x" * 5000} + + apply_content_policy(meta, "truncate") + + assert len(meta["output_messages"]) < 5000 + assert "truncated" in meta["output_messages"] + assert "5002" in meta["output_messages"] or "5000" in meta["output_messages"] + + +def test_a_circular_reference_falls_back_to_str_instead_of_raising() -> None: + cyclic: dict = {"a": 1} + cyclic["self"] = cyclic + meta = {"state_diff": cyclic} + + apply_content_policy(meta, "hash") # must not raise + + assert meta["state_diff"].startswith("sha256:") + + +def test_apply_content_policy_accepts_a_custom_field_list() -> None: + meta = {"my_custom_field": "secret-ish", "input_messages": "untouched"} + + apply_content_policy(meta, "off", fields=("my_custom_field",)) + + assert "content capture is off" in meta["my_custom_field"] + assert meta["input_messages"] == "untouched" # not in the custom field list + + +# --- integration through Logger --------------------------------------------- + + +def test_the_default_content_policy_is_off() -> None: + assert Logger("app").content_policy == "off" + + +def test_a_seeded_secret_inside_a_prompt_never_reaches_any_transport_by_default() -> None: + sink = CollectingTransport() + logger = Logger("app.agent", transports=[sink]) + + logger.llm_call( + "chat", + model="m", + input_messages=[ + { + "role": "user", + "parts": [{"type": "text", "content": "sk-ant-leaked-secret-key-material-here"}], + } + ], + ) + + assert "sk-ant-leaked-secret-key-material-here" not in str(sink.records[0]) + + +def test_content_policy_full_passes_content_through() -> None: + sink = CollectingTransport() + logger = Logger("app.agent", transports=[sink], content_policy="full") + + logger.llm_call("chat", model="m", input_messages=["raw prompt"]) + + assert sink.records[0]["meta"]["input_messages"] == ["raw prompt"] + + +def test_content_policy_applies_before_any_plugin_sees_the_record() -> None: + from logquill.plugins.plugin import Plugin + + seen = {} + + class Spy(Plugin): + def before_log(self, record): # type: ignore[no-untyped-def] + seen["input_messages"] = record["meta"].get("input_messages") + return record + + logger = Logger("app.agent", plugins=[Spy()]) # default content_policy="off" + + logger.llm_call("chat", model="m", input_messages=["raw prompt"]) + + assert seen["input_messages"] != ["raw prompt"] + assert "content capture is off" in seen["input_messages"] + + +def test_content_policy_covers_span_captured_state_too() -> None: + sink = CollectingTransport() + logger = Logger("app.agent", transports=[sink]) + state = {"sensitive": "secret-value"} + + with logger.span("work", capture_state=lambda: state): + state["sensitive"] = "changed" + + assert "secret-value" not in str(sink.records[0]) + assert "changed" not in str(sink.records[0]) + + +def test_child_loggers_inherit_the_parent_content_policy() -> None: + parent = Logger("app", content_policy="hash") + + child = parent.child("sub") + + assert child.content_policy == "hash" + + +def test_an_invalid_content_policy_raises_at_construction() -> None: + with pytest.raises(ValueError, match="content_policy"): + Logger("app", content_policy="plaintext") diff --git a/tests/test_state_diff.py b/tests/test_state_diff.py index 581d299..0d49e58 100644 --- a/tests/test_state_diff.py +++ b/tests/test_state_diff.py @@ -8,7 +8,10 @@ def _logger() -> tuple[Logger, CollectingTransport]: sink = CollectingTransport() - return Logger("app.agent", transports=[sink]), sink + # content_policy="full": these tests are about capture_state's diffing + # logic, not the privacy policy that governs meta.state_diff's visibility + # by default (see tests/test_privacy.py for that). + return Logger("app.agent", transports=[sink], content_policy="full"), sink def _span_meta(sink: CollectingTransport) -> dict[str, Any]: