poller.py

Go to the documentation of this file.

"""Inbound transport gates: pinned sender, flood bound, hash-only drop log.
 
Proposal V1 §4 and §10.4. The poller is the only thing that turns a Telegram
update into a stored message, and it refuses far more than it accepts:
 
* **Pinned sender only.** The account id *and* the chat id must both match the
  destination pinned at setup, and the sender must not be a bot. Everything
  else is dropped, hash-logged, and never written to the inbox. Dropped
  updates still advance the cursor: re-reading a hostile sender's message on
  every poll would be a self-inflicted flood.
* **Text only.** A v1 message without text is dropped the same way. Widening
  to attachments is new attack surface and needs its own ruling.
* **Flood bound.** At most :data:`FLOOD_LIMIT` accepted messages per
  :data:`FLOOD_WINDOW_SECONDS`. Updates are processed oldest-first, and once
  the window is full the poller *stops* and leaves the cursor where it is, so
  the overflow is deferred rather than discarded and arrives in order on a
  later poll. The overflow itself is logged.
* **Taps are bound twice (plan 236).** A ``callback_query`` update is
  admitted only when its sender passes the same pinned check as a message
  AND its callback data names a SENT question whose transport message is the
  one the button sits under. Each failed binding is refused by name
  (:data:`CALLBACK_REFUSALS`), hash-logged, never stored, and never
  acknowledged. An admitted tap is stored as a structured reply, counts
  against the flood window like a message, and is acknowledged afterwards;
  a failed acknowledgement is logged and never loses the tap.
 
The transport is always injected. There is no code path in this module that
constructs an HTTP client, which is what keeps the suite free of live calls.
 
Dependencies
------------
- oc_connect.audit (AuditStore, AuditRecord, content_hash, timestamp, utc_now - hash-only received, dropped, overflow and tap-acknowledgement records)
- oc_connect.comms (CommsBus, new_ulid - the poll cursor path and inbound message ids)
- oc_connect.connectors.telegram.destinations (TelegramDestination - the pinned account and chat a sender must match)
- oc_connect.inbound.classify (classify - mechanical classification of each admitted message)
- oc_connect.inbound.inbox (InboxStore - stores admitted messages and counts the flood window)
- oc_connect.outbox (Outbox, OutboxItem - correlates replies and taps to the sent question)
- oc_diagnostics (_dbg structured logger)
- option_c_connector_contracts (InboundMessage, InboundCorrelation, REPLY_VOCABULARY, lane - message model, correlation and lane risk)
"""
from __future__ import annotations
 
import json
import os
import secrets as _secrets
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Protocol
 
from oc_diagnostics import _dbg
from option_c_connector_contracts import (
    REPLY_VOCABULARY,
    InboundCorrelation,
    InboundMessage,
    lane,
)
 
from ..audit import AuditRecord, AuditStore, content_hash, timestamp, utc_now
from ..comms import CommsBus, new_ulid
from ..connectors.telegram.destinations import TelegramDestination
from ..outbox import Outbox, OutboxItem
from .classify import classify
from .inbox import InboxStore
 
FLOOD_LIMIT = 30
FLOOD_WINDOW_SECONDS = 60.0
 
INBOUND_LANE = "inbound-telegram"
OUTBOUND_LANE = "outbound-telegram"
 
#: Callback data is ``<TOKEN>:<audit_id>``; mirrors the transport's builder.
CALLBACK_SEPARATOR = ":"
 
#: Every named reason a tap can be refused for (plan 236 P2).
CALLBACK_REFUSALS: tuple[str, ...] = (
    "unpinned_sender",
    "malformed_callback",
    "uncorrelated_callback",
    "callback_message_mismatch",
)
 
