diff --git a/CHANGELOG.md b/CHANGELOG.md index acdeceb..d9ddcd2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,65 @@ All notable changes to this project are documented in this file. ## Unreleased +## 0.3.0 - 2026-08-31 + +- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based + elevation — with `transports=` set, a record that would be dropped is + buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of + discarded outright, and if any later record in that trace reaches + `elevate_at` (default `ERROR`), the whole trace — every buffered record + plus everything after — ships, flushed straight to `transports`. Buffering + is bounded by `max_buffered_records` and `max_traces`, oldest trace + evicted first. Without `transports`, behavior is unchanged from plain + rate-based sampling. +- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a + plain function alongside a `Plugin` instance — wrapped internally as an + anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style + transform doesn't require subclassing `Plugin` first. +- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values** + (emails, SSNs, credit-card numbers, phone numbers), recursing through + nested dicts/lists/tuples and matching regardless of which key holds the + value — complements `RedactPlugin`'s exact-key matching. Depth- and + cycle-bounded, so a circular reference or pathologically deep structure + can't hang or crash the caller. An opt-in `use_presidio=True` mode + (`pip install logquill[presidio]`) routes values through Microsoft + Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio + is imported lazily and stays a real, non-default dependency. +- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the + record's own content plus the previous record's `meta.hash`, stored as + `meta.prev_hash`), so editing, removing, or reordering a line in a + written log breaks the chain from that point on. Ships with a static + `TamperEvidentPlugin.verify_chain(records)` to check a log after the + fact. Opt-in — hashing every record has a real, measurable CPU cost. +- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`, + and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable + `threshold`), with the actual send always running on a background + thread so a slow or unreachable destination can never block the log call + that triggered it. Repeated identical errors (same level + logger + + message by default, or a custom `dedupe_key`) within + `dedupe_window_seconds` collapse into one follow-up alert carrying an + occurrence count instead of spamming the destination once per record. + `send_alert` failures are caught and routed to the plugin's own + `on_error`, same as any other plugin hook. Tracking is bounded to + `max_tracked_keys` concurrent dedupe windows — alerting degrades under + extreme cardinality, logging itself never does. All three concrete + plugins use only the stdlib (`urllib`, `smtplib`) — no new required + dependency. +- Fixed a pre-existing gap surfaced by a new property-based test (see + below): `Logger`'s per-transport dispatch had no error handling, so a + transport that failed to format or write a given record (e.g. + `JSONFormatter` on a `meta` value containing a circular reference) would + propagate the exception straight to the caller. Now caught and logged via + the same `logging.getLogger("logquill")` channel `BatchingTransport` + already uses, per transport, so one broken transport can't crash the + caller or stop other attached transports from receiving the record. +- Added a `hypothesis`-based property test (new `dev` dependency) that + drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`, + `PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` — + deeply nested structures, unusual scalar types, non-JSON-serializable + values, and circular references — asserting the pipeline never crashes + the caller, only ever fails closed. + - New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`, `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`, `DynamoDBTransport`, `RedisTransport`), message queues diff --git a/README.md b/README.md index e4f9ee9..01dcb33 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ landed so far. - **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill) - **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport` - **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own -- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging +- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins)) - **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP - **Typed throughout** — `mypy --strict` clean on the public API - *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md` @@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can transform a record or return `None` to drop it, `after_log(record)` runs once it's been dispatched to every transport, and `on_error(exc, record)` catches anything a plugin's own hooks raise — a broken plugin can't take down logging. +Records are **not** deep-copied through the pipeline — a plugin receives and +may mutate the same dict every other plugin sees; copy it yourself in +`before_log` if you need to preserve the original. ```python from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin @@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2") # (unless this call was one of the ~90% sampling dropped, in which case it's None) ``` -Write your own by subclassing `Plugin`; override only the hooks you need. +Write your own by subclassing `Plugin`; override only the hooks you need. For +a one-off transform, skip the subclass entirely — `.use()` also accepts a +plain function, wrapped internally as an anonymous `Plugin`: + +```python +from logquill import Logger + +def strip_ssn(record): + record["meta"].pop("ssn", None) + return record # or None to drop the record + +logger = Logger("app") +logger.use(strip_ssn) +logger.info("submit", ssn="123-45-6789", user_id=42) +# meta: {'user_id': 42} +``` + +### Tail-based sampling elevation + +Plain `SamplingPlugin(rate)` drops records independently of each other. Add +`transports=` and every record's `meta["trace_id"]` (configurable via +`trace_key`) turns sampling tail-based instead: a dropped record is buffered +under its trace id rather than discarded, and if any later record in that +same trace reaches `elevate_at` (default `ERROR`), the whole trace — every +buffered record plus everything from then on — ships, flushed straight to +`transports`. A request that looked unremarkable when it started still +produces a complete trace once it turns out to have failed. + +```python +from logquill import CollectingTransport, Logger, SamplingPlugin + +sink = CollectingTransport() +sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest +logger = Logger("app", transports=[sink], plugins=[sampling]) + +logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer +logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer +logger.error("query timed out", trace_id="req-42") # elevates the whole trace + +assert [r["message"] for r in sink.records] == [ + "received request", + "queried database", + "query timed out", +] +``` + +Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest +buffered trace is evicted once either limit is hit, so a single +high-cardinality or long-lived trace can't grow memory without limit. + +### PII redaction by pattern, not just key + +`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by +scanning `meta` **values** — recursively through nested dicts/lists/tuples — +for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches +wherever they appear, regardless of which key holds them: + +```python +from logquill import Logger, PIIRedactPlugin + +logger = Logger("app", plugins=[PIIRedactPlugin()]) + +logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789") +# meta: {'notes': 'reach me at ***, ssn ***'} +``` + +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. + +### Tamper-evident logs + +`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers +its own content plus the previous record's hash — so editing, removing, or +reordering a line in a written log breaks the chain from that point on. +Opt-in, since hashing every record has a real CPU cost: + +```python +from logquill import Logger, TamperEvidentPlugin + +logger = Logger("app", plugins=[TamperEvidentPlugin()]) +records = [logger.info(f"step {i}") for i in range(3)] + +assert TamperEvidentPlugin.verify_chain(records) is True + +records[1]["message"] = "tampered" # simulate an edited log line +assert TamperEvidentPlugin.verify_chain(records) is False +``` + +### Alerting on errors + +`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL +(or any configurable `threshold`). It never blocks the log call that +triggered it — the actual send runs on a background thread — and repeated +identical errors within `dedupe_window_seconds` collapse into a single +follow-up alert carrying an occurrence count, instead of spamming the +destination once per record. Concrete subclasses ship for Slack, PagerDuty, +and email: + +```python +from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin + +logger = Logger( + "app", + plugins=[ + SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"), + PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"), + ], +) + +logger.error("payment webhook failed") # posts to the Slack webhook +logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL) +``` + +Write your own destination by subclassing `AlertingPlugin` and implementing +`send_alert(record, occurrences)`; thresholding, deduplication, and the +never-block-the-caller behavior are all handled by the base class. ## Development diff --git a/logquill/__init__.py b/logquill/__init__.py index 5dad776..d3799a7 100644 --- a/logquill/__init__.py +++ b/logquill/__init__.py @@ -1,10 +1,16 @@ from logquill.formatter import Formatter, JSONFormatter from logquill.levels import Level, parse_level from logquill.logger import Logger +from logquill.plugins.alerting_plugin import AlertingPlugin from logquill.plugins.context_plugin import ContextPlugin -from logquill.plugins.plugin import Plugin +from logquill.plugins.email_alert_plugin import EmailAlertPlugin +from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin +from logquill.plugins.pii_redact_plugin import PIIRedactPlugin +from logquill.plugins.plugin import FunctionPlugin, Plugin from logquill.plugins.redact_plugin import RedactPlugin 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.records import LogRecord from logquill.transports.batching_transport import BatchingTransport from logquill.transports.cloud.app_insights_transport import AppInsightsTransport @@ -30,9 +36,10 @@ from logquill.transports.sql.sqlite_transport import SQLiteTransport from logquill.transports.transport import CollectingTransport, Transport -__version__ = "0.2.1" +__version__ = "0.3.0" __all__ = [ + "AlertingPlugin", "AppInsightsTransport", "BaseQueueTransport", "BaseSQLTransport", @@ -45,8 +52,10 @@ "DatadogTransport", "DynamoDBTransport", "ElasticsearchTransport", + "EmailAlertPlugin", "FileTransport", "Formatter", + "FunctionPlugin", "HTTPTransport", "JSONFormatter", "KafkaTransport", @@ -56,6 +65,8 @@ "MongoDBTransport", "MySQLTransport", "NewRelicTransport", + "PIIRedactPlugin", + "PagerDutyAlertPlugin", "Plugin", "PostgresTransport", "PubSubTransport", @@ -66,6 +77,8 @@ "SQLiteTransport", "SQSTransport", "SamplingPlugin", + "SlackAlertPlugin", + "TamperEvidentPlugin", "Transport", "parse_level", "__version__", diff --git a/logquill/logger.py b/logquill/logger.py index 208fdf8..7f75c5b 100644 --- a/logquill/logger.py +++ b/logquill/logger.py @@ -1,13 +1,16 @@ from __future__ import annotations import contextlib +import logging from typing import Any from logquill.levels import Level, parse_level -from logquill.plugins.plugin import Plugin +from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin from logquill.records import LogRecord, create_record from logquill.transports.transport import Transport +_logger = logging.getLogger("logquill") + class Logger: def __init__( @@ -15,12 +18,14 @@ def __init__( name: str, level: int | str | Level = Level.INFO, transports: list[Transport] | None = None, - plugins: list[Plugin] | None = None, + plugins: list[Plugin | MiddlewareFunc] | None = None, ) -> None: self.name = name self._level = parse_level(level) self.transports: list[Transport] = list(transports) if transports else [] - self.plugins: list[Plugin] = list(plugins) if plugins else [] + self.plugins: list[Plugin] = [] + for plugin in plugins or []: + self.use(plugin) @property def level(self) -> Level: @@ -29,8 +34,16 @@ def level(self) -> Level: def set_level(self, level: int | str | Level) -> None: self._level = parse_level(level) - def use(self, plugin: Plugin) -> Logger: - """Register a plugin. Returns `self` so calls can be chained.""" + def use(self, plugin: Plugin | MiddlewareFunc) -> Logger: + """Register a plugin, or a plain `before_log`-style function. + + A function is wrapped internally as an anonymous `Plugin` + (`FunctionPlugin`) — the same middleware ergonomics as Express/Koa, + without needing to read the `Plugin` base class first. Returns + `self` so calls can be chained. + """ + if not isinstance(plugin, Plugin): + plugin = FunctionPlugin(plugin) self.plugins.append(plugin) return self @@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord | record = result for transport in self.transports: - transport.write(transport.format(record), record) + try: + transport.write(transport.format(record), record) + except Exception: + # a transport that can't format or write this particular record + # (e.g. a circular reference in `meta`) must not crash the caller + _logger.exception("%s: failed to write a log record", type(transport).__name__) for plugin in self.plugins: try: diff --git a/logquill/plugins/alerting_plugin.py b/logquill/plugins/alerting_plugin.py new file mode 100644 index 0000000..86b1a9f --- /dev/null +++ b/logquill/plugins/alerting_plugin.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import contextlib +import threading +from typing import Callable + +from logquill.levels import Level, parse_level +from logquill.plugins.plugin import Plugin +from logquill.records import LogRecord + + +class _Window: + __slots__ = ("record", "count", "timer") + + def __init__(self, record: LogRecord, timer: threading.Timer) -> None: + self.record = record + self.count = 1 + self.timer = timer + + +class AlertingPlugin(Plugin): + """Base class for plugins that fire an external alert on ERROR/FATAL (or + any configurable `threshold`). + + A concrete subclass implements only `send_alert(record, occurrences)` — + everything else (thresholding, deduplication, never blocking the + caller, never letting a broken destination crash logging) lives here. + + The first record at or above `threshold` for a given dedupe key (by + default: level + logger + message) fires `send_alert` right away, on a + short-lived background thread — so the log call that triggered it is + never blocked on a webhook, SMTP handshake, or any other I/O, even if + the destination is slow or unreachable. This stands in for the shared + async dispatch queue a later phase will introduce; once that queue + exists, `AlertingPlugin` can route through it instead of spawning its + own thread per alert. + + Any further record matching the same dedupe key within + `dedupe_window_seconds` of the first is *not* sent again — it just + increments a counter. When the window closes, if more than one record + matched, exactly one follow-up alert is sent with the total occurrence + count, instead of spamming the destination once per record. Tracking is + bounded to `max_tracked_keys` distinct concurrent dedupe keys; beyond + that, new keys are dropped rather than tracked (alerting degrades under + extreme cardinality, logging itself never does). + + `send_alert` is always called from a background thread and is wrapped + so an exception in it can't crash that thread or the caller — it's + routed to this plugin's own `on_error`, the same as any other plugin + hook that raises. + """ + + def __init__( + self, + *, + threshold: int | str | Level = Level.ERROR, + dedupe_window_seconds: float = 300.0, + dedupe_key: Callable[[LogRecord], str] | None = None, + max_tracked_keys: int = 500, + ) -> None: + self.threshold = parse_level(threshold) + self.dedupe_window_seconds = dedupe_window_seconds + self._dedupe_key = dedupe_key or self._default_dedupe_key + self.max_tracked_keys = max_tracked_keys + self._lock = threading.Lock() + self._windows: dict[str, _Window] = {} + + @staticmethod + def _default_dedupe_key(record: LogRecord) -> str: + return f"{record['level']}:{record['logger']}:{record['message']}" + + def after_log(self, record: LogRecord) -> None: + if Level[record["level"]] < self.threshold: + return + + key = self._dedupe_key(record) + with self._lock: + window = self._windows.get(key) + if window is not None: + window.count += 1 + return + + if len(self._windows) >= self.max_tracked_keys: + return + + timer = threading.Timer(self.dedupe_window_seconds, self._flush, args=(key,)) + timer.daemon = True + self._windows[key] = _Window(record, timer) + timer.start() + + threading.Thread(target=self._safe_send, args=(record, 1), daemon=True).start() + + def _flush(self, key: str) -> None: + with self._lock: + window = self._windows.pop(key, None) + if window is None or window.count <= 1: + return + self._safe_send(window.record, window.count) + + def _safe_send(self, record: LogRecord, occurrences: int) -> None: + try: + self.send_alert(record, occurrences) + except Exception as exc: + with contextlib.suppress(Exception): + self.on_error(exc, record) + + def send_alert(self, record: LogRecord, occurrences: int) -> None: + """Send one alert for `record`, representing `occurrences` collapsed + duplicates (1 on first occurrence; the deduped total on a follow-up + flush). Override in a concrete subclass — never call this directly, + `AlertingPlugin` calls it from a background thread. + """ + raise NotImplementedError + + def close(self) -> None: + """Cancel any pending dedupe-window timers. Call on logger shutdown.""" + with self._lock: + windows = list(self._windows.values()) + self._windows.clear() + for window in windows: + window.timer.cancel() diff --git a/logquill/plugins/email_alert_plugin.py b/logquill/plugins/email_alert_plugin.py new file mode 100644 index 0000000..8b081a4 --- /dev/null +++ b/logquill/plugins/email_alert_plugin.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import smtplib +from email.message import EmailMessage +from typing import Any + +from logquill.plugins.alerting_plugin import AlertingPlugin +from logquill.records import LogRecord + + +class EmailAlertPlugin(AlertingPlugin): + """Sends deduplicated `AlertingPlugin` alerts by email over SMTP. + + Uses stdlib `smtplib`/`email` — no extra dependency required. Set + `use_tls=False` for an SMTP server that doesn't support STARTTLS (e.g. + a local relay); `username`/`password` are only used if both are set. + """ + + def __init__( + self, + *, + smtp_host: str, + smtp_port: int, + from_addr: str, + to_addrs: list[str], + username: str | None = None, + password: str | None = None, + use_tls: bool = True, + timeout: float = 10.0, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) + self.smtp_host = smtp_host + self.smtp_port = smtp_port + self.from_addr = from_addr + self.to_addrs = to_addrs + self.username = username + self.password = password + self.use_tls = use_tls + self.timeout = timeout + + def send_alert(self, record: LogRecord, occurrences: int) -> None: + message = EmailMessage() + subject = f"[{record['level']}] {record['logger']}" + if occurrences > 1: + subject += f" (x{occurrences})" + message["Subject"] = subject + message["From"] = self.from_addr + message["To"] = ", ".join(self.to_addrs) + message.set_content( + f"{record['message']}\n\n" + f"occurrences: {occurrences}\n" + f"timestamp: {record['timestamp']}\n" + f"meta: {record['meta']!r}" + ) + + with smtplib.SMTP(self.smtp_host, self.smtp_port, timeout=self.timeout) as client: + if self.use_tls: + client.starttls() + if self.username and self.password: + client.login(self.username, self.password) + client.send_message(message) diff --git a/logquill/plugins/pagerduty_alert_plugin.py b/logquill/plugins/pagerduty_alert_plugin.py new file mode 100644 index 0000000..7d2884f --- /dev/null +++ b/logquill/plugins/pagerduty_alert_plugin.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import json +import urllib.request +from typing import Any + +from logquill.plugins.alerting_plugin import AlertingPlugin +from logquill.records import LogRecord + +_ENDPOINT = "https://events.pagerduty.com/v2/enqueue" +_SEVERITY = {"ERROR": "error", "FATAL": "critical"} + + +class PagerDutyAlertPlugin(AlertingPlugin): + """Sends deduplicated `AlertingPlugin` alerts to PagerDuty via the + Events API v2 (`POST https://events.pagerduty.com/v2/enqueue`). + + `routing_key` is an Events API v2 integration key from a PagerDuty + service. Uses stdlib `urllib` — no extra dependency required. + """ + + def __init__(self, routing_key: str, *, timeout: float = 5.0, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.routing_key = routing_key + self.timeout = timeout + + def send_alert(self, record: LogRecord, occurrences: int) -> None: + summary = f"{record['logger']}: {record['message']}" + if occurrences > 1: + summary += f" (x{occurrences})" + body = json.dumps( + { + "routing_key": self.routing_key, + "event_action": "trigger", + "payload": { + "summary": summary, + "severity": _SEVERITY.get(record["level"], "error"), + "source": record["logger"], + "timestamp": record["timestamp"], + "custom_details": {"occurrences": occurrences, **record["meta"]}, + }, + } + ).encode("utf-8") + request = urllib.request.Request( + _ENDPOINT, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=self.timeout) as response: + if response.status >= 400: + raise RuntimeError( + f"PagerDutyAlertPlugin: Events API returned HTTP {response.status} — " + "check the routing key is a valid Events API v2 integration key" + ) diff --git a/logquill/plugins/pii_redact_plugin.py b/logquill/plugins/pii_redact_plugin.py new file mode 100644 index 0000000..2b18625 --- /dev/null +++ b/logquill/plugins/pii_redact_plugin.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from typing import Any + +from logquill.plugins.plugin import Plugin +from logquill.records import LogRecord + +#: Syntactic (not semantic) patterns — matched on shape, so both false +#: 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. +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"), +} + +_MAX_DEPTH = 50 + + +class PIIRedactPlugin(Plugin): + """Regex-based PII redaction over `meta` **values**, not just keys. + + Complements `RedactPlugin`, which redacts by exact key match — + `PIIRedactPlugin` scans string values (recursively through nested + dicts/lists/tuples) for emails, SSNs, credit-card numbers, and phone + numbers, and redacts matches wherever they appear, regardless of which + key holds them (a `notes` field containing a stray SSN is still caught). + + Detection is pattern-based by default: fast and dependency-free, but it + matches on syntactic shape, not meaning — a random 9-digit number can + false-positive as an SSN, and anything that doesn't fit these shapes + (a name, a street address) is a false negative. Pass your own + `patterns={"custom": re.compile(...)}` to extend or replace the + defaults. + + For fuzzier, ML-based PII detection instead, pass `use_presidio=True` + (`pip install logquill[presidio]`) to run values through Microsoft + Presidio's `AnalyzerEngine`/`AnonymizerEngine`. Presidio is a real + dependency (spaCy models included) — it's opt-in and imported lazily, + so `PIIRedactPlugin` works with zero extra dependencies as long as + `use_presidio` stays `False`. + + Recursion into nested `meta` structures is depth- and cycle-bounded, so + a circular reference or a pathologically deep structure can't hang or + crash the caller — it's left unredacted past the bound rather than + raising. + """ + + def __init__( + self, + *, + patterns: Mapping[str, re.Pattern[str]] | None = None, + replacement: str = "***", + use_presidio: bool = False, + presidio_entities: Sequence[str] | None = None, + presidio_language: str = "en", + ) -> None: + self.patterns: dict[str, re.Pattern[str]] = ( + dict(patterns) if patterns is not None else dict(DEFAULT_PII_PATTERNS) + ) + self.replacement = replacement + self.use_presidio = use_presidio + self.presidio_entities = list(presidio_entities) if presidio_entities else None + self.presidio_language = presidio_language + self._analyzer: Any = None + self._anonymizer: Any = None + if use_presidio: + self._analyzer, self._anonymizer = self._load_presidio() + + @staticmethod + def _load_presidio() -> tuple[Any, Any]: + try: + # stub availability for this optional dep varies by env + from presidio_analyzer import AnalyzerEngine # type: ignore[import-not-found] + from presidio_anonymizer import AnonymizerEngine # type: ignore[import-not-found] + except ImportError as exc: + raise ImportError( + "PIIRedactPlugin(use_presidio=True) requires the optional Presidio " + "dependencies — install with `pip install logquill[presidio]`." + ) from exc + return AnalyzerEngine(), AnonymizerEngine() + + def before_log(self, record: LogRecord) -> LogRecord | None: + record["meta"] = self._redact_value(record["meta"], set(), 0) + return record + + def _redact_value(self, value: Any, seen: set[int], depth: int) -> Any: + if depth > _MAX_DEPTH: + return value + if isinstance(value, str): + return self._redact_text(value) + if isinstance(value, (dict, list, tuple)): + obj_id = id(value) + if obj_id in seen: + return value # circular reference — leave as-is rather than recurse forever + seen = seen | {obj_id} + if isinstance(value, dict): + return {k: self._redact_value(v, seen, depth + 1) for k, v in value.items()} + if isinstance(value, list): + return [self._redact_value(v, seen, depth + 1) for v in value] + return tuple(self._redact_value(v, seen, depth + 1) for v in value) + return value + + 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 + + def _redact_with_presidio(self, text: str) -> str: + results = self._analyzer.analyze( + text=text, language=self.presidio_language, entities=self.presidio_entities + ) + anonymized = self._anonymizer.anonymize(text=text, analyzer_results=results) + return str(anonymized.text) diff --git a/logquill/plugins/plugin.py b/logquill/plugins/plugin.py index c6d7494..a344e72 100644 --- a/logquill/plugins/plugin.py +++ b/logquill/plugins/plugin.py @@ -1,7 +1,11 @@ from __future__ import annotations +from typing import Callable + from logquill.records import LogRecord +MiddlewareFunc = Callable[[LogRecord], "LogRecord | None"] + class Plugin: """Base class for the plugin pipeline: `before_log`, `after_log`, `on_error`. @@ -20,3 +24,20 @@ def after_log(self, record: LogRecord) -> None: def on_error(self, exc: Exception, record: LogRecord) -> None: """Called when one of this plugin's own hooks raises.""" + + +class FunctionPlugin(Plugin): + """Wraps a plain `before_log`-style function as a `Plugin`. + + `Logger.use()` builds one of these automatically when given a function + instead of a `Plugin` subclass — a one-off transform shouldn't require + subclassing ceremony. There's no `next()` chaining: the pipeline is + already an ordered list of hooks the `Logger` calls in sequence, so this + is sugar for a single-method `Plugin`, not a new execution model. + """ + + def __init__(self, func: MiddlewareFunc) -> None: + self._func = func + + def before_log(self, record: LogRecord) -> LogRecord | None: + return self._func(record) diff --git a/logquill/plugins/sampling_plugin.py b/logquill/plugins/sampling_plugin.py index e0d087c..0195e3f 100644 --- a/logquill/plugins/sampling_plugin.py +++ b/logquill/plugins/sampling_plugin.py @@ -1,20 +1,115 @@ from __future__ import annotations import random +from collections import OrderedDict from typing import Callable +from logquill.levels import Level, parse_level from logquill.plugins.plugin import Plugin from logquill.records import LogRecord +from logquill.transports.transport import Transport class SamplingPlugin(Plugin): - """Keeps roughly `rate` of records (0.0-1.0), dropping the rest.""" + """Keeps roughly `rate` of records (0.0-1.0), dropping the rest. - def __init__(self, rate: float, rng: Callable[[], float] | None = None) -> None: + With `transports` set, sampling becomes tail-based per trace: a record + that would otherwise be dropped is buffered under its `meta[trace_key]` + value instead of discarded outright. If any later record sharing that + trace id reaches `elevate_at` or above, the whole trace is "elevated" — + every buffered record for that trace id is flushed straight to + `transports`, and every subsequent record for that trace id ships + unconditionally. This is what lets a sampled-out request still produce + a complete trace once it turns out to matter (it errored). + + Flushing writes buffered records directly to `transports` — pass the + same list given to the `Logger`. This bypasses `before_log`/`after_log`/ + `on_error` for any plugin *after* `SamplingPlugin` in the pipeline (the + plugins before it already ran, since that's how the buffered record was + built); put `SamplingPlugin` last if that matters for your pipeline. + + Without `transports`, tail-based elevation is inactive and this behaves + exactly like plain rate-based sampling (the original behavior) — a + record without `meta[trace_key]` is also just rate-sampled, since there's + no trace to buffer it under. + + Buffering is bounded: at most `max_buffered_records` records total and + `max_traces` distinct trace ids are held at once. Once either limit is + hit, the oldest buffered trace is evicted (and its records are lost, not + flushed) — a deliberate bounded-memory trade-off, not a bug: an + unbounded per-trace buffer would let a single pathologically long-lived + or high-cardinality trace grow memory without limit. + """ + + def __init__( + self, + rate: float, + *, + rng: Callable[[], float] | None = None, + trace_key: str = "trace_id", + elevate_at: int | str | Level = Level.ERROR, + transports: list[Transport] | None = None, + max_buffered_records: int = 1000, + max_traces: int = 200, + ) -> None: if not 0.0 <= rate <= 1.0: raise ValueError(f"rate must be between 0 and 1, got {rate!r}") self.rate = rate self._rng = rng or random.random + self.trace_key = trace_key + self.elevate_at = parse_level(elevate_at) + self.transports = transports + self.max_buffered_records = max_buffered_records + self.max_traces = max_traces + self._buffer: OrderedDict[object, list[LogRecord]] = OrderedDict() + self._buffered_count = 0 + self._elevated: OrderedDict[object, None] = OrderedDict() def before_log(self, record: LogRecord) -> LogRecord | None: - return record if self._rng() < self.rate else None + transports = self.transports + trace_id = record["meta"].get(self.trace_key) if transports is not None else None + + if trace_id is not None and trace_id in self._elevated: + return record + + keep = self._rng() < self.rate + reached_elevate_level = Level[record["level"]] >= self.elevate_at + + if trace_id is not None and reached_elevate_level: + assert transports is not None # trace_id is only ever set when transports is + self._elevate(trace_id, transports) + return record + + if keep: + return record + + if trace_id is not None: + self._buffer_record(trace_id, record) + + return None + + def _elevate(self, trace_id: object, transports: list[Transport]) -> None: + self._elevated[trace_id] = None + buffered = self._buffer.pop(trace_id, []) + self._buffered_count -= len(buffered) + for buffered_record in buffered: + for transport in transports: + transport.write(transport.format(buffered_record), buffered_record) + + def _buffer_record(self, trace_id: object, record: LogRecord) -> None: + if trace_id in self._buffer: + self._buffer.move_to_end(trace_id) + else: + if len(self._buffer) >= self.max_traces: + self._evict_oldest_trace() + self._buffer[trace_id] = [] + + self._buffer[trace_id].append(record) + self._buffered_count += 1 + + while self._buffered_count > self.max_buffered_records and self._buffer: + self._evict_oldest_trace() + + def _evict_oldest_trace(self) -> None: + _, oldest_records = self._buffer.popitem(last=False) + self._buffered_count -= len(oldest_records) diff --git a/logquill/plugins/slack_alert_plugin.py b/logquill/plugins/slack_alert_plugin.py new file mode 100644 index 0000000..157c68a --- /dev/null +++ b/logquill/plugins/slack_alert_plugin.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +import json +import urllib.request +from typing import Any + +from logquill.plugins.alerting_plugin import AlertingPlugin +from logquill.records import LogRecord + + +class SlackAlertPlugin(AlertingPlugin): + """Sends deduplicated `AlertingPlugin` alerts to a Slack incoming webhook. + + `webhook_url` is the full "Incoming Webhook" URL from Slack's app + config. Uses stdlib `urllib` — no extra dependency required. + """ + + def __init__(self, webhook_url: str, *, timeout: float = 5.0, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.webhook_url = webhook_url + self.timeout = timeout + + def send_alert(self, record: LogRecord, occurrences: int) -> None: + body = json.dumps({"text": _format_message(record, occurrences)}).encode("utf-8") + request = urllib.request.Request( + self.webhook_url, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=self.timeout) as response: + if response.status >= 400: + raise RuntimeError( + f"SlackAlertPlugin: webhook returned HTTP {response.status} — " + "check the webhook URL is still valid in Slack's app config" + ) + + +def _format_message(record: LogRecord, occurrences: int) -> str: + suffix = f" (x{occurrences})" if occurrences > 1 else "" + return f"[{record['level']}] {record['logger']}: {record['message']}{suffix}" diff --git a/logquill/plugins/tamper_evident_plugin.py b/logquill/plugins/tamper_evident_plugin.py new file mode 100644 index 0000000..010706b --- /dev/null +++ b/logquill/plugins/tamper_evident_plugin.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterable, Mapping +from typing import Any + +from logquill.plugins.plugin import Plugin +from logquill.records import LogRecord + +GENESIS_HASH = "0" * 64 + + +class TamperEvidentPlugin(Plugin): + """Hash-chains every record so tampering with a written log can be + detected after the fact. + + Each record gets `meta.hash` = a SHA-256 hex digest over the record's + own content plus the previous record's hash (`meta.prev_hash`) — the + same hash-chain construction used by tamper-evident/append-only logs: + editing or deleting any one line breaks every hash after it in the + chain, even if the tamperer edits the file directly and not through + this plugin. Opt-in — hashing every record has a real, measurable CPU + cost, so it isn't part of the default pipeline. + + 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. + """ + + def __init__(self, *, genesis_hash: str = GENESIS_HASH) -> None: + self._genesis_hash = genesis_hash + self._last_hash = genesis_hash + + def before_log(self, record: LogRecord) -> LogRecord | None: + prev_hash = self._last_hash + digest = _compute_hash(record, prev_hash) + record["meta"] = {**record["meta"], "prev_hash": prev_hash, "hash": digest} + self._last_hash = digest + return record + + @staticmethod + def verify_chain( + records: Iterable[Mapping[str, Any]], *, genesis_hash: str = GENESIS_HASH + ) -> bool: + """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). + """ + 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 + + +def _compute_hash(record: Mapping[str, Any], prev_hash: str) -> str: + meta = record.get("meta", {}) + payload = json.dumps( + { + "timestamp": record.get("timestamp"), + "level": record.get("level"), + "logger": record.get("logger"), + "message": record.get("message"), + "meta": {k: v for k, v in meta.items() if k not in ("hash", "prev_hash")}, + }, + sort_keys=True, + default=str, + ) + return hashlib.sha256(f"{prev_hash}{payload}".encode()).hexdigest() diff --git a/pyproject.toml b/pyproject.toml index e4e4ccf..951c367 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "logquill" -version = "0.2.1" +version = "0.3.0" description = "A structured, leveled logging framework with pluggable transports and a plugin pipeline." readme = "README.md" license = "MIT" @@ -48,12 +48,14 @@ rabbitmq = ["pika>=1.3"] pubsub = ["google-cloud-pubsub>=2.21"] gcp-logging = ["google-cloud-logging>=3.10"] aws = ["boto3>=1.34"] +presidio = ["presidio-analyzer>=2.2", "presidio-anonymizer>=2.2"] dev = [ "ruff>=0.6", "mypy>=1.11", "pytest>=8.0", "pytest-asyncio>=0.24", "pytest-cov>=5.0", + "hypothesis>=6.100", "build>=1.2", "twine>=5.1", ] diff --git a/tests/test_plugin_pipeline_properties.py b/tests/test_plugin_pipeline_properties.py new file mode 100644 index 0000000..3da43a1 --- /dev/null +++ b/tests/test_plugin_pipeline_properties.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +from typing import Any + +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from logquill.logger import Logger +from logquill.plugins.context_plugin import ContextPlugin +from logquill.plugins.pii_redact_plugin import PIIRedactPlugin +from logquill.plugins.redact_plugin import RedactPlugin +from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin +from logquill.transports.transport import CollectingTransport + +# Deliberately adversarial: deeply nested containers, unusual scalar types, +# and non-JSON-serializable values (a raw object, bytes). Circular +# references are exercised separately below, since hypothesis strategies +# can't easily generate them. +_scalars = st.one_of( + st.none(), + st.booleans(), + st.integers(), + st.floats(allow_nan=True, allow_infinity=True), + st.text(), + st.binary(), + st.builds(object), +) + +_meta_values = st.recursive( + _scalars, + lambda children: st.one_of( + st.lists(children, max_size=5), + st.dictionaries(st.text(min_size=1, max_size=10), children, max_size=5), + ), + max_leaves=25, +) + +_meta_dicts = st.dictionaries(st.text(min_size=1, max_size=10), _meta_values, max_size=8) + + +def _build_logger() -> tuple[Logger, CollectingTransport]: + sink = CollectingTransport() + logger = Logger( + "app.test", + transports=[sink], + plugins=[ + ContextPlugin(service="api"), + RedactPlugin(), + PIIRedactPlugin(), + TamperEvidentPlugin(), + ], + ) + return logger, sink + + +@settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow]) +@given(meta=_meta_dicts) +def test_pipeline_never_crashes_on_adversarial_meta(meta: dict[str, Any]) -> None: + logger, _sink = _build_logger() + + # Must not raise, regardless of what's inside `meta` — a plugin that + # can't handle a value fails closed (drops or leaves it untouched), + # it never crashes the caller. + logger.info("adversarial", **meta) + + +def test_pipeline_never_crashes_on_circular_reference() -> None: + # The plugin pipeline itself handles the cycle fine (PIIRedactPlugin's + # cycle guard, TamperEvidentPlugin's `default=str` fallback). JSON + # genuinely can't represent a cycle, so the transport's `format()` call + # legitimately fails here — the point of this test is that the failure + # is caught and logged rather than propagated to the caller. + logger, _sink = _build_logger() + cyclic: dict[str, Any] = {"a": 1} + cyclic["self"] = cyclic + + record = logger.info("cyclic", data=cyclic) + + assert record is not None + + +def test_pipeline_never_crashes_on_deeply_nested_meta() -> None: + logger, _sink = _build_logger() + nested: dict[str, Any] = {} + cursor = nested + for _ in range(500): + cursor["child"] = {} + cursor = cursor["child"] + + logger.info("deeply nested", data=nested) diff --git a/tests/test_plugins/test_alerting_plugin.py b/tests/test_plugins/test_alerting_plugin.py new file mode 100644 index 0000000..a9951e6 --- /dev/null +++ b/tests/test_plugins/test_alerting_plugin.py @@ -0,0 +1,143 @@ +from __future__ import annotations + +import threading +import time + +from logquill.logger import Logger +from logquill.plugins.alerting_plugin import AlertingPlugin +from logquill.records import LogRecord + +_POLL_TIMEOUT = 2.0 +_POLL_INTERVAL = 0.01 + + +def _wait_until(predicate: object, timeout: float = _POLL_TIMEOUT) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): # type: ignore[operator] + return True + time.sleep(_POLL_INTERVAL) + return predicate() # type: ignore[operator] + + +class RecordingAlertPlugin(AlertingPlugin): + def __init__(self, **kwargs: object) -> None: + super().__init__(**kwargs) # type: ignore[arg-type] + self.sent: list[tuple[LogRecord, int]] = [] + self._sent_lock = threading.Lock() + + def send_alert(self, record: LogRecord, occurrences: int) -> None: + with self._sent_lock: + self.sent.append((record, occurrences)) + + def sent_count(self) -> int: + with self._sent_lock: + return len(self.sent) + + +class BrokenAlertPlugin(AlertingPlugin): + def __init__(self, **kwargs: object) -> None: + super().__init__(**kwargs) # type: ignore[arg-type] + self.errors: list[Exception] = [] + + def send_alert(self, record: LogRecord, occurrences: int) -> None: + raise RuntimeError("destination unreachable") + + def on_error(self, exc: Exception, record: LogRecord) -> None: + self.errors.append(exc) + + +def test_error_level_record_fires_an_alert() -> None: + alerting = RecordingAlertPlugin(dedupe_window_seconds=60) + logger = Logger("app.test", plugins=[alerting]) + + logger.error("something broke") + + assert _wait_until(lambda: alerting.sent_count() >= 1) + record, occurrences = alerting.sent[0] + assert record["message"] == "something broke" + assert occurrences == 1 + + +def test_below_threshold_records_do_not_fire_an_alert() -> None: + alerting = RecordingAlertPlugin(dedupe_window_seconds=60) + logger = Logger("app.test", plugins=[alerting]) + + logger.info("just info") + logger.warn("just a warning") + time.sleep(0.1) + + assert alerting.sent_count() == 0 + + +def test_never_blocks_the_caller_even_when_destination_is_unreachable() -> None: + broken = BrokenAlertPlugin(dedupe_window_seconds=60) + logger = Logger("app.test", plugins=[broken]) + + record = logger.error("boom") # must return promptly, not hang or raise + + assert record is not None + assert _wait_until(lambda: len(broken.errors) >= 1) + assert isinstance(broken.errors[0], RuntimeError) + + +def test_duplicate_errors_within_window_collapse_into_one_followup_alert() -> None: + alerting = RecordingAlertPlugin(dedupe_window_seconds=0.1) + logger = Logger("app.test", plugins=[alerting]) + + for _ in range(5): + logger.error("repeated failure") + + # first occurrence sends immediately with occurrences=1 + assert _wait_until(lambda: alerting.sent_count() >= 1) + # once the window closes, exactly one follow-up alert reports the total + assert _wait_until(lambda: alerting.sent_count() >= 2, timeout=2.0) + time.sleep(0.2) + assert alerting.sent_count() == 2 + _, first_occurrences = alerting.sent[0] + _, followup_occurrences = alerting.sent[1] + assert first_occurrences == 1 + assert followup_occurrences == 5 + + +def test_a_single_occurrence_gets_no_followup_alert() -> None: + alerting = RecordingAlertPlugin(dedupe_window_seconds=0.05) + logger = Logger("app.test", plugins=[alerting]) + + logger.error("one-off failure") + + assert _wait_until(lambda: alerting.sent_count() >= 1) + time.sleep(0.2) # let the dedupe window close + assert alerting.sent_count() == 1 + + +def test_new_dedupe_keys_beyond_max_tracked_keys_are_dropped() -> None: + # Bounded memory: once max_tracked_keys concurrent dedupe windows are + # open, a new distinct key is dropped outright rather than tracked or + # sent — alerting degrades under extreme cardinality, logging itself + # never does. + alerting = RecordingAlertPlugin(dedupe_window_seconds=60, max_tracked_keys=1) + logger = Logger("app.test", plugins=[alerting]) + + logger.error("first distinct error") + assert _wait_until(lambda: alerting.sent_count() >= 1) + + # second, different error while the first key's window is still open + # (max_tracked_keys=1) — dropped, not sent + logger.error("second distinct error") + time.sleep(0.2) + + assert alerting.sent_count() == 1 + + +def test_close_cancels_pending_dedupe_timers() -> None: + alerting = RecordingAlertPlugin(dedupe_window_seconds=60) + logger = Logger("app.test", plugins=[alerting]) + + logger.error("first") + logger.error("first") # buffered as a pending follow-up + assert _wait_until(lambda: alerting.sent_count() >= 1) + + alerting.close() + + assert alerting._windows == {} diff --git a/tests/test_plugins/test_email_alert_plugin.py b/tests/test_plugins/test_email_alert_plugin.py new file mode 100644 index 0000000..c4e5a76 --- /dev/null +++ b/tests/test_plugins/test_email_alert_plugin.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from logquill.levels import Level +from logquill.plugins.email_alert_plugin import EmailAlertPlugin +from logquill.records import create_record + + +def _record(message: str = "boom") -> object: + return create_record(level=Level.ERROR, logger="app.test", message=message, meta={}) + + +def test_send_alert_sends_via_smtp_with_starttls_and_login() -> None: + plugin = EmailAlertPlugin( + smtp_host="smtp.example.com", + smtp_port=587, + from_addr="alerts@example.com", + to_addrs=["oncall@example.com"], + username="user", + password="pass", + ) + fake_client = MagicMock() + fake_client.__enter__.return_value = fake_client + + with patch("smtplib.SMTP", return_value=fake_client) as mock_smtp: + plugin.send_alert(_record(), 1) # type: ignore[arg-type] + + mock_smtp.assert_called_once_with("smtp.example.com", 587, timeout=10.0) + fake_client.starttls.assert_called_once() + fake_client.login.assert_called_once_with("user", "pass") + fake_client.send_message.assert_called_once() + sent_message = fake_client.send_message.call_args[0][0] + assert sent_message["To"] == "oncall@example.com" + assert "boom" in sent_message.get_content() + + +def test_send_alert_skips_login_without_credentials() -> None: + plugin = EmailAlertPlugin( + smtp_host="smtp.example.com", + smtp_port=25, + from_addr="alerts@example.com", + to_addrs=["oncall@example.com"], + use_tls=False, + ) + fake_client = MagicMock() + fake_client.__enter__.return_value = fake_client + + with patch("smtplib.SMTP", return_value=fake_client): + plugin.send_alert(_record(), 1) # type: ignore[arg-type] + + fake_client.starttls.assert_not_called() + fake_client.login.assert_not_called() + + +def test_subject_includes_occurrence_count_when_greater_than_one() -> None: + plugin = EmailAlertPlugin( + smtp_host="smtp.example.com", + smtp_port=25, + from_addr="alerts@example.com", + to_addrs=["oncall@example.com"], + ) + fake_client = MagicMock() + fake_client.__enter__.return_value = fake_client + + with patch("smtplib.SMTP", return_value=fake_client): + plugin.send_alert(_record(), 9) # type: ignore[arg-type] + + sent_message = fake_client.send_message.call_args[0][0] + assert "x9" in sent_message["Subject"] diff --git a/tests/test_plugins/test_pagerduty_alert_plugin.py b/tests/test_plugins/test_pagerduty_alert_plugin.py new file mode 100644 index 0000000..97331da --- /dev/null +++ b/tests/test_plugins/test_pagerduty_alert_plugin.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +from logquill.levels import Level +from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin +from logquill.records import create_record + + +def _record(level: Level = Level.ERROR, message: str = "boom") -> object: + return create_record(level=level, logger="app.test", message=message, meta={"user_id": 42}) + + +def test_send_alert_posts_events_api_v2_payload() -> None: + plugin = PagerDutyAlertPlugin("routing-key-123") + fake_response = MagicMock() + fake_response.status = 202 + fake_response.__enter__.return_value = fake_response + + with patch("urllib.request.urlopen", return_value=fake_response) as mock_urlopen: + plugin.send_alert(_record(), 1) # type: ignore[arg-type] + + request = mock_urlopen.call_args[0][0] + assert request.full_url == "https://events.pagerduty.com/v2/enqueue" + body = json.loads(request.data) + assert body["routing_key"] == "routing-key-123" + assert body["event_action"] == "trigger" + assert body["payload"]["severity"] == "error" + assert body["payload"]["custom_details"]["user_id"] == 42 + + +def test_summary_includes_occurrence_count_when_greater_than_one() -> None: + plugin = PagerDutyAlertPlugin("routing-key-123") + fake_response = MagicMock() + fake_response.status = 202 + fake_response.__enter__.return_value = fake_response + + with patch("urllib.request.urlopen", return_value=fake_response) as mock_urlopen: + plugin.send_alert(_record(), 6) # type: ignore[arg-type] + + body = json.loads(mock_urlopen.call_args[0][0].data) + assert "x6" in body["payload"]["summary"] + + +def test_fatal_level_maps_to_critical_severity() -> None: + plugin = PagerDutyAlertPlugin("routing-key-123") + fake_response = MagicMock() + fake_response.status = 202 + fake_response.__enter__.return_value = fake_response + + with patch("urllib.request.urlopen", return_value=fake_response) as mock_urlopen: + plugin.send_alert(_record(level=Level.FATAL), 1) # type: ignore[arg-type] + + body = json.loads(mock_urlopen.call_args[0][0].data) + assert body["payload"]["severity"] == "critical" + + +def test_send_alert_raises_on_http_error_status() -> None: + plugin = PagerDutyAlertPlugin("routing-key-123") + fake_response = MagicMock() + fake_response.status = 400 + fake_response.__enter__.return_value = fake_response + + with patch("urllib.request.urlopen", return_value=fake_response): + try: + plugin.send_alert(_record(), 1) # type: ignore[arg-type] + raised = False + except RuntimeError: + raised = True + + assert raised diff --git a/tests/test_plugins/test_pii_redact_plugin.py b/tests/test_plugins/test_pii_redact_plugin.py new file mode 100644 index 0000000..b18fae0 --- /dev/null +++ b/tests/test_plugins/test_pii_redact_plugin.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import re +import sys +import types +from unittest.mock import MagicMock + +import pytest + +from logquill.logger import Logger +from logquill.plugins.pii_redact_plugin import PIIRedactPlugin + + +def test_redacts_email_in_a_string_value() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info("signup", note="contact me at jane.doe@example.com please") + + assert record is not None + assert "jane.doe@example.com" not in record["meta"]["note"] + assert "***" in record["meta"]["note"] + + +def test_redacts_ssn_regardless_of_key_name() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info("free text field", notes="ssn on file: 123-45-6789") + + assert record is not None + assert "123-45-6789" not in record["meta"]["notes"] + + +def test_redacts_phone_number() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info("contact", note="call me at 415-555-0199") + + assert record is not None + assert "415-555-0199" not in record["meta"]["note"] + + +def test_non_string_values_are_left_untouched() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info("counts", user_id=42, active=True, ratio=0.5) + + assert record is not None + assert record["meta"] == {"user_id": 42, "active": True, "ratio": 0.5} + + +def test_redacts_credit_card_number() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info("payment", note="card on file: 4242 4242 4242 4242") + + assert record is not None + assert "4242 4242 4242 4242" not in record["meta"]["note"] + + +def test_redacts_recursively_through_tuples() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info("tuple meta", pair=("a@example.com", "safe")) + + assert record is not None + assert record["meta"]["pair"][0] == "***" + assert record["meta"]["pair"][1] == "safe" + + +def test_redacts_recursively_through_nested_dicts_and_lists() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + + record = logger.info( + "nested", + user={"email": "a@example.com", "tags": ["contact: b@example.com"]}, + ) + + assert record is not None + assert "a@example.com" not in record["meta"]["user"]["email"] + assert "b@example.com" not in record["meta"]["user"]["tags"][0] + + +def test_circular_reference_does_not_crash() -> None: + logger = Logger("app.test", plugins=[PIIRedactPlugin()]) + cyclic: dict[str, object] = {} + cyclic["self"] = cyclic + + record = logger.info("cyclic", data=cyclic) + + assert record is not None # must not raise / hang + + +def test_custom_patterns_override_defaults() -> None: + custom = {"employee_id": re.compile(r"\bEMP-\d{4}\b")} + logger = Logger("app.test", plugins=[PIIRedactPlugin(patterns=custom, replacement="[X]")]) + + record = logger.info("badge scan", note="badge EMP-1234 scanned, ssn 123-45-6789 ignored") + + assert record is not None + assert "EMP-1234" not in record["meta"]["note"] + assert "123-45-6789" in record["meta"]["note"] # default ssn pattern not active + + +def test_use_presidio_without_dependency_installed_raises_actionable_error() -> None: + with pytest.raises(ImportError, match="logquill\\[presidio\\]"): + PIIRedactPlugin(use_presidio=True) + + +def test_use_presidio_routes_text_through_the_analyzer_and_anonymizer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Fakes injected via sys.modules — the same pattern this repo already + # uses for other optional-dependency drivers (e.g. the cloud/NoSQL + # transports) — so this exercises the real Presidio code path without + # requiring the actual (heavy) dependency to be installed. + fake_analyzer_engine = MagicMock() + fake_analyzer_engine.return_value.analyze.return_value = "fake-analysis" + + fake_anonymizer_engine = MagicMock() + fake_anonymize_result = MagicMock() + fake_anonymize_result.text = "REDACTED" + fake_anonymizer_engine.return_value.anonymize.return_value = fake_anonymize_result + + analyzer_module = types.ModuleType("presidio_analyzer") + analyzer_module.AnalyzerEngine = fake_analyzer_engine # type: ignore[attr-defined] + anonymizer_module = types.ModuleType("presidio_anonymizer") + anonymizer_module.AnonymizerEngine = fake_anonymizer_engine # type: ignore[attr-defined] + + monkeypatch.setitem(sys.modules, "presidio_analyzer", analyzer_module) + monkeypatch.setitem(sys.modules, "presidio_anonymizer", anonymizer_module) + + plugin = PIIRedactPlugin(use_presidio=True, presidio_entities=["EMAIL_ADDRESS"]) + logger = Logger("app.test", plugins=[plugin]) + + record = logger.info("free text", note="jane@example.com") + + assert record is not None + assert record["meta"]["note"] == "REDACTED" + fake_analyzer_engine.return_value.analyze.assert_called_once_with( + text="jane@example.com", language="en", entities=["EMAIL_ADDRESS"] + ) + fake_anonymizer_engine.return_value.anonymize.assert_called_once_with( + text="jane@example.com", analyzer_results="fake-analysis" + ) diff --git a/tests/test_plugins/test_plugin.py b/tests/test_plugins/test_plugin.py index f7da2a8..5359620 100644 --- a/tests/test_plugins/test_plugin.py +++ b/tests/test_plugins/test_plugin.py @@ -105,3 +105,60 @@ def test_use_registers_a_plugin_and_returns_self_for_chaining() -> None: assert result is logger logger.info("hi") assert sink.records[0]["message"] == "HI" + + +def test_use_accepts_a_plain_function_as_middleware() -> None: + def strip_ssn(record: LogRecord) -> LogRecord | None: + record["meta"].pop("ssn", None) + return record + + sink = CollectingTransport() + logger = Logger("app.test", transports=[sink]) + + logger.use(strip_ssn) + record = logger.info("submit", ssn="123-45-6789", user_id=42) + + assert record is not None + assert "ssn" not in record["meta"] + assert record["meta"]["user_id"] == 42 + + +def test_function_middleware_behaves_identically_to_an_equivalent_plugin() -> None: + def uppercase(record: LogRecord) -> LogRecord | None: + record["message"] = record["message"].upper() + return record + + function_sink = CollectingTransport() + function_logger = Logger("app.test", transports=[function_sink]) + function_logger.use(uppercase) + + plugin_sink = CollectingTransport() + plugin_logger = Logger("app.test", transports=[plugin_sink], plugins=[UppercasePlugin()]) + + function_logger.info("hello") + plugin_logger.info("hello") + + assert function_sink.records[0]["message"] == plugin_sink.records[0]["message"] == "HELLO" + + +def test_use_accepting_a_function_returning_none_drops_the_record() -> None: + def drop_everything(record: LogRecord) -> LogRecord | None: + return None + + logger = Logger("app.test") + logger.use(drop_everything) + + assert logger.info("hello") is None + + +def test_constructor_plugins_list_accepts_functions_too() -> None: + def strip_ssn(record: LogRecord) -> LogRecord | None: + record["meta"].pop("ssn", None) + return record + + logger = Logger("app.test", plugins=[strip_ssn]) + + record = logger.info("submit", ssn="123-45-6789") + + assert record is not None + assert "ssn" not in record["meta"] diff --git a/tests/test_plugins/test_sampling_plugin.py b/tests/test_plugins/test_sampling_plugin.py index 4df7456..0b4d526 100644 --- a/tests/test_plugins/test_sampling_plugin.py +++ b/tests/test_plugins/test_sampling_plugin.py @@ -2,6 +2,7 @@ from logquill.logger import Logger from logquill.plugins.sampling_plugin import SamplingPlugin +from logquill.transports.transport import CollectingTransport def test_rate_zero_drops_everything() -> None: @@ -30,3 +31,89 @@ def test_custom_rng_controls_keep_or_drop() -> None: assert logger_keep.info("hello") is not None assert logger_drop.info("hello") is None + + +def test_without_transports_a_trace_id_does_not_enable_tail_elevation() -> None: + # Backward-compatible: no `transports` given means plain rate sampling — + # level is irrelevant, even for a record that carries a trace id and + # would otherwise trigger elevation. + sink = CollectingTransport() + sampling = SamplingPlugin(0.0, rng=lambda: 0.9) + logger = Logger("app.test", transports=[sink], plugins=[sampling]) + + assert logger.info("step 1", trace_id="t1") is None + assert logger.error("step 2", trace_id="t1") is None + assert sink.records == [] + + +def test_tail_based_elevation_flushes_buffered_records_from_the_same_trace() -> None: + sink = CollectingTransport() + sampling = SamplingPlugin(0.0, rng=lambda: 0.9, transports=[sink]) + logger = Logger("app.test", transports=[sink], plugins=[sampling]) + + assert logger.info("step 1", trace_id="t1") is None + assert logger.info("step 2", trace_id="t1") is None + assert sink.records == [] + + record = logger.error("step 3", trace_id="t1") + + assert record is not None + messages = [r["message"] for r in sink.records] + assert messages == ["step 1", "step 2", "step 3"] + + +def test_elevation_only_affects_the_matching_trace() -> None: + sink = CollectingTransport() + sampling = SamplingPlugin(0.0, rng=lambda: 0.9, transports=[sink]) + logger = Logger("app.test", transports=[sink], plugins=[sampling]) + + assert logger.info("other trace", trace_id="t2") is None + assert logger.info("step 1", trace_id="t1") is None + logger.error("step 2", trace_id="t1") + + messages = [r["message"] for r in sink.records] + assert "other trace" not in messages + assert messages == ["step 1", "step 2"] + + +def test_records_after_elevation_ship_unconditionally() -> None: + sink = CollectingTransport() + sampling = SamplingPlugin(0.0, rng=lambda: 0.9, transports=[sink]) + logger = Logger("app.test", transports=[sink], plugins=[sampling]) + + logger.error("triggers elevation", trace_id="t1") + record = logger.info("after elevation", trace_id="t1") + + assert record is not None + assert sink.records[-1]["message"] == "after elevation" + + +def test_buffer_is_bounded_by_max_traces() -> None: + sink = CollectingTransport() + sampling = SamplingPlugin(0.0, rng=lambda: 0.9, transports=[sink], max_traces=1) + logger = Logger("app.test", transports=[sink], plugins=[sampling]) + + logger.info("trace one", trace_id="t1") + logger.info("trace two", trace_id="t2") # evicts t1's buffer (max_traces=1) + logger.error("elevates t1", trace_id="t1") + + # t1's earlier buffered record was evicted, so only the elevating record ships + messages = [r["message"] for r in sink.records] + assert "trace one" not in messages + assert "elevates t1" in messages + + +def test_buffer_is_bounded_by_max_buffered_records() -> None: + sink = CollectingTransport() + sampling = SamplingPlugin(0.0, rng=lambda: 0.9, transports=[sink], max_buffered_records=1) + logger = Logger("app.test", transports=[sink], plugins=[sampling]) + + logger.info("trace one, record one", trace_id="t1") + # second buffered record (still t1) exceeds max_buffered_records=1, + # evicting the whole oldest trace's buffer (t1's first record) + logger.info("trace one, record two", trace_id="t1") + logger.error("elevates t1", trace_id="t1") + + messages = [r["message"] for r in sink.records] + assert "trace one, record one" not in messages + assert "elevates t1" in messages diff --git a/tests/test_plugins/test_slack_alert_plugin.py b/tests/test_plugins/test_slack_alert_plugin.py new file mode 100644 index 0000000..aaa1ee7 --- /dev/null +++ b/tests/test_plugins/test_slack_alert_plugin.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +from logquill.levels import Level +from logquill.plugins.slack_alert_plugin import SlackAlertPlugin +from logquill.records import create_record + + +def _record(message: str = "boom") -> object: + return create_record(level=Level.ERROR, logger="app.test", message=message, meta={}) + + +def test_send_alert_posts_json_body_with_text() -> None: + plugin = SlackAlertPlugin("https://hooks.slack.example/T000/B000/xxx") + fake_response = MagicMock() + fake_response.status = 200 + fake_response.__enter__.return_value = fake_response + + with patch("urllib.request.urlopen", return_value=fake_response) as mock_urlopen: + plugin.send_alert(_record(), 1) # type: ignore[arg-type] + + request = mock_urlopen.call_args[0][0] + assert request.full_url == "https://hooks.slack.example/T000/B000/xxx" + body = json.loads(request.data) + assert "boom" in body["text"] + + +def test_send_alert_includes_occurrence_count_when_greater_than_one() -> None: + plugin = SlackAlertPlugin("https://hooks.slack.example/T000/B000/xxx") + fake_response = MagicMock() + fake_response.status = 200 + fake_response.__enter__.return_value = fake_response + + with patch("urllib.request.urlopen", return_value=fake_response) as mock_urlopen: + plugin.send_alert(_record(), 7) # type: ignore[arg-type] + + body = json.loads(mock_urlopen.call_args[0][0].data) + assert "x7" in body["text"] + + +def test_send_alert_raises_on_http_error_status() -> None: + plugin = SlackAlertPlugin("https://hooks.slack.example/T000/B000/xxx") + fake_response = MagicMock() + fake_response.status = 500 + fake_response.__enter__.return_value = fake_response + + with patch("urllib.request.urlopen", return_value=fake_response): + try: + plugin.send_alert(_record(), 1) # type: ignore[arg-type] + raised = False + except RuntimeError: + raised = True + + assert raised diff --git a/tests/test_plugins/test_tamper_evident_plugin.py b/tests/test_plugins/test_tamper_evident_plugin.py new file mode 100644 index 0000000..18ebfdd --- /dev/null +++ b/tests/test_plugins/test_tamper_evident_plugin.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import copy + +from logquill.logger import Logger +from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin + + +def test_each_record_gets_a_hash_and_prev_hash() -> None: + logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + + record = logger.info("hello") + + assert record is not None + assert isinstance(record["meta"]["hash"], str) + assert record["meta"]["prev_hash"] == "0" * 64 + + +def test_chain_links_consecutive_records() -> None: + logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + + first = logger.info("one") + second = logger.info("two") + + assert first is not None and second is not None + assert second["meta"]["prev_hash"] == first["meta"]["hash"] + + +def test_verify_chain_passes_on_an_untampered_log() -> None: + logger = Logger("app.test", 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()]) + 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] + + assert TamperEvidentPlugin.verify_chain(tampered) is False # type: ignore[arg-type] + + +def test_verify_chain_detects_a_removed_record() -> None: + logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + records = [logger.info(f"event {i}", n=i) for i in range(5)] + tampered = records[:2] + records[3:] # remove index 2 + + assert TamperEvidentPlugin.verify_chain(tampered) is False # type: ignore[arg-type] + + +def test_verify_chain_detects_reordered_records() -> None: + logger = Logger("app.test", plugins=[TamperEvidentPlugin()]) + records = [logger.info(f"event {i}", n=i) for i in range(3)] + reordered = [records[1], records[0], records[2]] + + assert TamperEvidentPlugin.verify_chain(reordered) is False # type: ignore[arg-type] + + +def test_verify_chain_on_empty_input_is_true() -> None: + assert TamperEvidentPlugin.verify_chain([]) is True