"""Validated records for operator-attention acknowledgements.""" from __future__ import annotations import hashlib import hmac import json from collections.abc import Mapping from dataclasses import asdict, dataclass from gateway.platforms.telegram_nutrition_onboarding_publication_outbox import ( GatewayPublicationReceipt, ) @dataclass(frozen=True, slots=True) class GatewayOperatorAttentionAcknowledgement: session_id: str customer_key: str attention_identity: str candidate_epoch: str actor_user_id: int authority: tuple[int, int, int] route: tuple[str, str] message_id: int update_id: int callback_data: str publication_generation: int publication_dispatch_identity: str publication_receipt_integrity: str event_integrity: str def canonical(value: object) -> bytes: return json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") def build_acknowledgement( *, key: bytes, session_id: str, customer_key: str, attention_identity: str, candidate_epoch: str, actor_user_id: int, authority: tuple[int, int, int], route: tuple[str, str], message_id: int, update_id: int, callback_data: str, publication: GatewayPublicationReceipt, ) -> GatewayOperatorAttentionAcknowledgement: _text(session_id, "session ID") _text(customer_key, "customer key") _digest(attention_identity, "attention identity") _digest(candidate_epoch, "candidate epoch") _text(callback_data, "callback data") if ( type(actor_user_id) is not int or type(message_id) is not int or message_id <= 0 or type(update_id) is not int or update_id < 0 or len(authority) != 3 or any(type(item) is not int for item in authority) or len(route) != 2 or any(not isinstance(item, str) for item in route) or actor_user_id != authority[0] or route != (str(authority[1]), str(authority[2])) or publication.session_id != session_id or publication.role != "owner" or publication.state != "COMMITTED" or publication.route != route or publication.message_id != message_id or publication.payload.get("operator_attention_identity") != attention_identity or publication.payload.get("operator_delivery_epoch") != candidate_epoch or not isinstance(publication.receipt_integrity, str) ): raise ValueError("operator attention acknowledgement authority is invalid") values: dict[str, object] = { "session_id": session_id, "customer_key": customer_key, "attention_identity": attention_identity, "candidate_epoch": candidate_epoch, "actor_user_id": actor_user_id, "authority": list(authority), "route": list(route), "message_id": message_id, "update_id": update_id, "callback_data": callback_data, "publication_generation": publication.generation, "publication_dispatch_identity": publication.dispatch_identity, "publication_receipt_integrity": publication.receipt_integrity, } integrity = hmac.new(key, canonical(values), hashlib.sha256).hexdigest() return GatewayOperatorAttentionAcknowledgement( session_id=session_id, customer_key=customer_key, attention_identity=attention_identity, candidate_epoch=candidate_epoch, actor_user_id=actor_user_id, authority=authority, route=route, message_id=message_id, update_id=update_id, callback_data=callback_data, publication_generation=publication.generation, publication_dispatch_identity=publication.dispatch_identity, publication_receipt_integrity=publication.receipt_integrity, event_integrity=integrity, ) def parse_acknowledgement( value: object, *, key: bytes, ) -> GatewayOperatorAttentionAcknowledgement: if not isinstance(value, Mapping): raise ValueError("operator attention acknowledgement is invalid") raw: dict[str, object] = {} for name, item in value.items(): if not isinstance(name, str): raise ValueError("operator attention acknowledgement is invalid") raw[name] = item authority = raw.get("authority") route = raw.get("route") actor_user_id = raw.get("actor_user_id") message_id = raw.get("message_id") update_id = raw.get("update_id") publication_generation = raw.get("publication_generation") if not isinstance(authority, list) or len(authority) != 3: raise ValueError("operator attention acknowledgement is invalid") typed_authority: list[int] = [] for item in authority: if type(item) is not int: raise ValueError("operator attention acknowledgement is invalid") typed_authority.append(item) if not isinstance(route, list) or len(route) != 2: raise ValueError("operator attention acknowledgement is invalid") typed_route: list[str] = [] for item in route: if not isinstance(item, str): raise ValueError("operator attention acknowledgement is invalid") typed_route.append(item) if ( type(actor_user_id) is not int or type(message_id) is not int or type(update_id) is not int or type(publication_generation) is not int ): raise ValueError("operator attention acknowledgement is invalid") try: record = GatewayOperatorAttentionAcknowledgement( session_id=str(raw["session_id"]), customer_key=str(raw["customer_key"]), attention_identity=str(raw["attention_identity"]), candidate_epoch=str(raw["candidate_epoch"]), actor_user_id=actor_user_id, authority=( typed_authority[0], typed_authority[1], typed_authority[2], ), route=(typed_route[0], typed_route[1]), message_id=message_id, update_id=update_id, callback_data=str(raw["callback_data"]), publication_generation=publication_generation, publication_dispatch_identity=str(raw["publication_dispatch_identity"]), publication_receipt_integrity=str(raw["publication_receipt_integrity"]), event_integrity=str(raw["event_integrity"]), ) except (KeyError, TypeError, ValueError) as exc: raise ValueError("operator attention acknowledgement is invalid") from exc unsigned = asdict(record) event_integrity = str(unsigned.pop("event_integrity")) if not hmac.compare_digest( event_integrity, hmac.new(key, canonical(unsigned), hashlib.sha256).hexdigest(), ): raise ValueError("operator attention acknowledgement integrity is invalid") return record def validate_attention_key(session_id: str, attention_identity: str) -> None: _text(session_id, "session ID") _digest(attention_identity, "attention identity") def _text(value: str, label: str) -> None: if not isinstance(value, str) or not value or len(value) > 256: raise ValueError(f"operator attention {label} is invalid") def _digest(value: str, label: str) -> None: if ( not isinstance(value, str) or len(value) != 64 or set(value) - set("0123456789abcdef") ): raise ValueError(f"operator attention {label} is invalid") __all__ = [ "GatewayOperatorAttentionAcknowledgement", "build_acknowledgement", "canonical", "parse_acknowledgement", "validate_attention_key", ]