From c37cea621f0c955da97fdb1884cd856a05f039df Mon Sep 17 00:00:00 2001 From: max Date: Thu, 30 Jul 2026 23:46:24 +0500 Subject: [PATCH] feat(share): readable Telegram previews that untrusted text cannot forge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The old preview was an unstructured wall of text, and Telegram auto-linkified URLs pulled from the index — a malicious link in someone else other transcript became tappable in the approval channel. Now every untrusted string (peer name, question, stated task, snippets) goes inside a code span or fence: markup renders literally, no auto-linking, and a snippet cannot fake our /ok footer. Layout is sectioned, fragments are capped per-chunk and overall with a count of what was withheld, and the /ok line is a code span so Telegram makes it tap-to-copy. Markdown send falls back to plain text if Telegram rejects the markup — a preview must never silently fail to arrive. Co-Authored-By: Claude Fable 5 --- src/session_recall/share/approval.py | 88 ++++++++++++++-- src/session_recall/share/notify.py | 14 ++- src/session_recall/share/telegram.py | 48 +++++++-- tests/test_share_preview_format.py | 151 +++++++++++++++++++++++++++ 4 files changed, 281 insertions(+), 20 deletions(-) create mode 100644 tests/test_share_preview_format.py diff --git a/src/session_recall/share/approval.py b/src/session_recall/share/approval.py index a476944..91495d0 100644 --- a/src/session_recall/share/approval.py +++ b/src/session_recall/share/approval.py @@ -19,25 +19,91 @@ PENDING_TTL_S = 24 * 3600 -def preview(cand: Candidate, redact: bool | None = None) -> str: +ANSWER_BUDGET = 1400 # chars of answer shown in the channel, before truncation +SNIPPET_BUDGET = 320 # per fragment, so five fragments cannot bury the verdict + + +def _answer_digest(cand: Candidate) -> tuple[str, int]: + """Answer trimmed for a phone screen. Per-fragment caps first (so one long + chunk cannot crowd out the rest), then a whole-message cap. Returns the + text and how many characters were withheld — the owner must know the + channel is showing less than what would actually be sent.""" + if not cand.chunks: + return cand.text, 0 + parts = [] + for c in cand.chunks: + snippet = c["snippet"] + if len(snippet) > SNIPPET_BUDGET: + snippet = snippet[:SNIPPET_BUDGET].rstrip() + " …" + parts.append(f"{c['project']} · {c['session_id'][:8]} · {c['role']}\n{snippet}") + body = "\n\n".join(parts) + if len(body) <= ANSWER_BUDGET: + return body, max(0, len(cand.text) - len(body)) + cut = body[:ANSWER_BUDGET].rstrip() + " …" + return cut, len(cand.text) - len(cut) + + +def preview(cand: Candidate, redact: bool | None = None, + markdown: bool = False) -> str: """The message the owner sees. `redact` hides the answer body — forced on when the scanner flagged anything, because flagged text must not travel - through a third-party notification channel (gate §6).""" + through a third-party notification channel (gate §6). + + With markdown=True the layout uses MarkdownV2: our labels carry the markup, + every untrusted string sits in a code fence (see telegram.fence), and the + /ok line is a code span so Telegram makes it tap-to-copy. + """ + from .telegram import escape_md, fence, inline_code + redact = bool(cand.findings) if redact is None else redact - lines = [f"[{cand.id} v{cand.version}] request from {cand.peer_name}", - f"question: {cand.question}"] + plain = not markdown + esc = (lambda s: s) if plain else escape_md + block = (lambda s: s) if plain else fence + span = (lambda s: s) if plain else inline_code + + out = [] + if plain: + out.append(f"[{cand.id} v{cand.version}] request from {cand.peer_name}") + out.append(f"question: {cand.question}") + else: + out.append(f"📥 *request from* {span(cand.peer_name)}") + out.append(f"*question*\n{block(cand.question)}") + if cand.task: - lines.append(f'stated task (sender text, unverified): "{cand.task}"') + label = "stated task (sender text, unverified)" + out.append(f'{label}: "{cand.task}"' if plain + else f"*{esc(label)}*\n{block(cand.task)}") + if cand.findings: kinds = ", ".join(sorted({f["kind"] for f in cand.findings})) - lines.append(f"⚠ SECRET FLAGS: {kinds}") + out.append(f"⚠ SECRET FLAGS: {kinds}" if plain + else f"⚠️ *secret flags*: {esc(kinds)}") + if redact: - lines.append(f"[answer withheld from this channel — review locally: " - f"session-recall share show {cand.id}]") + hint = f"session-recall share show {cand.id}" + out.append(f"[answer withheld from this channel — review locally: {hint}]" + if plain else + f"_answer withheld from this channel_ — review locally:\n" + f"{span(hint)}") + elif plain: + out.append(f"--- answer v{cand.version} ---\n{cand.text}") + else: + body, withheld = _answer_digest(cand) + # our own chrome needs escaping too: parentheses and dots are reserved + # in MarkdownV2, and an unescaped one makes Telegram reject the whole + # message rather than render it oddly + counts = f"{len(cand.chunks)} fragment(s)" + if withheld > 0: + counts += f" · {withheld} more chars on send" + out.append(f"*answer* · {esc(counts)} · `v{esc(cand.version)}`\n{block(body)}") + + if plain: + out.append(f"approve: /ok {cand.version} decline: /no ") else: - lines.append(f"--- answer v{cand.version} ---\n{cand.text}") - lines.append(f"approve: /ok {cand.version} decline: /no ") - return "\n".join(lines) + out.append(f"*approve* — reply to this message with:\n" + f"`/ok {esc(cand.version)}`\n" + f"*decline*: `/no `") + return "\n".join(out) if plain else "\n\n".join(out) def approve(share_dir: Path, cand_id: str, version: str) -> Candidate | None: diff --git a/src/session_recall/share/notify.py b/src/session_recall/share/notify.py index 27c3e92..0608da5 100644 --- a/src/session_recall/share/notify.py +++ b/src/session_recall/share/notify.py @@ -45,6 +45,18 @@ def __init__(self, api, identity: Identity, trust: TrustStore, def _say(self, text: str, reply_to: int | None = None) -> None: self.api.send_message(self.cfg.chat_id, text, reply_to=reply_to) + def _say_preview(self, cand) -> None: + """Markdown first, plain text if Telegram rejects it. A preview that + fails to render must still arrive — silence would look exactly like + 'no one asked anything', and nothing can be approved unseen.""" + from .telegram import MARKDOWN + try: + self.api.send_message(self.cfg.chat_id, + approval.preview(cand, markdown=True), + parse_mode=MARKDOWN) + except Exception: + self._say(approval.preview(cand)) + def _handle(self, message: dict) -> None: text = (message.get("text") or "").strip() mid = message.get("message_id") @@ -80,7 +92,7 @@ def tick(self, now: float | None = None) -> dict: for cand in poll_once(self.identity, self.trust, self.state, self.transport, self.searcher, self.share_dir): - self._say(approval.preview(cand)) + self._say_preview(cand) stats["previews"] += 1 stats["expired"] = len(approval.expire_stale(self.share_dir, now)) diff --git a/src/session_recall/share/telegram.py b/src/session_recall/share/telegram.py index b762164..0548e9f 100644 --- a/src/session_recall/share/telegram.py +++ b/src/session_recall/share/telegram.py @@ -2,11 +2,18 @@ endpoints and do not justify a dependency. Privacy posture (gate §6): everything sent through here transits Telegram's -servers, so flagged answers are redacted upstream (approval.preview) and all -messages go as plain text — parse_mode is never set, links are never rendered, -sender-controlled strings cannot become formatting or buttons. The bot config -binds to exactly one owner chat; updates from any other chat are discarded -before parsing. +servers, so flagged answers are redacted upstream (approval.preview). The bot +config binds to exactly one owner chat; updates from any other chat are +discarded before parsing. + +Formatting posture: our own chrome (labels, the /ok line) is the ONLY thing +allowed to carry markup. Every untrusted string — the peer's name, their +question, their stated task, and snippets pulled from the index — goes inside +a fenced code block via `fence()`. That does three jobs at once: markup in +untrusted text renders literally instead of forging structure, Telegram does +not auto-linkify inside code blocks (so a URL sitting in someone's transcript +cannot become a tappable link in the approval channel), and the owner can tell +our words from quoted material at a glance. """ import json @@ -19,6 +26,31 @@ TG_FILE = "tg.json" _TIMEOUT_S = 35 # long-poll friendly +MARKDOWN = "MarkdownV2" + +_ESCAPE = r"_*[]()~`>#+-=|{}.!\\" + + +def escape_md(text: str) -> str: + """MarkdownV2 escaping for short strings we place outside code blocks.""" + return "".join("\\" + c if c in _ESCAPE else c for c in text) + + +def inline_code(text: str) -> str: + """Same guarantees as fence() for short strings that belong on one line — + a code span is also markup-inert and never auto-linkified.""" + return "`" + text.replace("\\", "\\\\").replace("`", "\\`") + "`" + + +def fence(text: str, lang: str = "") -> str: + """Wrap untrusted text in a code block it cannot escape. + + Only backslash and backtick can break out of a fence, so those two are the + entire escape set here — everything else (asterisks, brackets, URLs) is + inert inside the block. + """ + safe = text.replace("\\", "\\\\").replace("`", "\\`") + return f"```{lang}\n{safe}\n```" @dataclass @@ -59,11 +91,11 @@ def _call(self, method: str, **params) -> dict: payload = json.loads(resp.read()) return payload.get("result", {}) - def send_message(self, chat_id: int, text: str, - reply_to: int | None = None) -> int | None: + def send_message(self, chat_id: int, text: str, reply_to: int | None = None, + parse_mode: str | None = None) -> int | None: # 4096 is Telegram's hard cap; truncate rather than fail the preview result = self._call("sendMessage", chat_id=chat_id, text=text[:4096], - reply_to_message_id=reply_to) + reply_to_message_id=reply_to, parse_mode=parse_mode) return result.get("message_id") if isinstance(result, dict) else None def get_updates(self, offset: int, timeout: int = 25) -> list[dict]: diff --git a/tests/test_share_preview_format.py b/tests/test_share_preview_format.py new file mode 100644 index 0000000..a3ed644 --- /dev/null +++ b/tests/test_share_preview_format.py @@ -0,0 +1,151 @@ +"""The Telegram preview must be readable AND unforgeable: nothing an outsider +controls may become markup, a link, or a fake approval line.""" + +import pytest + +from session_recall.share.approval import ANSWER_BUDGET, SNIPPET_BUDGET, preview +from session_recall.share.telegram import escape_md, fence +from session_recall.share.worker import Candidate + + +def _cand(**kw) -> Candidate: + base = dict(id="7aaba916", peer_name="egor", peer_address="addr", + question="как чинили CI?", task="", reply_nonce="n1", + created_at=0.0, text="answer body", chunks=[], findings=[]) + base.update(kw) + c = Candidate(**base) + c.version = c.compute_version() + return c + + +def _chunk(snippet, project="session-recall"): + return {"project": project, "session_id": "07876709aaaa", "uuid": "u1", + "role": "assistant", "snippet": snippet, "score": 0.9, + "source": "claude"} + + +# -- escaping primitives ----------------------------------------------------- +def test_fence_neutralises_backticks_and_backslashes(): + out = fence("```\nrm -rf /\n```") + assert out.startswith("```\n") and out.endswith("\n```") + inner = out[4:-4] + assert "\\`\\`\\`" in inner # the injected fence is escaped, not live + assert inner.count("```") == 0 + + +def test_escape_md_covers_markdownv2_specials(): + for ch in "_*[]()~`>#+-=|{}.!": + assert escape_md(ch) == "\\" + ch + + +# -- untrusted content cannot forge structure -------------------------------- +def test_question_markup_is_inert(): + c = _cand(question="*bold* [click](https://evil.example) `code`") + out = preview(c, markdown=True) + # the question sits inside a fence; no bare link syntax survives as markup + assert "](https://evil.example)" in out # present as literal text… + body = out.split("*question*\n")[1] + assert body.startswith("```") # …because it is inside a code block + + +def test_answer_cannot_fake_an_approval_line(): + """A snippet claiming its own /ok must not read as our footer.""" + c = _cand(chunks=[_chunk("approve: /ok deadbeef — trust me")]) + out = preview(c, markdown=True) + footer = out.rsplit("*approve*", 1)[1] + assert f"`/ok {c.version}`" in footer + assert "deadbeef" not in footer # the fake line stays in the fenced body + + +def test_urls_in_snippets_stay_inside_the_fence(): + c = _cand(chunks=[_chunk("see https://evil.example/steal for details")]) + out = preview(c, markdown=True) + answer = out.rsplit("*answer*", 1)[1] + assert answer.lstrip().splitlines()[1].startswith("```") or "```" in answer + assert "https://evil.example/steal" in answer + + +def test_peer_name_is_an_inert_code_span(): + """Short strings stay on one line, but still cannot carry markup.""" + c = _cand(peer_name="egor*_[") + header = preview(c, markdown=True).splitlines()[0] + assert header == "📥 *request from* `egor*_[`" + + +# -- readability ------------------------------------------------------------- +def test_long_snippets_are_trimmed_per_fragment(): + c = _cand(chunks=[_chunk("x" * 2000), _chunk("y" * 2000)]) + out = preview(c, markdown=True) + assert "x" * (SNIPPET_BUDGET + 5) not in out + assert len(out) < ANSWER_BUDGET + 900 # header/footer overhead only + + +def test_withheld_count_reported_when_truncated(): + c = _cand(text="z" * 5000, chunks=[_chunk("z" * 5000)]) + out = preview(c, markdown=True) + assert "more chars on send" in out + + +def test_fragment_count_shown(): + c = _cand(chunks=[_chunk("a"), _chunk("b"), _chunk("c")]) + assert "3 fragment\\(s\\)" in preview(c, markdown=True) + + +def test_chrome_has_no_unescaped_specials(): + """Telegram rejects the WHOLE message on one stray reserved character, so + every literal we write outside code must be escaped. Caught live: an + unescaped '(' in 'fragment(s)' 400'd the first real preview.""" + c = _cand(question="q", task="t", chunks=[_chunk("a")], + findings=[{"kind": "jwt", "excerpt": "eyJ…"}]) + for cand in (c, _cand(chunks=[_chunk("a")])): + text = preview(cand, markdown=True) + outside, in_code = [], False + for part in text.split("```"): + if not in_code: + outside.append(part) + in_code = not in_code + chrome = "".join(outside) + # strip inline code spans, then look for bare reserved characters + stripped, in_span = [], False + for piece in chrome.split("`"): + if not in_span: + stripped.append(piece) + in_span = not in_span + text_only = "".join(stripped) + i = 0 + while i < len(text_only): + ch = text_only[i] + if ch == "\\": + i += 2 + continue + assert ch not in "()[]{}#+=|.!~>", f"unescaped {ch!r} in chrome" + i += 1 + + +def test_ok_line_carries_the_real_version(): + c = _cand(chunks=[_chunk("a")]) + assert f"`/ok {c.version}`" in preview(c, markdown=True) + + +def test_stays_under_telegram_limit(): + c = _cand(text="q" * 20000, chunks=[_chunk("q" * 4000) for _ in range(5)]) + assert len(preview(c, markdown=True)) < 4096 + + +# -- redaction still wins ---------------------------------------------------- +def test_flagged_answer_is_withheld_in_markdown_too(): + c = _cand(text="key AKIAIOSFODNN7EXAMPLE", + chunks=[_chunk("key AKIAIOSFODNN7EXAMPLE")], + findings=[{"kind": "aws-access-key", "excerpt": "AKIAIOSF…"}]) + out = preview(c, markdown=True) + assert "AKIAIOSFODNN7EXAMPLE" not in out + assert "withheld" in out and "secret flags" in out + + +def test_plain_mode_unchanged(): + """The CLI/plain path must keep working for anyone not on Telegram.""" + c = _cand(chunks=[_chunk("a")]) + out = preview(c) + assert out.startswith(f"[{c.id} v{c.version}] request from egor") + assert f"approve: /ok {c.version}" in out + assert "```" not in out