delivery.py

Go to the documentation of this file.

"""Grant-checked, filtered, durable outbound delivery orchestration.
 
Dependencies
------------
- oc_connect.audit (AuditStore, content_hash - audits a blocked send and finds the audit id a keyboard names)
- oc_connect.connectors.telegram.destinations (DestinationStore - resolves the named destination and its connector)
- oc_connect.connectors.telegram.transport (inline_keyboard - builds the reply-vocabulary keyboard for --choices)
- oc_connect.credentials (CredentialBroker - handed to the connector factory)
- oc_connect.egress (require_safe, EgressBlockedError - the egress filter every send passes)
- oc_connect.grants (GrantStore - resolves the send risk and requires a matching grant)
- oc_connect.outbox (Outbox, OutboxItem - queues and processes the durable send)
- oc_connect.registry (ConnectorRegistry, load_builtin_connectors - finds the connector for a destination)
- oc_diagnostics (_dbg structured logger)
- option_c_connector_contracts (REPLY_VOCABULARY, DeliveryReceipt - allowed choices and the send receipt)
"""
from __future__ import annotations
 
from collections.abc import Callable, Mapping, Sequence
from typing import Any, Protocol
 
from oc_diagnostics import _dbg
from option_c_connector_contracts import REPLY_VOCABULARY, DeliveryReceipt
 
from .audit import AuditStore, content_hash
from .connectors.telegram.destinations import DestinationStore
from .connectors.telegram.transport import inline_keyboard
from .credentials import CredentialBroker
from .egress import EgressBlockedError, require_safe
from .grants import GrantStore
from .outbox import Outbox, OutboxItem
from .registry import ConnectorRegistry, load_builtin_connectors
 
#: The authority sentence for ``--choices`` (plan 236 P5). It is printed in
#: the ``send`` help and is the rule the orchestrator applies when deciding
#: which questions get buttons.
CHOICES_RULE = (
    "Security rulings (scanner waivers, thresholds, M8 files) are NEVER sent "
    "with --choices; buttons carry operational decisions only."
)
 
 
class DeliveryConnector(Protocol):
    """Runtime adapter that can deliver one outbox item.
 
    ``deliver`` returns the connector's own message identifier when it has
    one. That reference is what an inbound reply correlates against (§4);
    returning ``None`` simply means replies to this message cannot be
    correlated, never that the send failed. ``reply_markup`` (plan 236) is
    only ever passed when a send carries choices, so a connector that
    predates it still delivers every plain send.
    """
 
    def deliver(
        self,
        item: OutboxItem,
        *,
        reply_markup: Mapping[str, Any] | None = None,
    ) -> str | None: ...
 
 
def parse_choices(raw: Sequence[str]) -> tuple[str, ...]:
    """Return the vocabulary tokens a keyboard may carry, or refuse by name.
 
    Structured-only ruling (plan 236 P1): a label is accepted only when it is
    exactly a :data:`REPLY_VOCABULARY` token. Case is not folded, a URL is a
    label like any other and is refused the same way, and a repeated token is
    refused because a keyboard with two identical buttons answers nothing.
    """
    accepted: list[str] = []
    for label in raw:
        candidate = label.strip()
        if candidate not in REPLY_VOCABULARY:
            raise ValueError(
                f"choice {candidate!r} is not a reply-vocabulary token; buttons "
                f"may carry only {', '.join(REPLY_VOCABULARY)}"
            )
        if candidate in accepted:
            raise ValueError(f"choice {candidate!r} is listed more than once")
        accepted.append(candidate)
    if not accepted:
        raise ValueError("--choices needs at least one reply-vocabulary token")
    return tuple(accepted)
 
 
