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/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/nerve/channels/archives.py b/nerve/channels/archives.py new file mode 100644 index 00000000..a6572079 --- /dev/null +++ b/nerve/channels/archives.py @@ -0,0 +1,187 @@ +"""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. +# 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): + """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 ( + 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 + # 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) + 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/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 a0c970db..43c575a2 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, @@ -1207,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}`.", @@ -1236,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}`.", @@ -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.""" @@ -1953,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" @@ -1983,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 @@ -2021,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_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 diff --git a/tests/test_channel_archives.py b/tests/test_channel_archives.py new file mode 100644 index 00000000..a384b1e5 --- /dev/null +++ b/tests/test_channel_archives.py @@ -0,0 +1,178 @@ +"""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 os +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) + 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) + _, 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]") + 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 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