#: Audit key prefix of the outbound ``answerCallbackQuery`` row for one stored
#: tap: ``callback_ack-<inbound message id>``. The row's digest is the hash of
#: the callback id; it carries no content.
CALLBACK_ACK_KEY_PREFIX = "callback_ack-"
 
 
class InboundTransport(Protocol):
    """The Bot API methods the inbound path needs."""
 
    def get_updates(
        self,
        *,
        offset: int | None = None,
        timeout: int = 0,
    ) -> list[dict[str, Any]]: ...
 
    def answer_callback_query(self, callback_query_id: str) -> bool: ...
 
    def close(self) -> None: ...
 
 
@dataclass(frozen=True)
class PollResult:
    """What one poll did, in terms an operator-facing CLI can print."""
 
    stored: tuple[InboundMessage, ...]
    dropped: int
    deferred: int
 
    @property
    def accepted(self) -> int:
        """Return how many messages this poll admitted to the inbox."""
        return len(self.stored)
 
 
def parse_callback_data(data: Any) -> tuple[str, str] | None:
    """Return ``(token, audit_id)`` for well-formed callback data, else None.
 
    Exact match only: the token must be a :data:`REPLY_VOCABULARY` member as
    written, and the audit id must be a single non-blank run with no
    whitespace and no second separator. Nothing is normalised.
    """
    if not isinstance(data, str):
        return None
    token, separator, audit_id = data.partition(CALLBACK_SEPARATOR)
    if not separator or token not in REPLY_VOCABULARY:
        return None
    if not audit_id or audit_id != audit_id.strip() or any(
        character.isspace() for character in audit_id
    ):
        return None
    if CALLBACK_SEPARATOR in audit_id:
        return None
    return token, audit_id
 
 
@dataclass(frozen=True)
class _Question:
    """The outbound send a tap resolved to: its outbox item and SENT record."""
 
    item: OutboxItem
    sent: AuditRecord
 
 
