From 112b710aad37ddf779465aea2984b961f5c6909d Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Sun, 23 Aug 2026 19:46:34 +0200 Subject: [PATCH 1/4] Share bounded archive handling across channels Extract Telegram's ZIP handling into a channel-neutral helper with limits on entry count, expanded size, per-entry size, and compression ratio. This both closes decompression amplification and gives the Slack channel a safe implementation to build on. --- nerve/channels/archives.py | 176 +++++++++++++++++++++++++++++++++ nerve/channels/telegram.py | 109 ++++---------------- tests/test_channel_archives.py | 158 +++++++++++++++++++++++++++++ 3 files changed, 353 insertions(+), 90 deletions(-) create mode 100644 nerve/channels/archives.py create mode 100644 tests/test_channel_archives.py diff --git a/nerve/channels/archives.py b/nerve/channels/archives.py new file mode 100644 index 00000000..ca85629f --- /dev/null +++ b/nerve/channels/archives.py @@ -0,0 +1,176 @@ +"""Bounded ZIP unpacking, shared by the chat channels. + +Slack and Telegram both accept an attached archive and both cap the file +they download. That cap is on the *compressed* bytes, so it says nothing +about what the archive expands to: a few megabytes of zeros expand to +gigabytes, and reading them into base64 blocks exhausts the daemon. + +Every limit here is checked against the archive's own directory before any +entry is read, and the read itself is bounded as well, because a ZIP header +can under-report an entry's size. An entry past a limit is refused with a +line saying so; nothing is silently cut short. +""" + +from __future__ import annotations + +import base64 +import io +import logging +import zipfile + +logger = logging.getLogger(__name__) + +TEXT_EXTENSIONS: frozenset[str] = frozenset({ + ".txt", ".py", ".js", ".ts", ".jsx", ".tsx", ".json", ".yaml", ".yml", + ".toml", ".xml", ".html", ".htm", ".css", ".scss", ".less", + ".md", ".rst", ".csv", ".tsv", ".sql", ".sh", ".bash", ".zsh", + ".rb", ".go", ".rs", ".java", ".kt", ".c", ".cpp", ".h", ".hpp", + ".swift", ".lua", ".r", ".m", ".pl", ".php", ".env", ".ini", ".cfg", + ".conf", ".log", ".diff", ".patch", ".vue", ".svelte", +}) + +IMAGE_EXT_TO_MIME: dict[str, str] = { + ".jpg": "image/jpeg", ".jpeg": "image/jpeg", + ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp", +} + +# Inline text budget for the whole archive. +MAX_TEXT_SIZE = 512 * 1024 +# Files in one archive. A directory listing longer than this is a machine +# dump, not something a person meant to show the agent. +MAX_ENTRIES = 100 +# Uncompressed bytes for one entry, and for the archive as a whole. Both +# are what the prompt has to carry, so they are far below the ~20 MB +# compressed cap the channels put on the download. +MAX_ENTRY_SIZE = 20_000_000 +MAX_TOTAL_SIZE = 50_000_000 +# Uncompressed / compressed for one entry. Ordinary text reaches about 10; +# an archive built to expand reaches thousands. +MAX_RATIO = 100 + + +class _EntryTooLarge(Exception): + """An entry produced more bytes than its directory record promised.""" + + +def _read_bounded(zf: zipfile.ZipFile, info: zipfile.ZipInfo, limit: int) -> bytes: + """Read one entry, refusing more than *limit* bytes. + + The bound is on what comes out, not only on the size the central + directory declares, so the caller's budget holds even for an archive + whose records do not describe its contents. + """ + with zf.open(info) as handle: + raw = handle.read(limit + 1) + if len(raw) > limit: + raise _EntryTooLarge(info.filename) + return raw + + +def _refusal(info: zipfile.ZipInfo, reason: str) -> str: + return f"- {info.filename} ({info.file_size} bytes) [{reason}]" + + +def extract_zip(data: bytes, meta_line: str) -> tuple[list[dict[str, str]], str]: + """Unpack a ZIP one level — text inline, images and PDFs as blocks. + + Returns ``(content_blocks, context_text)``. ``meta_line`` is the caller's + one-line description of the archive and heads the context text. + """ + buf = io.BytesIO(data) + if not zipfile.is_zipfile(buf): + return [], f"{meta_line}\n(Invalid or corrupted ZIP archive)" + buf.seek(0) + + blocks: list[dict[str, str]] = [] + parts: list[str] = [meta_line] + try: + with zipfile.ZipFile(buf) as zf: + entries = [ + i for i in zf.infolist() + if not i.is_dir() and not i.filename.startswith("__MACOSX/") + ] + if len(entries) > MAX_ENTRIES: + return [], ( + f"{meta_line}\n(Archive holds {len(entries)} files; " + f"the limit is {MAX_ENTRIES})" + ) + + parts.append(f"Archive contains {len(entries)} file(s):") + total_text = 0 + total_read = 0 + for info in entries: + name = info.filename + size = info.file_size + ext = "" + if "." in name.rsplit("/", 1)[-1]: + ext = "." + name.rsplit(".", 1)[-1].lower() + wanted = ext in TEXT_EXTENSIONS or ext in IMAGE_EXT_TO_MIME or ext == ".pdf" + + if not wanted: + parts.append(f"- {name} ({size} bytes)") + continue + + # Every bound below is read off the central directory, so a + # refusal costs nothing but the listing itself. + if size > MAX_ENTRY_SIZE: + parts.append(_refusal(info, "too large to read")) + continue + if info.compress_size and size / info.compress_size > MAX_RATIO: + logger.warning( + "Refusing ZIP entry %s: %d bytes from %d compressed", + name, size, info.compress_size, + ) + parts.append(_refusal(info, "compression ratio too high")) + continue + if total_read + size > MAX_TOTAL_SIZE: + parts.append(_refusal(info, "archive size budget spent")) + continue + + if ext in TEXT_EXTENSIONS and total_text + size > MAX_TEXT_SIZE: + parts.append(_refusal(info, "text, too large to inline")) + continue + + try: + raw = _read_bounded(zf, info, size) + except _EntryTooLarge: + logger.warning( + "Refusing ZIP entry %s: it expands past the %d bytes " + "its header declares", name, size, + ) + parts.append(_refusal(info, "larger than it declares")) + continue + except RuntimeError: + # A password-protected archive fails the same way on + # every entry, so it is reported once, below. + raise + except Exception: + parts.append(_refusal(info, "read error")) + continue + total_read += len(raw) + + if ext in TEXT_EXTENSIONS: + total_text += len(raw) + parts.append( + f"--- {name} ({size} bytes) ---\n" + f"```\n{raw.decode('utf-8', errors='replace')}\n```" + ) + else: + is_pdf = ext == ".pdf" + blocks.append({ + "type": "base64", + "media_type": ( + "application/pdf" if is_pdf else IMAGE_EXT_TO_MIME[ext] + ), + "data": base64.b64encode(raw).decode("utf-8"), + }) + parts.append( + f"- {name} ({size} bytes) [{'PDF' if is_pdf else 'image'}]" + ) + except zipfile.BadZipFile: + return [], f"{meta_line}\n(Invalid or corrupted ZIP archive)" + except RuntimeError as e: + # Password-protected archives. + return [], f"{meta_line}\n(Cannot extract: {e})" + + return blocks, "\n".join(parts) diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index a0c970db..56a3e7ec 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -10,7 +10,6 @@ import asyncio import base64 import collections -import io import html as _html import logging import re @@ -18,7 +17,6 @@ import subprocess import sys import time -import zipfile from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, TYPE_CHECKING @@ -28,6 +26,12 @@ from telegram.constants import ChatAction, ParseMode from telegram.ext import Application, CallbackQueryHandler, CommandHandler, MessageHandler, MessageReactionHandler, filters +from nerve.channels.archives import ( + IMAGE_EXT_TO_MIME, + MAX_TEXT_SIZE, + TEXT_EXTENSIONS, + extract_zip, +) from nerve.channels.base import ( BaseChannel, ChannelCapability, @@ -1356,26 +1360,18 @@ async def _extract_sticker( "application/csv", } - # Extensions treated as text when MIME type is missing or generic - _TEXT_EXTENSIONS: set[str] = { - ".txt", ".py", ".js", ".ts", ".jsx", ".tsx", ".json", ".yaml", ".yml", - ".toml", ".xml", ".html", ".htm", ".css", ".scss", ".less", - ".md", ".rst", ".csv", ".tsv", ".sql", ".sh", ".bash", ".zsh", - ".rb", ".go", ".rs", ".java", ".kt", ".c", ".cpp", ".h", ".hpp", - ".swift", ".lua", ".r", ".m", ".pl", ".php", ".env", ".ini", ".cfg", - ".conf", ".log", ".diff", ".patch", ".vue", ".svelte", - } + # Extensions treated as text when MIME type is missing or generic. Shared + # with the Slack channel and with the archive unpacker, so a file type + # read inline here is read inline everywhere. + _TEXT_EXTENSIONS = TEXT_EXTENSIONS _IMAGE_MIMES: set[str] = {"image/jpeg", "image/png", "image/gif", "image/webp"} _ARCHIVE_MIMES: set[str] = {"application/zip", "application/x-zip-compressed"} - _IMAGE_EXT_TO_MIME: dict[str, str] = { - ".jpg": "image/jpeg", ".jpeg": "image/jpeg", - ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp", - } + _IMAGE_EXT_TO_MIME = IMAGE_EXT_TO_MIME - _MAX_TEXT_SIZE: int = 512 * 1024 # 512 KB — inline text cap + _MAX_TEXT_SIZE: int = MAX_TEXT_SIZE # 512 KB — inline text cap _MAX_DOWNLOAD_SIZE: int = 20_000_000 # ~20 MB — Telegram Bot API limit async def _extract_document( @@ -1469,7 +1465,12 @@ async def _extract_document( async def _extract_zip( self, doc: Any, file_name: str, meta_line: str, ) -> tuple[list[dict[str, str]], str]: - """Extract ZIP archive contents — text inline, images/PDFs as blocks.""" + """Extract ZIP archive contents — text inline, images/PDFs as blocks. + + The 20 MB download cap is on the compressed archive, so the bounds on + what it expands to live in :mod:`nerve.channels.archives`, shared with + the Slack channel. + """ try: tg_file = await doc.get_file() data = await tg_file.download_as_bytearray() @@ -1477,79 +1478,7 @@ async def _extract_zip( logger.warning("Failed to download ZIP %s: %s", file_name, e) return [], meta_line - buf = io.BytesIO(bytes(data)) - if not zipfile.is_zipfile(buf): - return [], f"{meta_line}\n(Invalid or corrupted ZIP archive)" - buf.seek(0) - - blocks: list[dict[str, str]] = [] - parts: list[str] = [meta_line] - - try: - with zipfile.ZipFile(buf) as zf: - entries = [ - i for i in zf.infolist() - if not i.is_dir() and not i.filename.startswith("__MACOSX/") - ] - parts.append(f"Archive contains {len(entries)} file(s):") - - total_text = 0 - for info in entries: - ename = info.filename - esize = info.file_size - eext = "" - if "." in ename.rsplit("/", 1)[-1]: - eext = "." + ename.rsplit(".", 1)[-1].lower() - - is_text = eext in self._TEXT_EXTENSIONS - is_image = eext in self._IMAGE_EXT_TO_MIME - is_pdf = eext == ".pdf" - - if is_text and total_text + esize <= self._MAX_TEXT_SIZE: - try: - raw = zf.read(info.filename) - total_text += len(raw) - text_content = raw.decode("utf-8", errors="replace") - parts.append( - f"--- {ename} ({esize} bytes) ---\n" - f"```\n{text_content}\n```" - ) - except Exception: - parts.append(f"- {ename} ({esize} bytes) [read error]") - elif is_text: - parts.append(f"- {ename} ({esize} bytes) [text, too large to inline]") - elif is_image: - try: - raw = zf.read(info.filename) - img_mime = self._IMAGE_EXT_TO_MIME.get(eext, "image/png") - blocks.append({ - "type": "base64", - "media_type": img_mime, - "data": base64.b64encode(raw).decode("utf-8"), - }) - parts.append(f"- {ename} ({esize} bytes) [image]") - except Exception: - parts.append(f"- {ename} ({esize} bytes) [read error]") - elif is_pdf: - try: - raw = zf.read(info.filename) - blocks.append({ - "type": "base64", - "media_type": "application/pdf", - "data": base64.b64encode(raw).decode("utf-8"), - }) - parts.append(f"- {ename} ({esize} bytes) [PDF]") - except Exception: - parts.append(f"- {ename} ({esize} bytes) [read error]") - else: - parts.append(f"- {ename} ({esize} bytes)") - except zipfile.BadZipFile: - return [], f"{meta_line}\n(Invalid or corrupted ZIP archive)" - except RuntimeError as e: - # Password-protected archives - return [], f"{meta_line}\n(Cannot extract: {e})" - - return blocks, "\n".join(parts) + return extract_zip(bytes(data), meta_line) async def _handle_message(self, update: Update, context: Any) -> None: """Handle incoming text and photo messages — delegate to router.""" diff --git a/tests/test_channel_archives.py b/tests/test_channel_archives.py new file mode 100644 index 00000000..333f2c59 --- /dev/null +++ b/tests/test_channel_archives.py @@ -0,0 +1,158 @@ +"""Bounded ZIP unpacking for the chat channels. + +A channel caps the archive it downloads at about 20 MB of *compressed* +bytes. That says nothing about what the archive expands to, so an +authorized user could hand the daemon a few megabytes that unpack to +gigabytes of base64 blocks. Every bound is read off the archive's own +directory before an entry is opened, so refusing one costs nothing. +""" + +from __future__ import annotations + +import io +import zipfile +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nerve.channels import archives +from nerve.channels.archives import extract_zip + +# A one-pixel PNG. Real bytes, so the image branch is genuinely exercised. +_PNG = bytes.fromhex( + "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4" + "890000000a49444154789c6360000002000100ffff03000006000557bfabd400" + "00000049454e44ae426082" +) + + +def _zip(entries: dict[str, bytes], compression=zipfile.ZIP_DEFLATED) -> bytes: + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", compression) as zf: + for name, data in entries.items(): + zf.writestr(name, data) + return buf.getvalue() + + +class TestOrdinaryArchives: + def test_a_text_file_is_inlined(self): + _, text = extract_zip(_zip({"notes.md": b"hello"}), "[File: a.zip]") + assert "notes.md" in text + assert "hello" in text + + def test_an_image_becomes_a_content_block(self): + blocks, text = extract_zip(_zip({"shot.png": _PNG}), "[File: a.zip]") + assert [b["media_type"] for b in blocks] == ["image/png"] + assert "[image]" in text + + def test_an_unknown_type_is_listed_without_being_read(self): + _, text = extract_zip(_zip({"blob.bin": b"\x00" * 32}), "[File: a.zip]") + assert "blob.bin" in text + + def test_a_corrupt_archive_is_reported(self): + blocks, text = extract_zip(b"not a zip at all", "[File: a.zip]") + assert blocks == [] + assert "Invalid or corrupted" in text + + def test_a_password_protected_archive_is_reported_once(self): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("a.txt", b"one") + zf.writestr("b.txt", b"two") + for info in zf.infolist(): + info.flag_bits |= 0x1 + blocks, text = extract_zip(buf.getvalue(), "[File: a.zip]") + assert blocks == [] + assert "Cannot extract" in text + assert "read error" not in text + + +class TestDecompressionBounds: + def test_too_many_entries_reject_the_whole_archive(self, monkeypatch): + monkeypatch.setattr(archives, "MAX_ENTRIES", 3) + data = _zip({f"f{i}.txt": b"x" for i in range(4)}) + blocks, text = extract_zip(data, "[File: a.zip]") + assert blocks == [] + assert "the limit is 3" in text + + def test_an_entry_over_the_size_cap_is_refused_not_truncated( + self, monkeypatch, + ): + monkeypatch.setattr(archives, "MAX_ENTRY_SIZE", 100) + data = _zip({"big.txt": b"a" * 500, "small.txt": b"ok"}) + _, text = extract_zip(data, "[File: a.zip]") + assert "big.txt (500 bytes) [too large to read]" in text + assert "a" * 500 not in text + # A refusal is per entry, so the rest of the archive still arrives. + assert "ok" in text + + def test_a_high_compression_ratio_is_refused(self): + # 4 MB of zeros compresses to a few kilobytes. The outer download + # cap never sees it; only the ratio does. + blocks, text = extract_zip( + _zip({"bomb.png": b"\x00" * 4_000_000}), "[File: a.zip]", + ) + assert blocks == [] + assert "compression ratio too high" in text + + def test_the_ratio_is_checked_before_anything_is_read(self, monkeypatch): + # The whole point of reading ZipInfo first: refusing must not cost + # the memory the refusal exists to save. + data = _zip({"bomb.png": b"\x00" * 4_000_000}) + + def _boom(*args, **kwargs): + raise AssertionError("the entry was opened") + + monkeypatch.setattr(zipfile.ZipFile, "open", _boom) + blocks, text = extract_zip(data, "[File: a.zip]") + assert blocks == [] + assert "compression ratio too high" in text + + def test_the_aggregate_budget_stops_later_entries(self, monkeypatch): + monkeypatch.setattr(archives, "MAX_TOTAL_SIZE", 120) + monkeypatch.setattr(archives, "MAX_RATIO", 100_000) + data = _zip({"a.txt": b"a" * 100, "b.txt": b"b" * 100}) + _, text = extract_zip(data, "[File: a.zip]") + assert "a" * 100 in text + assert "b.txt (100 bytes) [archive size budget spent]" in text + + def test_the_text_budget_refuses_rather_than_cuts(self, monkeypatch): + monkeypatch.setattr(archives, "MAX_TEXT_SIZE", 50) + monkeypatch.setattr(archives, "MAX_RATIO", 100_000) + _, text = extract_zip(_zip({"long.txt": b"c" * 200}), "[File: a.zip]") + assert "long.txt (200 bytes) [text, too large to inline]" in text + assert "c" * 200 not in text + + def test_a_refused_entry_still_appears_in_the_listing(self, monkeypatch): + monkeypatch.setattr(archives, "MAX_ENTRY_SIZE", 10) + _, text = extract_zip(_zip({"big.txt": b"a" * 500}), "[File: a.zip]") + assert "Archive contains 1 file(s):" in text + assert "big.txt" in text + + def test_a_stored_entry_at_ratio_one_is_allowed(self): + # Ratio 1 is what an already-compressed payload looks like; the + # guard must not turn into a size cap by another name. + data = _zip({"plain.txt": b"d" * 5000}, compression=zipfile.ZIP_STORED) + _, text = extract_zip(data, "[File: a.zip]") + assert "d" * 5000 in text + + +@pytest.mark.asyncio +class TestTelegramUsesTheBounds: + async def test_telegram_refuses_a_bomb_in_a_document(self): + from nerve.channels.telegram import TelegramChannel + from nerve.config import NerveConfig + + channel = TelegramChannel(lambda: NerveConfig(), router=MagicMock()) + tg_file = MagicMock() + tg_file.download_as_bytearray = AsyncMock( + return_value=bytearray(_zip({"bomb.png": b"\x00" * 4_000_000})), + ) + doc = MagicMock() + doc.get_file = AsyncMock(return_value=tg_file) + + blocks, text = await channel._extract_zip( + doc, "payload.zip", "[Document: payload.zip]", + ) + assert blocks == [] + assert "compression ratio too high" in text From 3b8fc6980d72102bb2835a61ac53ba75a35f501f Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 25 Aug 2026 12:32:10 +0200 Subject: [PATCH 2/4] Establish shared channel session and notification boundaries --- nerve/agent/engine.py | 9 ++ nerve/channels/router.py | 26 +++++ nerve/channels/telegram.py | 64 +++++------ .../v045_notification_deliveries.py | 36 +++++++ nerve/db/notifications.py | 92 +++++++++++++++- nerve/db/sessions.py | 34 ++++++ nerve/notifications/service.py | 100 ++++++++++++++++-- tests/test_db.py | 33 ++++++ tests/test_notification_lifecycle.py | 53 ++++++++++ 9 files changed, 395 insertions(+), 52 deletions(-) create mode 100644 nerve/db/migrations/v045_notification_deliveries.py diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 08bbe861..3b2a0c43 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -254,6 +254,7 @@ def __init__(self, config: NerveConfig, db: Database): # survives restarts without re-firing on every resume. self._observed_models: dict[str, str] = {} self._router = None # ChannelRouter — lazy-initialized via .router property + self._channel_runtimes: dict[str, Any] = {} # Gateways expose an ephemeral plaintext MCP listener on loopback for # co-located Codex processes. The gateway fills this after the listener # starts; the callable passed to CodexBackend reads it lazily. @@ -714,6 +715,14 @@ def register_channel(self, channel: Any) -> None: """Register a channel with the router.""" self.router.register(channel) + def register_channel_runtime(self, name: str, runtime: Any) -> None: + """Publish the lifecycle owner for a dynamically managed channel.""" + self._channel_runtimes[name] = runtime + + def get_channel_runtime(self, name: str) -> Any | None: + """Return a channel lifecycle owner, if one was installed.""" + return self._channel_runtimes.get(name) + # ------------------------------------------------------------------ # # File snapshot for diff tracking # # ------------------------------------------------------------------ # diff --git a/nerve/channels/router.py b/nerve/channels/router.py index 63e721a1..f904c754 100644 --- a/nerve/channels/router.py +++ b/nerve/channels/router.py @@ -71,6 +71,14 @@ def register(self, channel: BaseChannel) -> None: channel.name, channel.capabilities, ) + def unregister(self, channel: BaseChannel) -> bool: + """Remove *channel* only if it is still the registered instance.""" + if self._channels.get(channel.name) is not channel: + return False + del self._channels[channel.name] + logger.info("Unregistered channel: %s", channel.name) + return True + def get_channel(self, name: str) -> BaseChannel | None: """Get a registered channel by name.""" return self._channels.get(name) @@ -441,10 +449,28 @@ async def list_interactive_sessions( limit=limit, offset=offset, current_id=current_id, ) + async def list_conversation_sessions( + self, channel_key: str, limit: int = 20, + ) -> list[dict[str, Any]]: + """List live mappings for one exact conversation and its children.""" + return await self.engine.db.list_channel_sessions_for_conversation( + channel_key, + limit=limit, + exclude_statuses=("archived", "stopped"), + ) + + async def stop_session(self, session_id: str) -> bool: + """Stop a session through the engine lifecycle boundary.""" + return await self.engine.stop_session(session_id) + async def set_session_starred(self, session_id: str, starred: bool) -> bool: """Star/unstar a session. Starred sessions are never auto-archived.""" return await self.engine.sessions.set_starred(session_id, starred) + async def toggle_session_starred(self, session_id: str) -> bool: + """Toggle a session's starred state.""" + return await self.engine.sessions.toggle_starred(session_id) + async def get_session(self, session_id: str) -> dict[str, Any] | None: """Fetch a session row (title/status/…), or None if it is gone.""" return await self.engine.db.get_session(session_id) diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index 56a3e7ec..43c575a2 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -1211,7 +1211,7 @@ async def _handle_new_session(self, update: Update, context: Any) -> None: # Stop the current session before creating a new one prev = await self.router.get_last_session(channel_key) if prev: - stopped = await self.router.engine.stop_session(prev) + stopped = await self.router.stop_session(prev) if stopped: await update.message.reply_text( f"Stopped session `{prev}`.", @@ -1240,7 +1240,7 @@ async def _handle_stop(self, update: Update, context: Any) -> None: await update.message.reply_text("No active session.") return - stopped = await self.router.engine.stop_session(session_id) + stopped = await self.router.stop_session(session_id) if stopped: await update.message.reply_text( f"Stopped session `{session_id}`.", @@ -1882,19 +1882,21 @@ async def _handle_callback_query(self, update: Update, context: Any) -> None: await query.answer("Service unavailable", show_alert=True) return - success = await self._notification_service.handle_answer( - notification_id=notification_id, - answer=answer, - answered_by="telegram", + result = await self._notification_service.answer_delivered_notification( + notification_id, + answer, + channel="telegram", + target=str(update.effective_chat.id), + actor=str(query.from_user.id), ) - if success: + if result: status_line = f"\u2705 Answered: {answer}" toast = f"Answered: {answer}" # A snooze keeps the row pending with redeliver_at stamped \u2014 # confirm on the card that it will come back, instead of the # generic answered state (which read as "handled, gone"). - snoozed_until = await self._get_snoozed_until(notification_id) + snoozed_until = self._snoozed_until(result) if snoozed_until: status_line = ( f"\U0001F4A4 Snoozed until {snoozed_until} \u2014 will resurface" @@ -1912,26 +1914,17 @@ async def _handle_callback_query(self, update: Update, context: Any) -> None: else: await query.answer("Already answered or expired", show_alert=True) - async def _get_snoozed_until(self, notification_id: str) -> str | None: - """Return a human-readable re-delivery time if the row was snoozed. - - After ``handle_answer`` succeeds, a snoozed approval is the only - outcome that leaves the row ``pending`` with ``redeliver_at`` - set. Rendered in the host's local timezone. None when the answer - was a final decision (or anything fails \u2014 this is cosmetic). - """ + @staticmethod + def _snoozed_until(notification: dict[str, Any]) -> str | None: + """Render the next delivery time when an answer snoozed the row.""" try: - notif = await self._notification_service.db.get_notification( - notification_id, - ) if ( - not notif - or notif.get("status") != "pending" - or not notif.get("redeliver_at") + notification.get("status") != "pending" + or not notification.get("redeliver_at") ): return None from datetime import datetime - dt = datetime.fromisoformat(notif["redeliver_at"]) + dt = datetime.fromisoformat(notification["redeliver_at"]) return dt.astimezone().strftime("%Y-%m-%d %H:%M %Z") except Exception: return None @@ -1950,21 +1943,16 @@ async def _handle_reply(self, update: Update, context: Any) -> None: answer_text = " ".join(context.args) - pending = await self._notification_service.db.list_notifications( - status="pending", type="question", limit=1, - ) - if not pending: - await update.message.reply_text("No pending questions.") - return - - notification_id = pending[0]["id"] - success = await self._notification_service.handle_answer( - notification_id=notification_id, - answer=answer_text, - answered_by="telegram", + result = await self._notification_service.answer_latest_question( + answer_text, + channel="telegram", + target=str(update.effective_chat.id), + actor=str(update.effective_user.id), ) - if success: - await update.message.reply_text(f"Answer recorded for: {pending[0]['title']}") + if result: + await update.message.reply_text( + f"Answer recorded for: {result['title']}", + ) else: - await update.message.reply_text("Failed to record answer.") + await update.message.reply_text("No pending questions in this chat.") diff --git a/nerve/db/migrations/v045_notification_deliveries.py b/nerve/db/migrations/v045_notification_deliveries.py new file mode 100644 index 00000000..584d3021 --- /dev/null +++ b/nerve/db/migrations/v045_notification_deliveries.py @@ -0,0 +1,36 @@ +"""V45: Transport-neutral notification delivery records.""" + +from __future__ import annotations + +import logging + +import aiosqlite + +logger = logging.getLogger(__name__) + +SQL = """ +CREATE TABLE IF NOT EXISTS notification_deliveries ( + notification_id TEXT NOT NULL, + channel TEXT NOT NULL, + target TEXT NOT NULL DEFAULT '', + message_id TEXT NOT NULL DEFAULT '', + delivered_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (notification_id, channel, target), + FOREIGN KEY (notification_id) REFERENCES notifications(id) ON DELETE CASCADE +); +CREATE INDEX IF NOT EXISTS idx_notification_deliveries_scope + ON notification_deliveries(channel, target, delivered_at DESC); +""" + + +async def up(db: aiosqlite.Connection) -> None: + await db.executescript(SQL) + await db.execute( + """INSERT OR IGNORE INTO notification_deliveries + (notification_id, channel, target, message_id, delivered_at) + SELECT id, 'telegram', telegram_chat_id, + COALESCE(telegram_message_id, ''), created_at + FROM notifications + WHERE telegram_chat_id IS NOT NULL AND telegram_chat_id != ''""", + ) + logger.info("V45 migration: added scoped notification deliveries") diff --git a/nerve/db/notifications.py b/nerve/db/notifications.py index 17239d13..2f01cbed 100644 --- a/nerve/db/notifications.py +++ b/nerve/db/notifications.py @@ -6,6 +6,17 @@ from datetime import datetime, timezone +def _loads_object(raw: object) -> dict: + """Parse a JSON column into a dict, tolerating NULL and bad values.""" + if not raw: + return {} + try: + parsed = json.loads(raw) + except (TypeError, ValueError): + return {} + return parsed if isinstance(parsed, dict) else {} + + class NotificationStore: """Mixin providing notification CRUD operations.""" @@ -96,24 +107,95 @@ async def list_notifications( return [dict(row) async for row in cursor] async def answer_notification( - self, notification_id: str, answer: str, answered_by: str, + self, + notification_id: str, + answer: str, + answered_by: str, + actor: str | None = None, ) -> bool: + """Record an answer and retain the transport actor in metadata.""" now = datetime.now(timezone.utc).isoformat() async with self._atomic(): async with self.db.execute( - "SELECT id FROM notifications WHERE id = ? AND status = 'pending'", + """SELECT metadata FROM notifications + WHERE id = ? AND status = 'pending'""", (notification_id,), ) as cursor: - if not await cursor.fetchone(): + row = await cursor.fetchone() + if not row: return False + metadata = _loads_object(row[0]) + if actor: + metadata["answered_by_actor"] = actor await self.db.execute( """UPDATE notifications - SET answer = ?, answered_by = ?, answered_at = ?, status = 'answered' + SET answer = ?, answered_by = ?, answered_at = ?, + status = 'answered', metadata = ? WHERE id = ?""", - (answer, answered_by, now, notification_id), + ( + answer, + answered_by, + now, + json.dumps(metadata), + notification_id, + ), ) return True + async def record_notification_delivery( + self, + notification_id: str, + channel: str, + target: str = "", + message_id: str = "", + ) -> None: + """Record or replace a delivery within one transport target.""" + now = datetime.now(timezone.utc).isoformat() + await self._write( + """INSERT INTO notification_deliveries + (notification_id, channel, target, message_id, delivered_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(notification_id, channel, target) DO UPDATE SET + message_id = excluded.message_id, + delivered_at = excluded.delivered_at""", + (notification_id, channel, target, message_id, now), + ) + + async def get_notification_delivery( + self, + notification_id: str, + channel: str, + target: str, + ) -> dict | None: + """Return a notification's delivery in one exact channel target.""" + async with self.db.execute( + """SELECT * FROM notification_deliveries + WHERE notification_id = ? AND channel = ? AND target = ?""", + (notification_id, channel, target), + ) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + + async def find_pending_question_for_delivery( + self, + channel: str, + target: str, + ) -> dict | None: + """Find the newest pending question delivered to one exact target.""" + async with self.db.execute( + """SELECT n.*, s.title AS session_title + FROM notification_deliveries d + JOIN notifications n ON n.id = d.notification_id + LEFT JOIN sessions s ON n.session_id = s.id + WHERE d.channel = ? AND d.target = ? + AND n.status = 'pending' AND n.type = 'question' + ORDER BY n.created_at DESC + LIMIT 1""", + (channel, target), + ) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + async def dismiss_notification(self, notification_id: str) -> bool: async with self._atomic(): async with self.db.execute( diff --git a/nerve/db/sessions.py b/nerve/db/sessions.py index 17e34190..9ba3b7fc 100644 --- a/nerve/db/sessions.py +++ b/nerve/db/sessions.py @@ -392,6 +392,40 @@ async def get_channel_session(self, channel_key: str) -> dict | None: row = await cursor.fetchone() return dict(row) if row else None + async def list_channel_sessions_for_conversation( + self, + channel_key: str, + limit: int = 20, + exclude_statuses: tuple[str, ...] = (), + ) -> list[dict]: + """List one conversation mapping and delimiter-bounded descendants. + + Channel keys are otherwise opaque. A descendant is explicitly a key + beginning with ``:``; arbitrary string prefixes such as + ``chat:1`` and ``chat:12`` never share a scope. + """ + escaped = ( + channel_key.replace("\\", "\\\\") + .replace("%", "\\%") + .replace("_", "\\_") + ) + sql = """ + SELECT cs.channel_key, cs.session_id, cs.updated_at, + s.title, s.status, s.starred + FROM channel_sessions cs + JOIN sessions s ON s.id = cs.session_id + WHERE (cs.channel_key = ? OR cs.channel_key LIKE ? ESCAPE '\\') + """ + params: list = [channel_key, f"{escaped}:%"] + if exclude_statuses: + placeholders = ",".join("?" for _ in exclude_statuses) + sql += f" AND s.status NOT IN ({placeholders})" + params.extend(exclude_statuses) + sql += " ORDER BY cs.updated_at DESC LIMIT ?" + params.append(limit) + async with self.db.execute(sql, tuple(params)) as cursor: + return [dict(row) async for row in cursor] + async def set_channel_session(self, channel_key: str, session_id: str) -> None: """Persist a channel-to-session mapping.""" now = datetime.now(timezone.utc).isoformat() diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index 0cd9ee0d..fc7e6272 100644 --- a/nerve/notifications/service.py +++ b/nerve/notifications/service.py @@ -411,17 +411,72 @@ async def propose_action( return {"notification_id": notification_id, "status": "sent"} # ------------------------------------------------------------------ # - # Answer routing (called by REST API / Telegram callback) # + # Answer routing # # ------------------------------------------------------------------ # + async def answer_delivered_notification( + self, + notification_id: str, + answer: str, + *, + channel: str, + target: str, + actor: str | None = None, + ) -> dict[str, Any] | None: + """Answer an explicit notification within its delivery scope. + + The post-answer row lets an adapter render the result without reaching + through the service to its database. + """ + delivery = await self.db.get_notification_delivery( + notification_id, channel, target, + ) + if not delivery: + return None + accepted = await self.handle_answer( + notification_id, + answer, + answered_by=channel, + actor=actor, + ) + if not accepted: + return None + return await self.db.get_notification(notification_id) + + async def answer_latest_question( + self, + answer: str, + *, + channel: str, + target: str, + actor: str | None = None, + ) -> dict[str, Any] | None: + """Answer the newest pending question delivered to one exact target.""" + pending = await self.db.find_pending_question_for_delivery(channel, target) + if not pending: + return None + accepted = await self.handle_answer( + pending["id"], + answer, + answered_by=channel, + actor=actor, + ) + if not accepted: + return None + return await self.db.get_notification(pending["id"]) + async def handle_answer( self, notification_id: str, answer: str, answered_by: str, + actor: str | None = None, ) -> bool: """Process a user's answer to a question or approval. + ``answered_by`` identifies the transport; ``actor`` identifies the + person within that transport and is retained for audit. + - For ``type=approval`` rows: look up the dispatcher in the handler registry, run it, audit-log the outcome, then flip the row's status. Snooze answers keep the row pending and @@ -438,16 +493,17 @@ async def handle_answer( if notif.get("type") == "approval": return await self._handle_approval_answer( - notif, answer, answered_by, + notif, answer, answered_by, actor, ) success = await self.db.answer_notification( - notification_id, answer, answered_by, + notification_id, answer, answered_by, actor=actor, ) if not success: return False session_id = notif["session_id"] + attribution = {"answered_by_actor": actor} if actor else {} from nerve.agent.streaming import broadcaster @@ -468,6 +524,7 @@ async def handle_answer( "session_id": session_id, "answer": answer, "answered_by": answered_by, + **attribution, }) return True @@ -483,6 +540,7 @@ async def handle_answer( "answer": answer, "answered_by": answered_by, "content": injected_message, + **attribution, }) # Dispatch unconditionally — ``engine.run`` serializes per @@ -510,6 +568,7 @@ async def handle_answer( "session_id": session_id, "answer": answer, "answered_by": answered_by, + **attribution, }) return True @@ -519,6 +578,7 @@ async def _handle_approval_answer( notif: dict[str, Any], answer: str, answered_by: str, + actor: str | None = None, ) -> bool: """Route an approval answer through the dispatcher registry.""" notification_id = notif["id"] @@ -580,7 +640,10 @@ async def _handle_approval_answer( }, ) - await self._append_approval_audit(result.audit_event) + attribution: dict[str, Any] = {"answered_by": answered_by} + if actor: + attribution["answered_by_actor"] = actor + await self._append_approval_audit({**result.audit_event, **attribution}) # Snooze keeps the row pending and stamps ``redeliver_at`` so # the periodic maintenance tick (:meth:`redeliver_due`) fans it @@ -614,7 +677,7 @@ async def _handle_approval_answer( ) else: await self.db.answer_notification( - notification_id, answer, answered_by, + notification_id, answer, answered_by, actor=actor, ) from nerve.agent.streaming import broadcaster @@ -623,9 +686,9 @@ async def _handle_approval_answer( "notification_id": notification_id, "session_id": session_id, "answer": answer, - "answered_by": answered_by, "approval_status": "snoozed" if snoozed else "answered", "dispatch_ok": result.ok, + **attribution, } if snoozed: payload["snooze_until"] = snooze_until @@ -771,6 +834,9 @@ async def _deliver(channel_name: str) -> str | None: option_labels=option_labels, extra=extra_web, ) + await self.db.record_notification_delivery( + notification_id, "web", + ) return "web" elif channel_name == "telegram": msg_id = await self._deliver_telegram( @@ -783,7 +849,7 @@ async def _deliver(channel_name: str) -> str | None: notification_id, telegram_message_id=str(msg_id), ) - return "telegram" + return "telegram" if msg_id else None except Exception as e: logger.error( "Failed to deliver %s to %s: %s", @@ -862,6 +928,7 @@ async def _broadcast_silenced_web( await self.db.update_notification( notification_id, channels_delivered=json.dumps(["web"]), ) + await self.db.record_notification_delivery(notification_id, "web") message: dict[str, Any] = { "type": "notification", @@ -974,6 +1041,12 @@ async def _deliver_telegram( channel = self._get_telegram_channel() if channel: channel._cache_message(int(msg_id), chat_id, text) + await self.db.record_notification_delivery( + notification_id, + "telegram", + target=str(chat_id), + message_id=str(msg_id), + ) return msg_id @@ -1286,13 +1359,22 @@ async def _edit_telegram_expired(self, notif: dict[str, Any]) -> None: now-dead inline keyboard. Telegram refuses edits on old messages (>48h) — all failures are swallowed by design. """ - message_id = notif.get("telegram_message_id") + target = str(notif.get("telegram_chat_id") or "") + delivery = None + if target: + delivery = await self.db.get_notification_delivery( + notif["id"], "telegram", target, + ) + message_id = ( + (delivery or {}).get("message_id") + or notif.get("telegram_message_id") + ) if not message_id: return bot = self._get_telegram_bot() if not bot: return - chat_id = notif.get("telegram_chat_id") or self._resolve_telegram_chat_id() + chat_id = target or self._resolve_telegram_chat_id() if not chat_id: return diff --git a/tests/test_db.py b/tests/test_db.py index 2f8514dd..3e4f453a 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -277,6 +277,39 @@ async def test_overwrite(self, db: Database): row = await db.get_channel_session("tg:1") assert row["session_id"] == "ch-b" + async def test_conversation_scope_is_delimiter_bounded_before_limit( + self, db: Database, + ): + for session_id, key in ( + ("root", "chat:C1"), + ("thread", "chat:C1:1700.1"), + ): + await db.create_session(session_id) + await db.set_channel_session(key, session_id) + + # These sort later and outnumber the limit. A raw prefix query would + # select them first and crowd the real conversation out. + for index in range(25): + session_id = f"sibling-{index}" + await db.create_session(session_id) + await db.set_channel_session(f"chat:C12:{index}", session_id) + + rows = await db.list_channel_sessions_for_conversation( + "chat:C1", limit=2, + ) + assert {row["channel_key"] for row in rows} == { + "chat:C1", "chat:C1:1700.1", + } + + async def test_conversation_scope_escapes_like_wildcards(self, db: Database): + await db.create_session("literal") + await db.create_session("wildcard") + await db.set_channel_session("chat:C_1:thread", "literal") + await db.set_channel_session("chat:CX1:thread", "wildcard") + + rows = await db.list_channel_sessions_for_conversation("chat:C_1") + assert [row["session_id"] for row in rows] == ["literal"] + @pytest.mark.asyncio class TestMessages: diff --git a/tests/test_notification_lifecycle.py b/tests/test_notification_lifecycle.py index e20ec8f4..38d3800d 100644 --- a/tests/test_notification_lifecycle.py +++ b/tests/test_notification_lifecycle.py @@ -166,6 +166,59 @@ async def test_v037_columns_exist_with_defaults(self, db: Database): assert notif["redeliver_at"] is None assert notif["redelivery_count"] == 0 + async def test_v045_delivery_scope_exists(self, db: Database): + async with db.db.execute( + "PRAGMA table_info(notification_deliveries)", + ) as cur: + cols = {row[1] async for row in cur} + assert {"notification_id", "channel", "target", "message_id"} <= cols + + +@pytest.mark.asyncio +class TestScopedAnswers: + async def test_latest_question_is_scoped_to_delivery_target( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("wanted-session", source="external") + await db.create_session("other-session", source="external") + await db.create_notification( + "wanted", "wanted-session", "question", "Wanted question", + ) + await db.record_notification_delivery( + "wanted", "telegram", target="100", message_id="1", + ) + await db.create_notification( + "newer-other", "other-session", "question", "Other question", + ) + await db.record_notification_delivery( + "newer-other", "telegram", target="200", message_id="2", + ) + + result = await svc.answer_latest_question( + "yes", channel="telegram", target="100", actor="42", + ) + + assert result and result["id"] == "wanted" + assert (await db.get_notification("newer-other"))["status"] == "pending" + metadata = json.loads(result["metadata"]) + assert metadata["answered_by_actor"] == "42" + + async def test_explicit_answer_rejects_a_different_delivery_target( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1", source="external") + await db.create_notification("n1", "s1", "question", "Question") + await db.record_notification_delivery("n1", "telegram", target="100") + + result = await svc.answer_delivered_notification( + "n1", "yes", channel="telegram", target="200", actor="42", + ) + + assert result is None + assert (await db.get_notification("n1"))["status"] == "pending" + # ---------------------------------------------------------------------- # Snooze semantics From 5bf72d4ac3379ea2084b64996f28d39fd667f29a Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 25 Aug 2026 14:58:43 +0200 Subject: [PATCH 3/4] Bound archive expansion by what the prompt carries MAX_TOTAL_SIZE was above the compressed download cap, so a 703 KB upload of 100 padded PDF entries passed every check and produced 66.7 MB of base64 in one prompt. Set the entry and archive caps from what the prompt can carry once base64 adds its third. The compression ratio check refused small files that repeat for ordinary reasons: a 103 KB generated .py and a 105 KB log of one repeated line both scored above the ratio and were dropped, where Telegram inlined them before. Apply the ratio only above RATIO_FLOOR, where the size caps have not already bounded the entry. Check the text budget before the archive budget so a refusal names the limit the entry reached. --- nerve/channels/archives.py | 35 ++++++++++++++++++++++------------ tests/test_channel_archives.py | 24 +++++++++++++++++++++-- 2 files changed, 45 insertions(+), 14 deletions(-) diff --git a/nerve/channels/archives.py b/nerve/channels/archives.py index ca85629f..a6572079 100644 --- a/nerve/channels/archives.py +++ b/nerve/channels/archives.py @@ -39,14 +39,20 @@ # Files in one archive. A directory listing longer than this is a machine # dump, not something a person meant to show the agent. MAX_ENTRIES = 100 -# Uncompressed bytes for one entry, and for the archive as a whole. Both -# are what the prompt has to carry, so they are far below the ~20 MB -# compressed cap the channels put on the download. -MAX_ENTRY_SIZE = 20_000_000 -MAX_TOTAL_SIZE = 50_000_000 -# Uncompressed / compressed for one entry. Ordinary text reaches about 10; -# an archive built to expand reaches thousands. +# Uncompressed bytes for one entry, and for the archive as a whole. +# Images and PDFs reach the model as base64, which is 4/3 of the bytes +# read, so the total is set from what the prompt can carry after that +# expansion. The channels cap the download at about 20 MB of compressed +# bytes, which says nothing about what the archive expands to. +MAX_ENTRY_SIZE = 5_000_000 +MAX_TOTAL_SIZE = 12_000_000 +# Uncompressed / compressed for one entry, checked only above RATIO_FLOOR. +# Small files reach a high ratio for ordinary reasons, such as generated +# code or a log of one repeated line, and refusing those loses real +# content. MAX_ENTRY_SIZE and MAX_TOTAL_SIZE already bound what any entry +# adds to the prompt, so the ratio only has to catch the large entries. MAX_RATIO = 100 +RATIO_FLOOR = 1_000_000 class _EntryTooLarge(Exception): @@ -116,20 +122,25 @@ def extract_zip(data: bytes, meta_line: str) -> tuple[list[dict[str, str]], str] if size > MAX_ENTRY_SIZE: parts.append(_refusal(info, "too large to read")) continue - if info.compress_size and size / info.compress_size > MAX_RATIO: + if ( + size > RATIO_FLOOR + and info.compress_size + and size / info.compress_size > MAX_RATIO + ): logger.warning( "Refusing ZIP entry %s: %d bytes from %d compressed", name, size, info.compress_size, ) parts.append(_refusal(info, "compression ratio too high")) continue - if total_read + size > MAX_TOTAL_SIZE: - parts.append(_refusal(info, "archive size budget spent")) - continue - + # The type-specific budget is checked first so a refusal names + # the limit the entry actually hit. if ext in TEXT_EXTENSIONS and total_text + size > MAX_TEXT_SIZE: parts.append(_refusal(info, "text, too large to inline")) continue + if total_read + size > MAX_TOTAL_SIZE: + parts.append(_refusal(info, "archive size budget spent")) + continue try: raw = _read_bounded(zf, info, size) diff --git a/tests/test_channel_archives.py b/tests/test_channel_archives.py index 333f2c59..a384b1e5 100644 --- a/tests/test_channel_archives.py +++ b/tests/test_channel_archives.py @@ -10,6 +10,7 @@ from __future__ import annotations import io +import os import zipfile from unittest.mock import AsyncMock, MagicMock @@ -110,7 +111,6 @@ def _boom(*args, **kwargs): def test_the_aggregate_budget_stops_later_entries(self, monkeypatch): monkeypatch.setattr(archives, "MAX_TOTAL_SIZE", 120) - monkeypatch.setattr(archives, "MAX_RATIO", 100_000) data = _zip({"a.txt": b"a" * 100, "b.txt": b"b" * 100}) _, text = extract_zip(data, "[File: a.zip]") assert "a" * 100 in text @@ -118,11 +118,31 @@ def test_the_aggregate_budget_stops_later_entries(self, monkeypatch): def test_the_text_budget_refuses_rather_than_cuts(self, monkeypatch): monkeypatch.setattr(archives, "MAX_TEXT_SIZE", 50) - monkeypatch.setattr(archives, "MAX_RATIO", 100_000) _, text = extract_zip(_zip({"long.txt": b"c" * 200}), "[File: a.zip]") assert "long.txt (200 bytes) [text, too large to inline]" in text assert "c" * 200 not in text + def test_a_small_repetitive_text_file_still_arrives(self): + # Repetition is ordinary: generated code, or a log of one repeated + # error line. Both compress far past MAX_RATIO while staying small + # enough that no budget is at stake, so both must be inlined. + body = b"def handler(event):\n return {'ok': True}\n" * 2400 + _, text = extract_zip(_zip({"app.py": body}), "[File: a.zip]") + assert "compression ratio too high" not in text + assert "def handler(event):" in text + + def test_the_archive_budget_bounds_what_base64_adds_to_the_prompt(self): + # Each entry is under MAX_ENTRY_SIZE, and the entropy padding keeps + # it under MAX_RATIO. Only the aggregate budget stops a small upload + # from filling the prompt with base64. + body = os.urandom(6000) + b"\0" * 494_000 + data = _zip({f"doc{i:03}.pdf": body for i in range(100)}) + blocks, text = extract_zip(data, "[File: a.zip]") + encoded = sum(len(b["data"]) for b in blocks) + assert len(data) < 2_000_000, "the upload itself stays small" + assert encoded < 20_000_000, f"base64 reached {encoded} bytes" + assert "archive size budget spent" in text + def test_a_refused_entry_still_appears_in_the_listing(self, monkeypatch): monkeypatch.setattr(archives, "MAX_ENTRY_SIZE", 10) _, text = extract_zip(_zip({"big.txt": b"a" * 500}), "[File: a.zip]") From 6ac958da5d6d3742e7fb7b67b1ff9c0b4ba0f74d Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 25 Aug 2026 16:29:56 +0200 Subject: [PATCH 4/4] Add transport-neutral access matching A channel needs to decide whether a sender and a conversation may reach the agent, and the matching rules do not depend on the transport: allow and deny lists of case-insensitive globs, deny winning over allow, and a refusal when an identity could not be resolved well enough to check a deny list against. Aliases are split by who controls them. A deny rule may match any alias. An allow rule may match only the aliases the subject cannot set for itself, because a grant resting on a self-set name lets the subject pick its own access. Each transport decides which of its aliases fall on which side. No transport uses this yet. The Slack channel composes its user, channel, and direct-message rules on top of it. --- nerve/channels/access.py | 165 ++++++++++++++++++++++ tests/test_channel_access.py | 266 +++++++++++++++++++++++++++++++++++ 2 files changed, 431 insertions(+) create mode 100644 nerve/channels/access.py create mode 100644 tests/test_channel_access.py diff --git a/nerve/channels/access.py b/nerve/channels/access.py new file mode 100644 index 00000000..3955709e --- /dev/null +++ b/nerve/channels/access.py @@ -0,0 +1,165 @@ +"""Pattern-matching primitives for transport access policies. + +A gate matches an identity's platform ID and resolved aliases using +case-insensitive globs. Deny wins, a non-empty allow list must match, and an +incomplete identity cannot clear a deny list. + +Aliases are split by who controls them. A deny rule may match any of them. +An allow rule may match only the ones the subject cannot set for itself, +because a grant that rests on a self-set name lets the subject choose its +own access. +""" + +from __future__ import annotations + +import fnmatch +from dataclasses import dataclass, field +from typing import Callable + + +def _norm(value: object) -> str: + """Normalize a value for case-insensitive matching.""" + return str(value).strip().lower() + + +def _matches(value: str, pattern: str) -> bool: + """Case-insensitive shell-glob match of one value against one pattern.""" + return fnmatch.fnmatchcase(_norm(value), _norm(pattern)) + + +@dataclass(frozen=True) +class Identity: + """A platform ID and resolved names used for policy matching. + + ``names`` hold identity the platform or its administrators control, so a + grant may rest on them. ``self_set_names`` hold whatever the subject can + set without approval; a deny rule may match those, but an allow rule may + not, because the subject would then choose its own access. + + ``complete`` means the candidates cover every name relevant to the active + patterns. ID-only policies therefore need no lookup, while deny lists reject + an incomplete identity. + """ + + id: str = "" + names: tuple[str, ...] = () + self_set_names: tuple[str, ...] = () + complete: bool = True + + @property + def candidates(self) -> tuple[str, ...]: + """Every string an allow rule may grant on.""" + return tuple(v for v in (self.id, *self.names) if v) + + @property + def deny_candidates(self) -> tuple[str, ...]: + """Every string a deny rule may refuse on, self-set names included.""" + return tuple( + v for v in (self.id, *self.names, *self.self_set_names) if v + ) + + def __str__(self) -> str: + label = next(iter((*self.names, *self.self_set_names)), "") + if label and self.id: + return f"{label} ({self.id})" + return label or self.id or "unknown" + + +@dataclass(frozen=True) +class Decision: + """The outcome of a policy check, with a reason fit for a log line.""" + + allowed: bool + reason: str = "" + + def __bool__(self) -> bool: + return self.allowed + + +@dataclass +class PatternGate: + """Allow/deny matching for a labeled identity.""" + + label: str + allow: list[str] = field(default_factory=list) + deny: list[str] = field(default_factory=list) + + def any_deny_pattern(self, predicate: Callable[[str], bool]) -> bool: + """Whether any non-empty deny pattern satisfies *predicate*.""" + return any(predicate(p) for p in self.deny if p) + + def check(self, who: Identity) -> Decision: + """Decide whether *who* passes this gate.""" + candidates = who.candidates + + for pattern in self.deny: + for value in who.deny_candidates: + if _matches(value, pattern): + return Decision( + False, + f"{self.label} {who} matches deny pattern {pattern!r}", + ) + + # A deny list is only meaningful against a candidate set known to + # cover it. Refuse rather than let an unread name walk past the list + # that names it. + if self.deny and not who.complete: + return Decision( + False, + f"{self.label} {who} could not be fully identified, so the " + f"deny list cannot be checked", + ) + + if self.allow: + if not candidates: + return Decision( + False, f"{self.label} is unidentified and an allow list is set", + ) + for pattern in self.allow: + for value in candidates: + if _matches(value, pattern): + return Decision( + True, + f"{self.label} {who} matches allow pattern {pattern!r}", + ) + # Say so when the only match was on a name the subject sets, + # otherwise the rule looks broken rather than declined. + for pattern in self.allow: + for value in who.self_set_names: + if _matches(value, pattern): + return Decision( + False, + f"{self.label} {who} matches allow pattern " + f"{pattern!r} only on a profile name it sets " + f"itself; grant on the id, handle, or email", + ) + return Decision(False, f"{self.label} {who} is not on the allow list") + + return Decision(True, "") + + +def needs_name_resolution( + *gates: PatternGate, is_id: Callable[[str], bool] | None = None, +) -> bool: + """Whether any gate pattern requires names beyond the platform ID. + + ``is_id`` recognizes literal IDs. Omitting it forces resolution, as does + any glob; extra lookups are safer than skipping one needed by a deny rule. + """ + for gate in gates: + for pattern in (*gate.allow, *gate.deny): + if not pattern: + continue + if any(c in pattern for c in "*?["): + return True + if is_id is None or not is_id(pattern): + return True + return False + + +__all__ = [ + "Decision", + "Identity", + "PatternGate", + "needs_name_resolution", +] diff --git a/tests/test_channel_access.py b/tests/test_channel_access.py new file mode 100644 index 00000000..f052fc4f --- /dev/null +++ b/tests/test_channel_access.py @@ -0,0 +1,266 @@ +"""Transport-neutral identity and access-pattern matching.""" + +from __future__ import annotations + +import re + +import pytest + +from nerve.channels.access import ( + Identity, + PatternGate, + needs_name_resolution, +) + + +class TestIdentity: + def test_candidates_drop_empty_names(self): + who = Identity(id="opaque-1", names=("", "alex", "")) + assert who.candidates == ("opaque-1", "alex") + + def test_str_prefers_a_name_but_keeps_the_id(self): + assert str(Identity(id="opaque-1", names=("alex",))) == "alex (opaque-1)" + assert str(Identity(id="opaque-1")) == "opaque-1" + assert str(Identity()) == "unknown" + + def test_str_falls_back_to_a_self_set_name(self): + who = Identity(id="opaque-1", self_set_names=("Alex S",)) + assert str(who) == "Alex S (opaque-1)" + + def test_only_deny_candidates_carry_self_set_names(self): + who = Identity( + id="opaque-1", names=("alex",), self_set_names=("Alex S",), + ) + assert who.candidates == ("opaque-1", "alex") + assert who.deny_candidates == ("opaque-1", "alex", "Alex S") + + +class TestPatternGate: + def test_an_empty_gate_lets_everyone_through(self): + assert PatternGate("subject").check(Identity(id="opaque-1")).allowed + + def test_allow_matches_the_opaque_id(self): + gate = PatternGate("subject", allow=["ID-0123ABC"]) + assert gate.check(Identity(id="ID-0123ABC")).allowed + assert not gate.check(Identity(id="ID-9999ZZZ")).allowed + + def test_allow_matches_a_resolved_name(self): + gate = PatternGate("subject", allow=["alex.soffronow"]) + assert gate.check( + Identity(id="opaque-1", names=("alex.soffronow",)), + ).allowed + + def test_matching_is_case_insensitive(self): + gate = PatternGate("resource", allow=["ENG-Platform"]) + assert gate.check(Identity(id="opaque-1", names=("eng-platform",))).allowed + + def test_globs_match_a_family_of_names(self): + gate = PatternGate("resource", allow=["eng-*"]) + assert gate.check(Identity(id="opaque-1", names=("eng-platform",))).allowed + assert not gate.check( + Identity(id="opaque-2", names=("sales-emea",)), + ).allowed + + def test_deny_beats_allow(self): + gate = PatternGate("resource", allow=["eng-*"], deny=["eng-secret"]) + assert not gate.check( + Identity(id="opaque-1", names=("eng-secret",)), + ).allowed + + def test_deny_alone_admits_everything_else(self): + gate = PatternGate("subject", deny=["*-bot"]) + assert gate.check(Identity(id="opaque-1", names=("alex",))).allowed + assert not gate.check( + Identity(id="opaque-2", names=("deploy-bot",)), + ).allowed + + def test_an_unidentified_subject_cannot_satisfy_an_allow_list(self): + gate = PatternGate("subject", allow=["alex"]) + assert not gate.check(Identity()).allowed + + def test_a_self_set_name_cannot_satisfy_an_allow_list(self): + # Otherwise the subject picks its own access: it renames itself to + # whatever the allow list happens to say. + gate = PatternGate("subject", allow=["alex.soffronow"]) + mallory = Identity( + id="opaque-2", + names=("mallory",), + self_set_names=("alex.soffronow", "Alex Soffronow"), + ) + verdict = gate.check(mallory) + assert not verdict.allowed + assert "sets itself" in verdict.reason + # The real holder of the handle is unaffected. + assert gate.check( + Identity(id="opaque-1", names=("alex.soffronow",)), + ).allowed + + def test_a_self_set_name_still_satisfies_a_deny_list(self): + # Refusing on more names than a grant may rest on is always safe. + gate = PatternGate("subject", deny=["*-bot"]) + assert not gate.check( + Identity( + id="opaque-1", + names=("integration-42",), + self_set_names=("deploy-bot",), + ), + ).allowed + + def test_an_incomplete_name_set_is_refused_when_a_deny_list_exists(self): + # The whole point: a deny list that cannot be evaluated must not + # quietly pass the subject it was written to stop. + gate = PatternGate("subject", deny=["*-bot"]) + assert not gate.check(Identity(id="opaque-1", complete=False)).allowed + + def test_an_incomplete_name_set_is_fine_when_only_allow_ids_are_used(self): + gate = PatternGate("subject", allow=["opaque-1"]) + assert gate.check(Identity(id="opaque-1", complete=False)).allowed + + def test_any_deny_pattern_ignores_the_allow_list(self): + gate = PatternGate( + "subject", allow=["a@b.c"], deny=["blocked@example.com"], + ) + assert gate.any_deny_pattern(lambda p: "@" in p) + # An allow pattern is not a deny pattern: allow already fails closed. + assert not PatternGate("subject", allow=["a@b.c"]).any_deny_pattern( + lambda p: "@" in p, + ) + + def test_the_reason_names_the_gate_and_the_pattern(self): + gate = PatternGate("resource", deny=["*-random"]) + verdict = gate.check(Identity(id="opaque-1", names=("eng-random",))) + assert "resource" in verdict.reason + assert "*-random" in verdict.reason + + +def _is_id(pattern: str) -> bool: + """Stand-in for a transport's literal-ID predicate.""" + return bool(re.fullmatch(r"ID-[A-Z0-9]+", pattern)) + + +class TestNeedsNameResolution: + def test_plain_ids_need_no_lookup(self): + assert not needs_name_resolution( + PatternGate("subject", allow=["ID-0123ABC", "ID-456DEF"]), + is_id=_is_id, + ) + + def test_a_glob_needs_a_lookup(self): + assert needs_name_resolution( + PatternGate("resource", allow=["ENG-*"]), is_id=_is_id, + ) + + def test_a_lowercase_handle_needs_a_lookup(self): + assert needs_name_resolution( + PatternGate("subject", allow=["alex.soffronow"]), is_id=_is_id, + ) + + def test_a_deny_pattern_counts_too(self): + assert needs_name_resolution( + PatternGate( + "subject", allow=["ID-0123ABC"], deny=["*-bot"], + ), + is_id=_is_id, + ) + + def test_an_empty_gate_needs_nothing(self): + assert not needs_name_resolution(PatternGate("subject"), is_id=_is_id) + + def test_an_uppercase_name_is_not_mistaken_for_an_id(self): + # A case heuristic read ALICE as an id, skipped the lookup, and let + # the deny list pass the person it named. + assert needs_name_resolution( + PatternGate("subject", deny=["ALICE"]), is_id=_is_id, + ) + assert needs_name_resolution( + PatternGate("resource", deny=["ENGINEERING"]), is_id=_is_id, + ) + + def test_a_glob_is_never_an_id_however_it_is_spelled(self): + assert needs_name_resolution( + PatternGate("subject", allow=["ID-0123AB*"]), is_id=lambda p: True, + ) + + def test_omitting_the_predicate_forces_a_lookup(self): + # The safe direction: a wrong "this is an id" guess skips a lookup + # the deny list depends on, so no guess means always look up. + assert needs_name_resolution( + PatternGate("subject", allow=["ID-0123ABC"]), + ) + + +# Deny rules must fail closed for every lookup outcome and pattern kind. + +# How the transport's name lookup turned out, as the Identity it produces +# for one subject whose handle is "alice" and whose email is "a@corp.com". +LOOKUPS = { + # Skipped: every pattern was a literal id, so the id alone suffices. + "skipped": Identity(id="ID-0123ABC", complete=True), + # Complete: every alias the patterns need came back. + "complete": Identity( + id="ID-0123ABC", names=("alice", "a@corp.com"), complete=True, + ), + # Partial: the provider returned an identity without a required alias. + "partial": Identity(id="ID-0123ABC", names=("alice",), complete=False), + # Failed: the lookup raised. + "failed": Identity(id="ID-0123ABC", complete=False), +} + +# A deny pattern of each kind, and whether the "complete" identity matches it. +DENY_PATTERNS = { + "id": ("ID-0123ABC", True), + "handle": ("alice", True), + "email": ("a@corp.com", True), + "glob": ("al*", True), + "uppercase-name": ("ALICE", True), + "unrelated": ("mallory", False), +} + + +@pytest.mark.parametrize("lookup_name", sorted(LOOKUPS)) +@pytest.mark.parametrize("pattern_name", sorted(DENY_PATTERNS)) +def test_a_deny_list_is_never_evaluated_against_an_incomplete_name_set( + lookup_name, pattern_name, +): + """A deny list must refuse unless it can prove the subject is not on it. + + Only a complete candidate set can prove that. Anything else — a lookup + that failed, or one that came back short — is refused, whether or not + the pattern happens to match what little was read. + """ + who = LOOKUPS[lookup_name] + pattern, matches_complete_identity = DENY_PATTERNS[pattern_name] + gate = PatternGate("subject", allow=["*"], deny=[pattern]) + + verdict = gate.check(who) + + if not who.complete: + assert not verdict.allowed, ( + f"{lookup_name} lookup + {pattern_name} deny pattern was admitted; " + "an incomplete candidate set cannot clear a deny list" + ) + return + + # A complete set is evaluated on its merits. The skipped-lookup identity + # carries only the id, which is complete precisely when the pattern is + # an id — the case that let the lookup be skipped in the first place. + expected_match = ( + matches_complete_identity if who.names else pattern_name == "id" + ) + assert verdict.allowed is not expected_match + + +@pytest.mark.parametrize("lookup_name", sorted(LOOKUPS)) +def test_an_allow_list_fails_closed_on_every_lookup_outcome(lookup_name): + """An allow list needs no completeness rule — an unread name matches + nothing, so a short candidate set refuses on its own.""" + who = LOOKUPS[lookup_name] + assert not PatternGate("subject", allow=["mallory"]).check(who).allowed + + +@pytest.mark.parametrize("lookup_name", sorted(LOOKUPS)) +def test_an_id_allow_list_admits_on_every_lookup_outcome(lookup_name): + """The id is the one candidate always present, so an id allow list works + even when the name lookup failed.""" + who = LOOKUPS[lookup_name] + assert PatternGate("subject", allow=["ID-0123ABC"]).check(who).allowed