class OutboundSender:
    """Enforce the complete outbound authorization and delivery pipeline."""
 
    def __init__(
        self,
        *,
        broker: CredentialBroker,
        destinations: DestinationStore,
        grants: GrantStore,
        outbox: Outbox,
        audit: AuditStore,
        connectors: ConnectorRegistry | None = None,
    ) -> None:
        self.broker = broker
        self.destinations = destinations
        self.grants = grants
        self.outbox = outbox
        self.audit = audit
        self.connectors = connectors or load_builtin_connectors()
 
    def _connector(self, name: str) -> DeliveryConnector:
        factory = self.connectors.get(name)
        connector = (
            factory(self.broker, self.destinations)
            if callable(factory)
            else factory
        )
        if not hasattr(connector, "deliver"):
            raise TypeError(f"connector {name!r} does not implement delivery")
        return connector  # type: ignore[return-value]
 
    def _deliver_with(
        self,
        connector: DeliveryConnector,
        idempotency_key: str,
        tokens: tuple[str, ...] | None,
    ) -> Callable[[OutboxItem], str | None]:
        """Return the delivery callable, with the keyboard bound when asked.
 
        The keyboard's callback data names an audit record of THIS send: the
        record the outbox wrote when it queued the item under this key. The
        ``sent`` record does not exist until the transport has returned, so it
        cannot be inside the message it describes; the poller resolves either
        id through the key to the SENT record, which is the id a typed reply
        also reads (plan 236 P2).
        """
        if tokens is None:
            return connector.deliver
        question = self.audit.latest(idempotency_key)
        if question is None:
            raise RuntimeError(
                f"no audit record for {idempotency_key!r}; a keyboard needs the "
                "send's audit id"
            )
        markup = inline_keyboard(tokens, question.audit_id)
 
        def deliver(item: OutboxItem) -> str | None:
            return connector.deliver(item, reply_markup=markup)
 
        return deliver
 
    def send(
        self,
        *,
        grantee: str,
        destination: str,
        text: str,
        grant_id: str,
        risk_class: str | None = None,
        idempotency_key: str,
        choices: Sequence[str] | None = None,
    ) -> DeliveryReceipt:
        """Authorize, scan, queue, deliver, and audit one outbound message.
 
        A ``risk_class`` of ``None`` means "use the operator grant's recorded
        class" (CONNECT-02 §7.2). Resolution is downward only: the resolved
        value is still checked by :meth:`GrantStore.require_send_grant`, so no
        send is ever authorized by the default rather than by the grant.
 
        *choices* (plan 236) attaches a one-row keyboard of reply-vocabulary
        buttons; anything outside the vocabulary is refused by name before the
        grant, the filter, or the outbox is touched. The text itself still
        passes the egress filter unchanged.
        """
        tokens = parse_choices(choices) if choices is not None else None
        target = self.destinations.get(destination)
        if risk_class is None:
            risk_class = self.grants.resolve_send_risk(grant_id)
        self.grants.require_send_grant(
            grant_id=grant_id,
            grantee=grantee,
            connector=target.connector,
            destination=destination,
            risk_class=risk_class,
        )
        try:
            require_safe(text)
        except EgressBlockedError:
            self.audit.append(
                connector=target.connector,
                grant_id=grant_id,
                destination=destination,
                outcome="blocked",
                digest=content_hash(text),
                idempotency_key=idempotency_key,
                attempts=0,
                error_code="egress_finding",
                risk_class=risk_class,
            )
            _dbg(
                "OC_CONNECT.SEND",
                "send",
                f"connector={target.connector} grant={grant_id} "
                f"destination={destination} outcome=blocked",
                1,
                override=True,
            )
            raise
        item = self.outbox.enqueue(
            connector=target.connector,
            grant_id=grant_id,
            destination=destination,
            content=text,
            idempotency_key=idempotency_key,
            risk_class=risk_class,
        )
        receipt = self.outbox.process(
            idempotency_key,
            self._deliver_with(
                self._connector(target.connector), item.idempotency_key, tokens
            ),
        )
        _dbg(
            "OC_CONNECT.SEND",
            "send",
            f"connector={target.connector} grant={grant_id} "
            f"destination={destination} outcome={receipt.outcome} "
            f"choices={len(tokens) if tokens else 0}",
            2,
        )
        return receipt
 
 
__all__ = ["CHOICES_RULE", "DeliveryConnector", "OutboundSender", "parse_choices"]