"""Twilio SMS provider — international coverage (all countries except Iran). API documentation: https://www.twilio.com/docs/messaging/api Twilio is chosen as the international fallback because: - Covers 180+ countries - Competitive pricing for OTP/transactional SMS - Twilio Verify API for OTP (improved deliverability + fraud protection) - Simple HTTP REST API (no SDK dependency required) Credentials (stored in SmsProvider.credentials JSON): account_sid : str — Twilio Account SID (starts with "AC") auth_token : str — Twilio Auth Token from_number : str — Twilio sender number or Messaging Service SID Extra params (stored in SmsProvider.extra_params JSON): use_verify : bool (default False) — use Twilio Verify service for OTP verify_service_sid : str — required if use_verify=True (starts with "VA") otp_message_template: str — plain SMS OTP template, e.g. "Your code is {code}" Used when use_verify=False (default: "Your verification code: {code}") """ from __future__ import annotations import base64 import json as json_lib import structlog from decimal import Decimal from typing import Any from urllib import request as urllib_request from urllib.error import HTTPError, URLError from urllib.parse import urlencode from simorgh.apps.notifications.sms.base import SmsSendResult _log = structlog.get_logger("simorgh.notifications.sms.twilio") _MESSAGES_URL = "https://api.twilio.com/2010-04-01/Accounts/{account_sid}/Messages.json" _VERIFY_CREATE_URL = "https://verify.twilio.com/v2/Services/{service_sid}/Verifications" _TIMEOUT = 20 # seconds _DEFAULT_OTP_TEMPLATE = "Your verification code: {code}" def _basic_auth(account_sid: str, auth_token: str) -> str: credentials = f"{account_sid}:{auth_token}".encode("utf-8") return "Basic " + base64.b64encode(credentials).decode("utf-8") def _http_post_form(url: str, data: dict, auth_header: str) -> dict: """POST application/x-www-form-urlencoded — Twilio's preferred format.""" encoded = urlencode(data).encode("utf-8") req = urllib_request.Request( url, data=encoded, method="POST", headers={ "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json", "Authorization": auth_header, }, ) try: with urllib_request.urlopen(req, timeout=_TIMEOUT) as resp: return json_lib.loads(resp.read().decode("utf-8")) except HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") try: parsed = json_lib.loads(body) except Exception: raise RuntimeError(f"Twilio HTTP {exc.code}: {body}") from exc msg = parsed.get("message", body) raise RuntimeError(f"Twilio error {parsed.get('code', exc.code)}: {msg}") from exc except URLError as exc: raise RuntimeError(f"Twilio connection error: {exc.reason}") from exc class TwilioProvider: """Twilio — global SMS provider for international numbers.""" slug = "twilio" name = "Twilio" # Empty list = handles all countries; the routing engine treats Iran (+98) # as excluded because SmsIrProvider has explicit "+98" with higher priority. # However, we set supported_country_codes to exclude +98 explicitly so the # routing logic always prefers SMS.ir for Iran. supported_country_codes: list[str] = [] # all countries (fallback) priority: int = 10 # lower than SMS.ir so it only handles non-Iran def __init__( self, account_sid: str, auth_token: str, from_number: str, verify_service_sid: str = "", otp_message_template: str = _DEFAULT_OTP_TEMPLATE, use_verify: bool = False, ) -> None: self._account_sid = account_sid self._auth_token = auth_token self._from_number = from_number self._verify_service_sid = verify_service_sid self._otp_message_template = otp_message_template self._use_verify = use_verify and bool(verify_service_sid) self._auth_header = _basic_auth(account_sid, auth_token) # ------------------------------------------------------------------ # Public interface # ------------------------------------------------------------------ def send( self, *, mobile: str, text: str, tenant_id: int | None = None, ) -> SmsSendResult: """Send a plain-text SMS via Twilio Messaging API.""" url = _MESSAGES_URL.format(account_sid=self._account_sid) payload: dict[str, Any] = { "To": mobile, "From": self._from_number, "Body": text, } _log.info("twilio.send", mobile=mobile, tenant_id=tenant_id) try: resp = _http_post_form(url, payload, self._auth_header) except Exception as exc: _log.error("twilio.send_error", error=str(exc), mobile=mobile) return SmsSendResult(success=False, error=str(exc), raw_response={}) return self._parse_messages_response(resp) def send_otp( self, *, mobile: str, code: str, tenant_id: int | None = None, extra: dict | None = None, ) -> SmsSendResult: """Send an OTP via Twilio Verify (if configured) or plain SMS.""" if self._use_verify: return self._send_via_verify(mobile=mobile, code=code) # Fall back to plain SMS with the configured template text = self._otp_message_template.format(code=code, **(extra or {})) return self.send(mobile=mobile, text=text, tenant_id=tenant_id) def supports_country_code(self, dial_code: str) -> bool: if not self.supported_country_codes: return True # all countries return dial_code in self.supported_country_codes # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _send_via_verify(self, *, mobile: str, code: str) -> SmsSendResult: """Use Twilio Verify v2 to deliver the OTP (better deliverability).""" url = _VERIFY_CREATE_URL.format(service_sid=self._verify_service_sid) payload: dict[str, Any] = { "To": mobile, "Channel": "sms", "CustomCode": code, } _log.info("twilio.send_otp_verify", mobile=mobile) try: resp = _http_post_form(url, payload, self._auth_header) except Exception as exc: _log.error("twilio.send_otp_error", error=str(exc), mobile=mobile) return SmsSendResult(success=False, error=str(exc), raw_response={}) sid = resp.get("sid", "") status = resp.get("status", "") success = status in ("pending", "approved") return SmsSendResult( success=success, provider_message_id=sid, error="" if success else resp.get("message", f"status={status}"), raw_response=resp, ) @staticmethod def _parse_messages_response(resp: dict) -> SmsSendResult: # Twilio returns status "queued" / "sent" / "delivered" / "failed" etc. sid = resp.get("sid", "") status = resp.get("status", "") price = resp.get("price") # negative string e.g. "-0.0075" error_message = resp.get("error_message") or "" if status in ("failed", "undelivered") or resp.get("error_code"): return SmsSendResult( success=False, provider_message_id=sid, error=error_message or status, raw_response=resp, ) cost = None if price: try: cost = abs(Decimal(str(price))) except Exception: pass return SmsSendResult( success=True, provider_message_id=sid, cost=cost, raw_response=resp, )