class InboundPoller:
    """Fetch, authenticate, bound, classify, and store inbound updates."""
 
    def __init__(
        self,
        *,
        bus: CommsBus,
        destination: TelegramDestination,
        transport: InboundTransport,
        inbox: InboxStore,
        audit: AuditStore,
        outbox: Outbox | None = None,
        now: Callable[[], datetime] = utc_now,
        flood_limit: int = FLOOD_LIMIT,
        flood_window_seconds: float = FLOOD_WINDOW_SECONDS,
        long_poll_seconds: int = 0,
        entropy: Callable[[], bytes] | None = None,
    ) -> None:
        if flood_limit < 1 or flood_window_seconds <= 0:
            raise ValueError("flood bound values must be positive")
        self.bus = bus
        self.destination = destination
        self.transport = transport
        self.inbox = inbox
        self.audit = audit
        self.outbox = outbox
        self._now = now
        self.flood_limit = flood_limit
        self.flood_window_seconds = flood_window_seconds
        self.long_poll_seconds = long_poll_seconds
        self._entropy = entropy
 
    def _read_cursor(self) -> int | None:
        path = self.bus.cursor_path
        if not path.is_file():
            return None
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError):
            _dbg(
                "OC_CONNECT.INBOUND",
                "_read_cursor",
                f"unreadable cursor at {path}; restarting from the transport default",
                1,
                override=True,
            )
            return None
        offset = payload.get("offset") if isinstance(payload, dict) else None
        return offset if isinstance(offset, int) else None
 
    def _write_cursor(self, offset: int | None) -> None:
        if offset is None:
            return
        path = self.bus.cursor_path
        path.parent.mkdir(parents=True, exist_ok=True)
        temporary = path.with_name(f".tmp-{path.name}-{_secrets.token_hex(8)}")
        try:
            with temporary.open("w", encoding="utf-8", newline="\n") as handle:
                json.dump({"version": 1, "offset": offset}, handle, sort_keys=True)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, path)
        finally:
            if temporary.exists():
                try:
                    temporary.unlink()
                except OSError:
                    pass
 
    def _is_pinned(self, message: dict[str, Any]) -> bool:
        account = message.get("from")
        chat = message.get("chat")
        if not isinstance(account, dict) or not isinstance(chat, dict):
            return False
        if account.get("is_bot") is True:
            return False
        return (
            str(account.get("id")) == self.destination.operator_account_id
            and str(chat.get("id")) == self.destination.chat_id
        )
 
    def _refuse(
        self,
        *,
        digest: str,
        chat_id: str,
        error_code: str,
        update_id: int,
    ) -> None:
        self.audit.append(
            connector="telegram",
            grant_id=None,
            destination=chat_id,
            outcome="dropped",
            digest=digest,
            idempotency_key=f"update-{update_id}",
            attempts=0,
            error_code=error_code,
            lane=INBOUND_LANE,
            direction="inbound",
            risk_class=lane(INBOUND_LANE).risk_ceiling,
        )
        _dbg(
            "OC_CONNECT.INBOUND",
            "_refuse",
            f"SECURITY_ERROR: dropped inbound update={update_id} reason={error_code} "
            f"digest={digest}",
            1,
            override=True,
        )
 
    def _log_refusal(
        self,
        message: dict[str, Any],
        *,
        error_code: str,
        update_id: int,
    ) -> None:
        text = message.get("text")
        chat = message.get("chat")
        self._refuse(
            digest=content_hash(text if isinstance(text, str) else ""),
            chat_id=str(chat.get("id")) if isinstance(chat, dict) else "unknown",
            error_code=error_code,
            update_id=update_id,
        )
 
    def _correlate(self, message: dict[str, Any]) -> InboundCorrelation:
        reply_to = message.get("reply_to_message")
        if not isinstance(reply_to, dict):
            return InboundCorrelation(reply_to_transport_ref=None, audit_id=None)
        reference = reply_to.get("message_id")
        if reference is None:
            return InboundCorrelation(reply_to_transport_ref=None, audit_id=None)
        transport_ref = str(reference)
        if self.outbox is None:
            return InboundCorrelation(
                reply_to_transport_ref=transport_ref, audit_id=None
            )
        item = self.outbox.find_by_transport_ref(transport_ref)
        if item is None:
            return InboundCorrelation(
                reply_to_transport_ref=transport_ref, audit_id=None
            )
        record = self.audit.latest(item.idempotency_key, outcome="sent")
        return InboundCorrelation(
            reply_to_transport_ref=transport_ref,
            audit_id=record.audit_id if record is not None else None,
        )
 
    def _build(self, message: dict[str, Any], received: datetime) -> InboundMessage:
        text = message["text"]
        correlation = self._correlate(message)
        classification, token = classify(
            text, correlated_audit_id=correlation.audit_id
        )
        entropy = self._entropy() if self._entropy is not None else None
        account = message["from"]
        chat = message["chat"]
        return InboundMessage(
            message_id=new_ulid(received, entropy=entropy),
            lane=INBOUND_LANE,
            direction="inbound",
            risk_class=lane(INBOUND_LANE).risk_ceiling,
            connector="telegram",
            transport_ref=str(message.get("message_id", "")) or "unknown",
            source_account_id=str(account.get("id")),
            chat_id=str(chat.get("id")),
            received_at_utc=timestamp(received),
            text=text,
            content_hash=content_hash(text),
            classification=classification,
            vocabulary_token=token,
            correlation=correlation,
        )
 
    # -- plan 236: callback_query admission ---------------------------------
 
    def _resolve_question(self, audit_id: str) -> _Question | None:
        """Return the SENT outbound send an audit id belongs to, or None.
 
        The id may name any audit record of the send (the keyboard carries the
        record that existed when the message left; a typed reply reads the
        ``sent`` one). Resolution is: audit id -> outbound record -> its
        idempotency key -> a SENT record for that key -> the outbox item in
        state ``sent``. Any missing link is "uncorrelated".
        """
        if self.outbox is None:
            return None
        named = next(
            (record for record in self.audit.read_all() if record.audit_id == audit_id),
            None,
        )
        if named is None or named.direction != "outbound":
            return None
        sent = self.audit.latest(named.idempotency_key, outcome="sent")
        if sent is None:
            return None
        try:
            item = self.outbox.get(named.idempotency_key)
        except KeyError:
            return None
        if item.state != "sent" or item.transport_ref is None:
            return None
        return _Question(item=item, sent=sent)
 
    def _admit_callback(
        self,
        callback: dict[str, Any],
        *,
        update_id: int,
    ) -> tuple[str, _Question] | None:
        """Apply the double binding; refuse by name or return (token, question)."""
        message = callback.get("message")
        chat = message.get("chat") if isinstance(message, dict) else None
        chat_id = str(chat.get("id")) if isinstance(chat, dict) else "unknown"
        data = callback.get("data")
        digest = content_hash(data if isinstance(data, str) else "")
 
        def refuse(reason: str) -> None:
            self._refuse(
                digest=digest, chat_id=chat_id, error_code=reason, update_id=update_id
            )
 
        if not self._is_pinned({"from": callback.get("from"), "chat": chat}):
            refuse("unpinned_sender")
            return None
        parsed = parse_callback_data(data)
        if parsed is None:
            refuse("malformed_callback")
            return None
        token, audit_id = parsed
        question = self._resolve_question(audit_id)
        if question is None:
            refuse("uncorrelated_callback")
            return None
        tapped = message.get("message_id") if isinstance(message, dict) else None
        if tapped is None or str(tapped) != question.item.transport_ref:
            refuse("callback_message_mismatch")
            return None
        return token, question
 
    def _build_callback(
        self,
        callback: dict[str, Any],
        received: datetime,
        *,
        token: str,
        question: _Question,
    ) -> InboundMessage:
        classification, vocabulary_token = classify(
            token, correlated_audit_id=question.sent.audit_id
        )
        entropy = self._entropy() if self._entropy is not None else None
        account = callback["from"]
        chat = callback["message"]["chat"]
        callback_id = str(callback.get("id", "")) or "unknown"
        return InboundMessage(
            message_id=new_ulid(received, entropy=entropy),
            lane=INBOUND_LANE,
            direction="inbound",
            risk_class=lane(INBOUND_LANE).risk_ceiling,
            connector="telegram",
            transport_ref=callback_id,
            source_account_id=str(account.get("id")),
            chat_id=str(chat.get("id")),
            received_at_utc=timestamp(received),
            text=token,
            content_hash=content_hash(token),
            classification=classification,
            vocabulary_token=vocabulary_token,
            correlation=InboundCorrelation(
                reply_to_transport_ref=question.item.transport_ref,
                audit_id=question.sent.audit_id,
                source="callback",
            ),
            callback_query_id=callback_id,
        )
 
    def _acknowledge(self, stored: InboundMessage) -> None:
        """Answer one stored tap; a failure is logged and never unstores it."""
        func = "_acknowledge"
        callback_id = stored.callback_query_id or ""
        error_code: str | None = None
        try:
            self.transport.answer_callback_query(callback_id)
        except Exception as exc:  # noqa: BLE001 - any transport failure is logged
            error_code = type(exc).__name__
            _dbg(
                "OC_CONNECT.INBOUND",
                func,
                # OCT-LINT: disable=no_hardcoded_secrets reason="Diagnostic prose naming the Bot API method answerCallbackQuery and the S6 class EXTERNAL_ERROR; scores 4.69 bits/char and carries no credential. Operator-ruled 2026-09-05 (plan 236 escalation, line-scoped)"
                f"EXTERNAL_ERROR: answerCallbackQuery failed for message="
                f"{stored.message_id} error={error_code}; the tap is kept",
                1,
                override=True,
            )
        self.audit.append(
            connector="telegram",
            grant_id=None,
            destination=stored.chat_id,
            outcome="failed" if error_code else "sent",
            digest=content_hash(callback_id),
            idempotency_key=f"{CALLBACK_ACK_KEY_PREFIX}{stored.message_id}",
            attempts=1,
            error_code=error_code,
            lane=OUTBOUND_LANE,
            direction="outbound",
            risk_class=lane(OUTBOUND_LANE).risk_ceiling,
        )
 
    def _store(self, built: InboundMessage) -> None:
        self.inbox.store(built)
        self.audit.append(
            connector="telegram",
            grant_id=None,
            destination=built.chat_id,
            outcome="received",
            digest=built.content_hash,
            idempotency_key=built.message_id,
            attempts=0,
            lane=built.lane,
            direction=built.direction,
            risk_class=built.risk_class,
        )
 
    def poll(self) -> PollResult:
        """Run one fetch-and-admit cycle and return what it did."""
        func = "poll"
        offset = self._read_cursor()
        updates = self.transport.get_updates(
            offset=offset, timeout=self.long_poll_seconds
        )
        ordered = sorted(
            (
                update
                for update in updates
                if isinstance(update, dict) and isinstance(update.get("update_id"), int)
            ),
            key=lambda update: update["update_id"],
        )
        received = self._now().astimezone(timezone.utc)
        cutoff = received - timedelta(seconds=self.flood_window_seconds)
        admitted_in_window = self.inbox.count_since(cutoff)
        stored: list[InboundMessage] = []
        dropped = 0
        deferred = 0
        next_offset = offset
        for index, update in enumerate(ordered):
            update_id = update["update_id"]
            message = update.get("message")
            callback = update.get("callback_query")
            built: InboundMessage | None = None
            if isinstance(callback, dict):
                admitted = self._admit_callback(callback, update_id=update_id)
                if admitted is None:
                    dropped += 1
                    next_offset = update_id + 1
                    continue
                if admitted_in_window >= self.flood_limit:
                    deferred = len(ordered) - index
                    self._log_overflow(deferred, update_id=update_id)
                    break
                token, question = admitted
                built = self._build_callback(
                    callback, received, token=token, question=question
                )
            elif isinstance(message, dict):
                if not self._is_pinned(message):
                    self._log_refusal(
                        message, error_code="unpinned_sender", update_id=update_id
                    )
                    dropped += 1
                    next_offset = update_id + 1
                    continue
                if not isinstance(message.get("text"), str):
                    self._log_refusal(
                        message,
                        error_code="unsupported_message_kind",
                        update_id=update_id,
                    )
                    dropped += 1
                    next_offset = update_id + 1
                    continue
                if admitted_in_window >= self.flood_limit:
                    deferred = len(ordered) - index
                    self._log_overflow(deferred, update_id=update_id)
                    break
                built = self._build(message, received)
            else:
                next_offset = update_id + 1
                continue
            self._store(built)
            stored.append(built)
            admitted_in_window += 1
            next_offset = update_id + 1
            if built.callback_query_id is not None:
                self._acknowledge(built)
        self._write_cursor(next_offset)
        _dbg(
            "OC_CONNECT.INBOUND",
            func,
            f"poll admitted={len(stored)} dropped={dropped} deferred={deferred}",
            2,
        )
        return PollResult(stored=tuple(stored), dropped=dropped, deferred=deferred)
 
    def _log_overflow(self, deferred: int, *, update_id: int) -> None:
        self.audit.append(
            connector="telegram",
            grant_id=None,
            destination=self.destination.chat_id,
            outcome="overflow",
            digest=content_hash(""),
            idempotency_key=f"update-{update_id}",
            attempts=deferred,
            error_code="inbound_flood_cap",
            lane=INBOUND_LANE,
            direction="inbound",
            risk_class=lane(INBOUND_LANE).risk_ceiling,
        )
        _dbg(
            "OC_CONNECT.INBOUND",
            "_log_overflow",
            f"inbound flood cap reached; deferred={deferred} from update={update_id}",
            1,
            override=True,
        )
 
 
__all__ = [
    "CALLBACK_ACK_KEY_PREFIX",
    "CALLBACK_REFUSALS",
    "CALLBACK_SEPARATOR",
    "FLOOD_LIMIT",
    "FLOOD_WINDOW_SECONDS",
    "INBOUND_LANE",
    "InboundPoller",
    "InboundTransport",
    "PollResult",
    "parse_callback_data",
]