from __future__ import annotations import hashlib import hmac import json import re import string from dataclasses import dataclass from datetime import datetime from typing import Any import phonenumbers from app.db import DeliveryStatus, SendStatus GSM_BASIC = ( "@£$¥èéùìòÇ\nØø\rÅåΔ_ΦΓΛΩΠΨΣΘΞ " "!\"#¤%&'()*+,-./0123456789:;<=>?" "¡ABCDEFGHIJKLMNOPQRSTUVWXYZÄÖÑܧ¿abcdefghijklmnopqrstuvwxyzäöñüà" ) GSM_EXTENDED = "^{}\\[~]|€" PHONE_RE = re.compile(r"^\+[1-9]\d{7,14}$") class DomainError(Exception): def __init__( self, code: str, status: int, message: str, details: dict[str, Any] | None = None ) -> None: self.code = code self.status = status self.message = message self.details = details or {} super().__init__(message) def normalize_phone(value: str) -> tuple[str, str, str]: if not PHONE_RE.fullmatch(value): raise DomainError("sms_request_invalid", 422, "phone_e164 must be valid E.164") try: parsed = phonenumbers.parse(value, None) except phonenumbers.NumberParseException: raise DomainError("sms_request_invalid", 422, "phone_e164 must be valid E.164") from None if not phonenumbers.is_valid_number(parsed): raise DomainError("sms_request_invalid", 422, "phone_e164 must be valid E.164") normalized = phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164) if normalized != value: raise DomainError("sms_request_invalid", 422, "phone_e164 must be canonical E.164") digits = normalized[1:] masked = f"+{digits[:1]}{'*' * max(0, len(digits) - 5)}{digits[-4:]}" return normalized, digits, masked def request_fingerprint(payload: dict[str, Any]) -> str: canonical = json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False ) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() def destination_hmac(phone_e164: str, key: bytes) -> str: return hmac.new(key, phone_e164.encode(), hashlib.sha256).hexdigest() def sms_parts(body: str) -> int: if not body or "\ufeff" in body or "\x00" in body: raise DomainError("sms_request_invalid", 422, "Rendered message contains invalid text") gsm_units = 0 for char in body: if char in GSM_BASIC: gsm_units += 1 elif char in GSM_EXTENDED: gsm_units += 2 else: total = len(body.encode("utf-16-be")) // 2 return 1 if total <= 70 else (total + 66) // 67 return 1 if gsm_units <= 160 else (gsm_units + 152) // 153 def render_template( body_template: str, placeholders: list[str], substitutions: dict[str, Any], max_parts: int, ) -> str: expected = set(placeholders) supplied = set(substitutions) if expected != supplied: raise DomainError( "sms_request_invalid", 422, "Substitutions do not match template placeholders", {"missing": sorted(expected - supplied), "unknown": sorted(supplied - expected)}, ) parsed = { field_name for _, field_name, format_spec, conversion in string.Formatter().parse(body_template) if field_name is not None and not format_spec and not conversion and field_name.isidentifier() } if parsed != expected or any( format_spec or conversion for _, field_name, format_spec, conversion in string.Formatter().parse(body_template) if field_name is not None ): raise DomainError("sms_request_invalid", 422, "Template placeholder contract is invalid") body = body_template.format_map({key: str(value) for key, value in substitutions.items()}) if len(body.encode("utf-8")) > 2048 or sms_parts(body) > max_parts: raise DomainError("sms_request_invalid", 422, "Rendered message exceeds template limit") return body @dataclass(frozen=True) class ProviderResult: send_status: SendStatus http_status: int | None = None message_uuid: str | None = None external_id: str | None = None error_code: str | None = None error_message: str | None = None retry_safe: bool = False contract_violation: bool = False DELIVERY_RANK = { DeliveryStatus.UNKNOWN: 0, DeliveryStatus.SENT: 1, DeliveryStatus.DELIVERED: 2, DeliveryStatus.UNDELIVERED: 2, DeliveryStatus.UNSENT: 2, } def delivery_transition(current: DeliveryStatus, incoming: str) -> DeliveryStatus | None: try: target = DeliveryStatus(incoming.lower()) except ValueError: return None if DELIVERY_RANK[target] < DELIVERY_RANK[current]: return current if DELIVERY_RANK[target] == DELIVERY_RANK[current] and target != current: return current return target def parse_status_time(value: str) -> datetime: try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: raise DomainError("callback_invalid", 422, "Invalid callback status_time") from None if parsed.tzinfo is None: raise DomainError("callback_invalid", 422, "Callback status_time requires timezone") return parsed