From 56a5f5d671bd7ca8a34537c185673e9f19063796 Mon Sep 17 00:00:00 2001 From: jagmarques <32335502+jagmarques@users.noreply.github.com> Date: Sat, 24 Oct 2026 15:35:31 +0200 Subject: [PATCH 1/2] feat(integrations): add asqav local tamper-evident audit-log callback Re-lands #30238 on main. callbacks=["asqav"] appends one hash-chained record per LLM call to a local JSONL file and can verify the chain offline. Opt-in, no network calls, content stored as SHA-256 digests by default. --- litellm/__init__.py | 1 + litellm/integrations/asqav/__init__.py | 5 + litellm/integrations/asqav/asqav.py | 376 +++++++++++++++++ .../custom_logger_registry.py | 2 + litellm/litellm_core_utils/litellm_logging.py | 12 + tests/unit/integrations/asqav/__init__.py | 0 tests/unit/integrations/asqav/test_asqav.py | 384 ++++++++++++++++++ 7 files changed, 780 insertions(+) create mode 100644 litellm/integrations/asqav/__init__.py create mode 100644 litellm/integrations/asqav/asqav.py create mode 100644 tests/unit/integrations/asqav/__init__.py create mode 100644 tests/unit/integrations/asqav/test_asqav.py diff --git a/litellm/__init__.py b/litellm/__init__.py index 1784deeb797..a2dafe66d24 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -156,6 +156,7 @@ _custom_logger_compatible_callbacks_literal = Literal[ "compression_interception", "newrelic", "signoz", + "asqav", ] cold_storage_custom_logger: Optional[_custom_logger_compatible_callbacks_literal] = None logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None diff --git a/litellm/integrations/asqav/__init__.py b/litellm/integrations/asqav/__init__.py new file mode 100644 index 00000000000..1d13ceca16a --- /dev/null +++ b/litellm/integrations/asqav/__init__.py @@ -0,0 +1,5 @@ +"""Asqav local-first, tamper-evident audit-log callback for LiteLLM.""" + +from litellm.integrations.asqav.asqav import AsqavLogger + +__all__ = ("AsqavLogger",) diff --git a/litellm/integrations/asqav/asqav.py b/litellm/integrations/asqav/asqav.py new file mode 100644 index 00000000000..0e072929490 --- /dev/null +++ b/litellm/integrations/asqav/asqav.py @@ -0,0 +1,376 @@ +"""Asqav local-first audit-log callback for LiteLLM. + +Each LLM call appends one record to a local JSONL file. Every record carries a +SHA-256 hash over its own canonical fields plus the previous record's hash, so +the file forms a tamper-evident chain that can be checked offline with the +standard library alone. + +- No dependencies beyond litellm itself, and no network calls. +- Fail-soft: an error while building or writing a record is logged and never + reaches the LLM call. +- Message and response content are stored as SHA-256 digests by default, so an + auditor can prove a payload existed without the log holding it. +""" + +import asyncio +import hashlib +import itertools +import json +import os +import threading +import time +from collections.abc import Iterable, Mapping, Sequence +from datetime import datetime, timezone +from typing import BinaryIO, Final + +from pydantic import ConfigDict, TypeAdapter, ValidationError + +from litellm._logging import verbose_logger +from litellm.integrations.custom_logger import CustomLogger + +__all__ = ("AsqavLogger",) + +# prev_hash of the first record in a chain (it has no predecessor). +GENESIS_HASH: Final = "0" * 64 + +# Overridable with ASQAV_LOG_PATH or the log_path argument. +DEFAULT_LOG_PATH: Final = os.path.join(os.path.expanduser("~"), ".litellm_asqav_audit.jsonl") + +_TAIL_CHUNK_BYTES: Final = 4096 + +_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True)) +_OBJECT_MAPPING: Final = TypeAdapter(Mapping[object, object], config=ConfigDict(hide_input_in_errors=True)) +_OBJECT_SEQUENCE: Final = TypeAdapter(Sequence[object], config=ConfigDict(hide_input_in_errors=True)) + +# Header and key values that must never reach the log. +_SENSITIVE_KEYS: Final = frozenset({"user_api_key", "Authorization", "authorization", "token", "api_key"}) + +# Proxy identity fields copied from litellm_params.metadata so each record says who made the call. +_PROXY_IDENTITY_KEYS: Final = frozenset( + { + "user_api_key_user_id", + "user_api_key_team_id", + "user_api_key_org_id", + "user_api_key_alias", + "user_id", + "team_id", + "org_id", + } +) + + +def _sha256_hex(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _canonical_bytes(record: Mapping[str, object]) -> bytes: + """Deterministic serialisation for hashing: sorted keys, no whitespace. + + Values JSON cannot encode are stored as their ``str()`` form, so the bytes + written to disk and the bytes hashed are the same after a JSON round trip. + """ + return json.dumps(record, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + + +def _content_digest(value: object) -> str | None: + """SHA-256 hex digest of a content value, or None when there is no content.""" + if value is None: + return None + return _sha256_hex(json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")) + + +def _read_tail(fh: BinaryIO, size: int) -> bytes: + """Read back from the end of fh, doubling the window until it holds the whole last line.""" + for shift in itertools.count(): + read_size = min(_TAIL_CHUNK_BYTES << shift, size) + fh.seek(size - read_size) + tail = fh.read(read_size) + if read_size == size or b"\n" in tail.rstrip(b"\n"): + return tail + return b"" + + +def _str_keyed(value: object) -> Mapping[str, object]: + """The string-keyed entries of value when it is a mapping, else an empty mapping.""" + if not isinstance(value, Mapping): + return {} + entries: Final = _OBJECT_MAPPING.validate_python(value) + return {key: item for key, item in entries.items() if isinstance(key, str)} + + +def _int_or_none(value: object) -> int | None: + return value if isinstance(value, int) and not isinstance(value, bool) else None + + +def _str_or_none(value: object) -> str | None: + return value if isinstance(value, str) else None + + +def _first_choice(response_obj: object) -> object: + choices: Final[object] = getattr(response_obj, "choices", None) + if isinstance(choices, list | tuple) and choices: + return _OBJECT_SEQUENCE.validate_python(choices)[0] + return None + + +def _response_content(response_obj: object) -> object: + message: Final[object] = getattr(_first_choice(response_obj), "message", None) + content: Final[object] = getattr(message, "content", None) + return content + + +def _latency_ms(start_time: object, end_time: object) -> int | None: + if isinstance(start_time, datetime) and isinstance(end_time, datetime): + return int((end_time - start_time).total_seconds() * 1000) + return None + + +def _proxy_identity(kwargs: Mapping[str, object]) -> Mapping[str, object]: + """Identity fields from litellm_params.metadata; raw keys and auth headers are never copied.""" + litellm_params: Final = _str_keyed(kwargs.get("litellm_params")) + proxy_metadata: Final = _str_keyed(litellm_params.get("metadata")) + return { + key: value + for key, value in proxy_metadata.items() + if key in _PROXY_IDENTITY_KEYS and key not in _SENSITIVE_KEYS + } + + +def _call_id(kwargs: Mapping[str, object]) -> str: + standard_logging_object: Final = _str_keyed(kwargs.get("standard_logging_object")) + candidates: Final = ( + standard_logging_object.get("id"), + standard_logging_object.get("litellm_call_id"), + kwargs.get("litellm_call_id"), + kwargs.get("id"), + ) + found: Final = next((str(candidate) for candidate in candidates if candidate), None) + return found if found is not None else str(int(time.time() * 1e6)) + + +def _extract_loggable( + kwargs: Mapping[str, object], + response_obj: object, + start_time: object, + end_time: object, + status: str, +) -> Mapping[str, object]: + """Metadata and content digests for one callback invocation. + + Message content and response text are never stored in the clear here; only + their SHA-256 digests appear, so a caller can prove a payload existed + without the log reconstructing it. + """ + root_metadata: Final = _str_keyed(kwargs.get("metadata")) or _str_keyed(kwargs.get("litellm_metadata")) + usage: Final[object] = getattr(response_obj, "usage", None) + hidden_params: Final = _str_keyed(getattr(response_obj, "_hidden_params", None)) + first_choice: Final = _first_choice(response_obj) + model: Final = kwargs.get("model") + + return { + "call_id": _call_id(kwargs), + "model": model if isinstance(model, str) else "", + "status": status, + "latency_ms": _latency_ms(start_time, end_time), + "prompt_tokens": _int_or_none(getattr(usage, "prompt_tokens", None)), + "completion_tokens": _int_or_none(getattr(usage, "completion_tokens", None)), + "total_tokens": _int_or_none(getattr(usage, "total_tokens", None)), + "finish_reason": _str_or_none(getattr(first_choice, "finish_reason", None)), + "provider_request_id": _str_or_none(hidden_params.get("x-request-id")) + or _str_or_none(hidden_params.get("cf-ray")), + "messages_digest": _content_digest(kwargs.get("messages")), + "response_content_digest": _content_digest(_response_content(response_obj)), + # Root metadata wins over the proxy identity copy, matching the user's own tags. + "metadata": {**_proxy_identity(kwargs), **root_metadata}, + } + + +def _private_opener(path: str, flags: int) -> int: + """Create the log owner-only (0o600) whatever the process umask.""" + return os.open(path, flags, 0o600) + + +def _redact_from_env() -> bool: + return os.environ.get("ASQAV_REDACT_CONTENT", "true").lower() != "false" + + +class AsqavLogger(CustomLogger): + """Tamper-evident local-first audit-log callback for LiteLLM. + + Enable with ``litellm.callbacks = ["asqav"]`` or ``callbacks: ["asqav"]`` + under ``litellm_settings`` in the proxy config. + + Configuration (environment variables, overridden by constructor arguments): + + ASQAV_LOG_PATH + Path to the JSONL audit log. Defaults to ~/.litellm_asqav_audit.jsonl. + + ASQAV_REDACT_CONTENT + Set to "false" to store message and response content in the clear + instead of as SHA-256 digests. Defaults to "true" (digests only). + + Multi-worker limitation: this logger expects a single audit writer per log + file. Its lock serialises threads within one process, not separate OS + processes. In a multi-worker proxy, give each worker its own + ASQAV_LOG_PATH or run one dedicated writer; otherwise records from several + workers share seq and prev_hash values and verify_chain reports a break. + """ + + def __init__( + self, + log_path: str | None = None, + redact_content: bool | None = None, + ) -> None: + super().__init__() # pyright: ignore[reportUnknownMemberType] # CustomLogger.__init__ declares untyped **kwargs + self._log_path: Final[str] = log_path or os.environ.get("ASQAV_LOG_PATH") or DEFAULT_LOG_PATH + self._redact_content: Final[bool] = _redact_from_env() if redact_content is None else redact_content + self._lock: Final = threading.Lock() + self._call_count: int = 0 + self._prev_hash: str = GENESIS_HASH + # Resume from an existing log so the chain continues across restarts. + self._load_chain_tail() + + def __repr__(self) -> str: + return f"AsqavLogger(log_path={self._log_path!r}, redact_content={self._redact_content})" + + @property + def log_path(self) -> str: + return self._log_path + + def _load_chain_tail(self) -> None: + """Read the last record of an existing log to resume seq and prev_hash.""" + try: + if not os.path.exists(self._log_path): + return + with open(self._log_path, "rb") as fh: + size: Final = fh.seek(0, os.SEEK_END) + if size == 0: + return + tail: Final = _read_tail(fh, size) + lines: Final = [line for line in tail.split(b"\n") if line.strip()] + if not lines: + return + last_record: Final = _JSON_OBJECT.validate_python(json.loads(lines[-1].decode("utf-8"))) + record_hash: Final = last_record.get("record_hash") + seq: Final = last_record.get("seq") + self._prev_hash = record_hash if isinstance(record_hash, str) else GENESIS_HASH + self._call_count = seq + 1 if isinstance(seq, int) else 0 + except (OSError, ValueError, ValidationError): + verbose_logger.debug("[AsqavLogger] Could not load chain tail", exc_info=True) + + def _build_and_append( + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: object, + end_time: object, + status: str, + ) -> None: + """Build one audit record and append it to the JSONL log. + + seq and prev_hash are assigned and the file is written under one lock, + so the on-disk order always matches the chain order. Chain state only + advances after a successful write: a failed write drops the record and + the chain continues from the last record on disk. + """ + try: + loggable: Final = _extract_loggable(kwargs, response_obj, start_time, end_time, status) + clear_content: Final[Mapping[str, object]] = ( + {} + if self._redact_content + else {"messages": kwargs.get("messages"), "response_content": _response_content(response_obj)} + ) + with self._lock: + # Every field enters the hash, so an auditor can reproduce it from the log alone. + hashable: Final[Mapping[str, object]] = { + "seq": self._call_count, + "ts": datetime.now(tz=timezone.utc).isoformat(), + "prev_hash": self._prev_hash, + **loggable, + **clear_content, + } + record_hash: Final = _sha256_hex(_canonical_bytes(hashable)) + if not self._write_record({**hashable, "record_hash": record_hash}): + return + self._prev_hash = record_hash + self._call_count += 1 + except Exception: # noqa: BLE001 # an audit-log failure must never break the LLM call + verbose_logger.debug("[AsqavLogger] Unhandled error while building a record", exc_info=True) + + def _write_record(self, record: Mapping[str, object]) -> bool: + """Append one record to the log file. Returns False if the write failed.""" + try: + parent: Final = os.path.dirname(self._log_path) + if parent: + os.makedirs(parent, exist_ok=True) + # Lock down a file an earlier run created under a permissive umask; new + # files are created 0o600 by _private_opener. + if os.path.exists(self._log_path): + os.chmod(self._log_path, 0o600) + with open(self._log_path, "a", encoding="utf-8", opener=_private_opener) as fh: + fh.write(json.dumps(record, separators=(",", ":"), default=str) + "\n") + except (OSError, TypeError, ValueError): + verbose_logger.warning("[AsqavLogger] Failed to write audit record", exc_info=True) + return False + else: + return True + + def log_success_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + self._build_and_append(kwargs, response_obj, start_time, end_time, "success") + + def log_failure_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + self._build_and_append(kwargs, response_obj, start_time, end_time, "failure") + + async def async_log_success_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + await asyncio.to_thread(self._build_and_append, kwargs, response_obj, start_time, end_time, "success") + + async def async_log_failure_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + await asyncio.to_thread(self._build_and_append, kwargs, response_obj, start_time, end_time, "failure") + + def verify_chain(self, log_path: str | None = None) -> tuple[bool, str]: + """Check the audit log at log_path (default: this logger's log). + + Returns (True, "ok") when every record's hash matches its content and its + prev_hash matches the previous record's hash, else (False, reason) for + the first violation found. Not called on the request path. + """ + path: Final = log_path or self._log_path + try: + with open(path, encoding="utf-8") as fh: + return _verify_lines(fh) + except FileNotFoundError: + return False, f"log file not found: {path}" + except (OSError, ValueError, ValidationError) as exc: + return False, f"verification error: {exc}" + + +def _verify_lines(lines: Iterable[str]) -> tuple[bool, str]: + prev_hash = GENESIS_HASH # rebind-ok: walks the chain one record at a time + for lineno, raw_line in enumerate(lines, start=1): + line = raw_line.strip() + if not line: + continue + record = _JSON_OBJECT.validate_python(json.loads(line)) + stored_hash = record.get("record_hash") + computed_hash = _sha256_hex(_canonical_bytes({k: v for k, v in record.items() if k != "record_hash"})) + if computed_hash != stored_hash: + return ( + False, + f"line {lineno}: hash mismatch (stored={str(stored_hash)[:12]}, computed={computed_hash[:12]})", + ) + record_prev = record.get("prev_hash", GENESIS_HASH) + if record_prev != prev_hash: + return ( + False, + f"line {lineno}: prev_hash chain break (expected={prev_hash[:12]}, got={str(record_prev)[:12]})", + ) + prev_hash = computed_hash + return True, "ok" diff --git a/litellm/litellm_core_utils/custom_logger_registry.py b/litellm/litellm_core_utils/custom_logger_registry.py index 74badc2c8c1..fd8130a1b1a 100644 --- a/litellm/litellm_core_utils/custom_logger_registry.py +++ b/litellm/litellm_core_utils/custom_logger_registry.py @@ -14,6 +14,7 @@ from litellm import _custom_logger_compatible_callbacks_literal from litellm.integrations.agentops import AgentOps from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook from litellm.integrations.argilla import ArgillaLogger +from litellm.integrations.asqav import AsqavLogger from litellm.integrations.azure_sentinel.azure_sentinel import AzureSentinelLogger from litellm.integrations.azure_storage.azure_storage import AzureBlobStorageLogger from litellm.integrations.bitbucket import BitBucketPromptManager @@ -121,6 +122,7 @@ class CustomLoggerRegistry: "ternary": TernaryLogger, "posthog": PostHogLogger, "newrelic": NewRelicLogger, + "asqav": AsqavLogger, } try: diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 41f13817d75..4ecace649bd 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -198,6 +198,7 @@ from litellm.utils import get_base_model_from_metadata, print_verbose from ..integrations.argilla import ArgillaLogger from ..integrations.arize.arize_phoenix import ArizePhoenixLogger +from ..integrations.asqav import AsqavLogger from ..integrations.athina import AthinaLogger from ..integrations.azure_sentinel.azure_sentinel import AzureSentinelLogger from ..integrations.azure_storage.azure_storage import AzureBlobStorageLogger @@ -5864,6 +5865,13 @@ def build_custom_logger_compatible_class( newrelic_logger: Final = NewRelicLogger() _in_memory_loggers.append(newrelic_logger) return newrelic_logger + elif logging_integration == "asqav": + for callback in _in_memory_loggers: + if isinstance(callback, AsqavLogger): + return callback + asqav_logger: Final = AsqavLogger() + _in_memory_loggers.append(asqav_logger) + return asqav_logger return None @@ -6276,6 +6284,10 @@ def get_custom_logger_compatible_class( return callback if isinstance(callback, NewRelicLogger): return callback + elif logging_integration == "asqav": + for callback in _in_memory_loggers: + if isinstance(callback, AsqavLogger): + return callback return None except Exception as e: diff --git a/tests/unit/integrations/asqav/__init__.py b/tests/unit/integrations/asqav/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/integrations/asqav/test_asqav.py b/tests/unit/integrations/asqav/test_asqav.py new file mode 100644 index 00000000000..1bd7e0c34f8 --- /dev/null +++ b/tests/unit/integrations/asqav/test_asqav.py @@ -0,0 +1,384 @@ +"""Tests for the Asqav local-first audit-log callback. + +No network and no provider calls: every test writes to a tmp_path log. + +The chain checks re-derive record hashes with hashlib and json directly, so they +also prove an auditor can verify the log without litellm installed. +""" + +import asyncio +import hashlib +import json +import os +import stat +import threading +from collections.abc import Mapping, Sequence +from datetime import datetime, timezone +from pathlib import Path +from typing import Final + +import pytest + +import litellm +from litellm.integrations.asqav import AsqavLogger +from litellm.integrations.asqav.asqav import GENESIS_HASH +from litellm.litellm_core_utils import litellm_logging as logging_module +from litellm.litellm_core_utils.custom_logger_registry import CustomLoggerRegistry +from litellm.types.utils import Choices, Message, ModelResponse, Usage +from tests._wait_helpers import wait_until + +START: Final = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) +END: Final = datetime(2026, 1, 1, 0, 0, 1, tzinfo=timezone.utc) + + +def _kwargs(content: str = "hello", call_id: str = "call-1") -> dict[str, object]: + return { + "model": "gpt-4o", + "messages": [{"role": "user", "content": content}], + "litellm_call_id": call_id, + } + + +def _response(content: str = "world") -> ModelResponse: + return ModelResponse( + id="chatcmpl-test", + model="gpt-4o", + choices=[Choices(finish_reason="stop", index=0, message=Message(content=content, role="assistant"))], + usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15), + ) + + +def _logger(path: Path, redact_content: bool = True) -> AsqavLogger: + return AsqavLogger(log_path=str(path), redact_content=redact_content) + + +def _append(logger: AsqavLogger, count: int, prefix: str = "msg") -> None: + for i in range(count): + logger.log_success_event(_kwargs(content=f"{prefix}-{i}"), _response(content=f"resp-{i}"), START, END) + + +def _records(path: Path) -> list[dict[str, object]]: + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] + + +def _rewrite(path: Path, records: Sequence[Mapping[str, object]]) -> None: + path.write_text("".join(json.dumps(r, separators=(",", ":")) + "\n" for r in records), encoding="utf-8") + + +def _stdlib_hash(record: Mapping[str, object]) -> str: + """The record hash, recomputed the way an auditor would with only the standard library.""" + body = {k: v for k, v in record.items() if k != "record_hash"} + return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() + + +def test_first_record_starts_the_chain(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + _logger(path).log_success_event(_kwargs(), _response(), START, END) + + [record] = _records(path) + assert record["seq"] == 0 + assert record["prev_hash"] == GENESIS_HASH + assert record["status"] == "success" + assert record["record_hash"] == _stdlib_hash(record) + + +def test_records_link_and_hash_reproduces_with_stdlib(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + _append(logger, 10) + + records = _records(path) + assert [r["seq"] for r in records] == list(range(10)) + assert records[0]["prev_hash"] == GENESIS_HASH + for previous, current in zip(records, records[1:]): + assert current["prev_hash"] == previous["record_hash"] + assert all(r["record_hash"] == _stdlib_hash(r) for r in records) + assert logger.verify_chain() == (True, "ok") + + +def test_record_carries_usage_latency_and_call_metadata(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + _logger(path).log_success_event(_kwargs(call_id="call-abc"), _response(), START, END) + + [record] = _records(path) + assert record["call_id"] == "call-abc" + assert record["model"] == "gpt-4o" + assert record["latency_ms"] == 1000 + assert (record["prompt_tokens"], record["completion_tokens"], record["total_tokens"]) == (10, 5, 15) + assert record["finish_reason"] == "stop" + + +def test_verify_detects_an_edited_field(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + _append(logger, 5) + records = _records(path) + _rewrite(path, [*records[:2], {**records[2], "model": "tampered-model"}, *records[3:]]) + + ok, reason = logger.verify_chain() + + assert ok is False + assert reason.startswith("line 3: hash mismatch") + + +def test_verify_detects_a_rehashed_prev_hash_edit(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + _append(logger, 5) + records = _records(path) + forged = {**records[3], "prev_hash": "a" * 64} + _rewrite(path, [*records[:3], {**forged, "record_hash": _stdlib_hash(forged)}, *records[4:]]) + + ok, reason = logger.verify_chain() + + assert ok is False + assert reason.startswith("line 4: prev_hash chain break") + + +def test_verify_detects_a_deleted_record(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + _append(logger, 5) + records = _records(path) + _rewrite(path, [*records[:2], *records[3:]]) + + ok, reason = logger.verify_chain() + + assert ok is False + assert reason.startswith("line 3: prev_hash chain break") + + +def test_verify_reports_a_missing_file(tmp_path: Path) -> None: + missing = tmp_path / "missing.jsonl" + + assert _logger(tmp_path / "audit.jsonl").verify_chain(str(missing)) == (False, f"log file not found: {missing}") + + +def test_chain_and_seq_resume_after_restart(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + _append(_logger(path), 5, prefix="first") + last_before_restart = _records(path)[-1] + + restarted = _logger(path) + _append(restarted, 1, prefix="second") + + records = _records(path) + assert len(records) == 6 + assert records[5]["seq"] == 5 + assert records[5]["prev_hash"] == last_before_restart["record_hash"] + assert restarted.verify_chain() == (True, "ok") + + +def test_restart_resumes_when_last_record_exceeds_tail_window(tmp_path: Path) -> None: + """A last record bigger than the 4 KB first read window still resumes the chain.""" + path = tmp_path / "audit.jsonl" + big_kwargs = {**_kwargs(content="big"), "metadata": {"blob": "x" * 10_000}} + _logger(path).log_success_event(big_kwargs, _response(), START, END) + assert os.path.getsize(path) > 4096 + + restarted = _logger(path) + _append(restarted, 2) + + records = _records(path) + assert [r["seq"] for r in records] == [0, 1, 2] + assert restarted.verify_chain() == (True, "ok") + + +def test_failure_event_is_logged(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + _logger(path).log_failure_event(_kwargs(), None, START, END) + + [record] = _records(path) + assert record["status"] == "failure" + assert record["prompt_tokens"] is None + + +def test_malformed_response_still_writes_a_record(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + _logger(path).log_success_event(_kwargs(), object(), START, END) + + [record] = _records(path) + assert record["status"] == "success" + assert record["response_content_digest"] is None + + +def test_failed_write_does_not_raise_or_advance_the_chain(tmp_path: Path) -> None: + blocker = tmp_path / "logs" + blocker.write_text("", encoding="utf-8") + path = blocker / "audit.jsonl" + logger = _logger(path) + + logger.log_success_event(_kwargs(call_id="call-dropped"), _response(), START, END) + blocker.unlink() + logger.log_success_event(_kwargs(call_id="call-kept"), _response(), START, END) + + [record] = _records(path) + assert record["call_id"] == "call-kept" + assert record["seq"] == 0 + assert record["prev_hash"] == GENESIS_HASH + + +def test_content_is_digested_by_default(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + _logger(path).log_success_event(_kwargs(content="my private prompt"), _response("my private answer"), START, END) + + raw = path.read_text(encoding="utf-8") + [record] = _records(path) + assert "my private prompt" not in raw + assert "my private answer" not in raw + expected_digest = hashlib.sha256( + json.dumps([{"role": "user", "content": "my private prompt"}], sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + assert record["messages_digest"] == expected_digest + assert isinstance(record["response_content_digest"], str) + + +def test_redact_content_false_stores_plaintext(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path, redact_content=False) + logger.log_success_event(_kwargs(content="visible prompt"), _response("visible answer"), START, END) + + [record] = _records(path) + assert record["messages"] == [{"role": "user", "content": "visible prompt"}] + assert record["response_content"] == "visible answer" + assert logger.verify_chain() == (True, "ok") + + +def test_env_vars_configure_path_and_redaction(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + path = tmp_path / "from-env.jsonl" + monkeypatch.setenv("ASQAV_LOG_PATH", str(path)) + monkeypatch.setenv("ASQAV_REDACT_CONTENT", "false") + + logger = AsqavLogger() + logger.log_success_event(_kwargs(content="env prompt"), _response(), START, END) + + assert logger.log_path == str(path) + assert _records(path)[0]["messages"] == [{"role": "user", "content": "env prompt"}] + + +def test_proxy_identity_is_kept_and_secrets_are_dropped(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + kwargs = { + **_kwargs(), + "litellm_params": { + "metadata": { + "user_api_key_user_id": "user-abc", + "user_api_key_team_id": "team-xyz", + "user_api_key_org_id": "org-123", + "user_api_key_alias": "my-key", + "user_api_key": "hashed-key-value", + "Authorization": "placeholder-auth-header", + "requester_ip_address": "10.0.0.1", + } + }, + } + _logger(path).log_success_event(kwargs, _response(), START, END) + + metadata = _records(path)[0]["metadata"] + assert metadata == { + "user_api_key_user_id": "user-abc", + "user_api_key_team_id": "team-xyz", + "user_api_key_org_id": "org-123", + "user_api_key_alias": "my-key", + } + + +def test_non_json_metadata_is_stored_as_text_and_verifies(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + kwargs = {**_kwargs(), "metadata": {"requested_at": START}} + logger.log_success_event(kwargs, _response(), START, END) + + assert _records(path)[0]["metadata"] == {"requested_at": str(START)} + assert logger.verify_chain() == (True, "ok") + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX file modes") +def test_log_file_is_private(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + path.write_text("", encoding="utf-8") + path.chmod(0o644) + + _logger(path).log_success_event(_kwargs(), _response(), START, END) + + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_concurrent_callbacks_keep_disk_order_equal_to_chain_order(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + threads_count, per_thread = 8, 25 + barrier = threading.Barrier(threads_count) + + def worker(thread_id: int) -> None: + barrier.wait() + _append(logger, per_thread, prefix=f"t{thread_id}") + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(threads_count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert [r["seq"] for r in _records(path)] == list(range(threads_count * per_thread)) + assert logger.verify_chain() == (True, "ok") + + +def test_async_hooks_write_records(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + + async def run() -> None: + await logger.async_log_success_event(_kwargs(content="async-ok"), _response(), START, END) + await logger.async_log_failure_event(_kwargs(content="async-fail"), None, START, END) + + asyncio.run(run()) + + assert [r["status"] for r in _records(path)] == ["success", "failure"] + assert logger.verify_chain() == (True, "ok") + + +def test_asqav_is_a_registered_callback_name() -> None: + assert "asqav" in litellm._known_custom_logger_compatible_callbacks + assert CustomLoggerRegistry.CALLBACK_CLASS_STR_TO_CLASS_TYPE["asqav"] is AsqavLogger + + +def test_callbacks_asqav_builds_one_logger_and_reuses_it(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """`callbacks=["asqav"]` goes through litellm_logging, which must hand back one instance.""" + monkeypatch.setenv("ASQAV_LOG_PATH", str(tmp_path / "audit.jsonl")) + monkeypatch.setattr(logging_module, "_in_memory_loggers", []) + + assert logging_module.get_custom_logger_compatible_class("asqav") is None + + first = logging_module._init_custom_logger_compatible_class( + logging_integration="asqav", internal_usage_cache=None, llm_router=None, custom_logger_init_args={} + ) + second = logging_module._init_custom_logger_compatible_class( + logging_integration="asqav", internal_usage_cache=None, llm_router=None, custom_logger_init_args={} + ) + + assert isinstance(first, AsqavLogger) + assert second is first + assert first.log_path == str(tmp_path / "audit.jsonl") + assert logging_module.get_custom_logger_compatible_class("asqav") is first + + +def test_completion_with_callbacks_asqav_appends_a_verifiable_record( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A real litellm.completion call (mock_response, no network) lands one chained record.""" + path = tmp_path / "audit.jsonl" + monkeypatch.setenv("ASQAV_LOG_PATH", str(path)) + monkeypatch.setattr(logging_module, "_in_memory_loggers", []) + monkeypatch.setattr(litellm, "callbacks", ["asqav"]) + + litellm.completion( + model="gpt-4o", messages=[{"role": "user", "content": "hi"}], mock_response="hello from the mock" + ) + wait_until(lambda: path.exists() and bool(_records(path)), message="asqav record written") + + [record] = _records(path) + assert record["status"] == "success" + assert record["model"] == "gpt-4o" + assert record["record_hash"] == _stdlib_hash(record) + assert "hello from the mock" not in path.read_text(encoding="utf-8") From 8fae48e4ba654f12a30ba4dab4a5950dadb8b06a Mon Sep 17 00:00:00 2001 From: jagmarques <32335502+jagmarques@users.noreply.github.com> Date: Sun, 25 Oct 2026 09:20:03 +0100 Subject: [PATCH 2/2] test(asqav): cover blank and corrupt log lines, null latency and build errors Six tests for edge cases the first commit left untested: verify skips blank lines and reports a non-JSON line, a restart on a blank file starts a fresh chain, a restart after a corrupt last line keeps logging while verify flags the corrupt line, latency is null without datetime timestamps, and an unexpected error while building a record neither raises nor advances the chain. Patch coverage for asqav.py goes from 94% to 99%; the remaining line is the unreachable return after the tail-read loop. --- tests/unit/integrations/asqav/test_asqav.py | 78 +++++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/tests/unit/integrations/asqav/test_asqav.py b/tests/unit/integrations/asqav/test_asqav.py index 1bd7e0c34f8..d271ade4860 100644 --- a/tests/unit/integrations/asqav/test_asqav.py +++ b/tests/unit/integrations/asqav/test_asqav.py @@ -108,6 +108,14 @@ def test_record_carries_usage_latency_and_call_metadata(tmp_path: Path) -> None: assert record["finish_reason"] == "stop" +def test_latency_is_null_when_times_are_not_datetimes(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + _logger(path).log_success_event(_kwargs(), _response(), None, None) + + [record] = _records(path) + assert record["latency_ms"] is None + + def test_verify_detects_an_edited_field(tmp_path: Path) -> None: path = tmp_path / "audit.jsonl" logger = _logger(path) @@ -154,6 +162,29 @@ def test_verify_reports_a_missing_file(tmp_path: Path) -> None: assert _logger(tmp_path / "audit.jsonl").verify_chain(str(missing)) == (False, f"log file not found: {missing}") +def test_verify_skips_blank_lines(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + _append(logger, 2) + first, second = path.read_text(encoding="utf-8").splitlines() + path.write_text(f"{first}\n\n{second}\n", encoding="utf-8") + + assert logger.verify_chain() == (True, "ok") + + +def test_verify_reports_a_line_that_is_not_json(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + _append(logger, 1) + with path.open("a", encoding="utf-8") as fh: + fh.write("not json\n") + + ok, reason = logger.verify_chain() + + assert ok is False + assert reason.startswith("verification error:") + + def test_chain_and_seq_resume_after_restart(tmp_path: Path) -> None: path = tmp_path / "audit.jsonl" _append(_logger(path), 5, prefix="first") @@ -184,6 +215,35 @@ def test_restart_resumes_when_last_record_exceeds_tail_window(tmp_path: Path) -> assert restarted.verify_chain() == (True, "ok") +def test_restart_on_a_blank_file_starts_a_new_chain(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + path.write_text("\n\n", encoding="utf-8") + + restarted = _logger(path) + _append(restarted, 1) + + [record] = _records(path) + assert (record["seq"], record["prev_hash"]) == (0, GENESIS_HASH) + assert restarted.verify_chain() == (True, "ok") + + +def test_restart_after_a_corrupt_last_line_keeps_logging_and_verify_flags_it(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + _append(_logger(path), 1) + with path.open("a", encoding="utf-8") as fh: + fh.write("{truncated\n") + + restarted = _logger(path) + _append(restarted, 1, prefix="after") + + lines = path.read_text(encoding="utf-8").splitlines() + assert len(lines) == 3 + assert json.loads(lines[2])["messages_digest"] is not None + ok, reason = restarted.verify_chain() + assert ok is False + assert reason.startswith("verification error:") + + def test_failure_event_is_logged(tmp_path: Path) -> None: path = tmp_path / "audit.jsonl" _logger(path).log_failure_event(_kwargs(), None, START, END) @@ -218,6 +278,24 @@ def test_failed_write_does_not_raise_or_advance_the_chain(tmp_path: Path) -> Non assert record["prev_hash"] == GENESIS_HASH +def test_unexpected_error_while_building_does_not_raise_or_advance_the_chain( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + path = tmp_path / "audit.jsonl" + logger = _logger(path) + + def _explode(_record: Mapping[str, object]) -> bool: + raise RuntimeError("boom") + + monkeypatch.setattr(logger, "_write_record", _explode) + logger.log_success_event(_kwargs(call_id="call-dropped"), _response(), START, END) + monkeypatch.undo() + logger.log_success_event(_kwargs(call_id="call-kept"), _response(), START, END) + + [record] = _records(path) + assert (record["call_id"], record["seq"], record["prev_hash"]) == ("call-kept", 0, GENESIS_HASH) + + def test_content_is_digested_by_default(tmp_path: Path) -> None: path = tmp_path / "audit.jsonl" _logger(path).log_success_event(_kwargs(content="my private prompt"), _response("my private answer"), START, END)