"""SMS provider clients for MFA OTP delivery. Providers (Crisaleo ecosystem / Crilide modules): - crisaleo → https://api.crisaleo.eu/ (generic SMS module) - sms2sms → https://api.sms2sms.eu/ (sms2sms queue send) Auth headers (same pattern as RemoteEaterClient / Crilide API): X-Client-Id, X-Client-Secret, X-Api-Key Send paths (defaults, overridable in DB/env): crisaleo: POST /api/v1/modules/sms/send sms2sms: POST /api/v1/modules/sms2sms/send Body fields: to, content [, route for sms2sms] """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Any import httpx from .crypto import decrypt_token, encrypt_token from .settings import get_settings logger = logging.getLogger(__name__) PROVIDER_DEFAULTS: dict[str, dict[str, str]] = { "crisaleo": { "base_url": "https://api.crisaleo.eu", "send_path": "/api/v1/modules/sms/send", "label": "Crisaleo SMS", }, "sms2sms": { "base_url": "https://api.sms2sms.eu", "send_path": "/api/v1/modules/sms2sms/send", "label": "SMS2SMS", }, } class SmsProviderError(Exception): """Raised when an SMS provider rejects or fails a send request.""" def __init__(self, message: str, *, status_code: int | None = None) -> None: super().__init__(message) self.status_code = status_code @dataclass class SmsProviderConfig: code: str enabled: bool base_url: str client_id: str client_secret: str api_key: str send_path: str from_sender: str = "" route: str = "internal" timeout_seconds: float = 15.0 @property def label(self) -> str: return PROVIDER_DEFAULTS.get(self.code, {}).get("label", self.code) def is_configured(self) -> bool: return bool(self.enabled and self.base_url and self.client_id and self.client_secret and self.api_key) def default_provider_config(code: str) -> SmsProviderConfig: defaults = PROVIDER_DEFAULTS.get(code) or PROVIDER_DEFAULTS["crisaleo"] settings = get_settings() if code == "crisaleo": return SmsProviderConfig( code="crisaleo", enabled=bool(getattr(settings, "sms_crisaleo_enabled", False)), base_url=(getattr(settings, "sms_crisaleo_base_url", None) or defaults["base_url"]).rstrip("/"), client_id=getattr(settings, "sms_crisaleo_client_id", "") or "", client_secret=getattr(settings, "sms_crisaleo_client_secret", "") or "", api_key=getattr(settings, "sms_crisaleo_api_key", "") or "", send_path=getattr(settings, "sms_crisaleo_send_path", None) or defaults["send_path"], from_sender=getattr(settings, "sms_crisaleo_from", "") or "", route="internal", timeout_seconds=float(getattr(settings, "sms_timeout_seconds", 15) or 15), ) if code == "sms2sms": return SmsProviderConfig( code="sms2sms", enabled=bool(getattr(settings, "sms_sms2sms_enabled", False)), base_url=(getattr(settings, "sms_sms2sms_base_url", None) or defaults["base_url"]).rstrip("/"), client_id=getattr(settings, "sms_sms2sms_client_id", "") or "", client_secret=getattr(settings, "sms_sms2sms_client_secret", "") or "", api_key=getattr(settings, "sms_sms2sms_api_key", "") or "", send_path=getattr(settings, "sms_sms2sms_send_path", None) or defaults["send_path"], from_sender=getattr(settings, "sms_sms2sms_from", "") or "", route=getattr(settings, "sms_sms2sms_route", None) or "internal", timeout_seconds=float(getattr(settings, "sms_timeout_seconds", 15) or 15), ) raise ValueError(f"Unknown SMS provider: {code}") def config_from_row(row: Any) -> SmsProviderConfig: """Build config from an MfaSmsProvider ORM row (secrets decrypted).""" defaults = PROVIDER_DEFAULTS.get(row.code, {}) return SmsProviderConfig( code=row.code, enabled=bool(row.enabled), base_url=(row.base_url or defaults.get("base_url") or "").rstrip("/"), client_id=row.client_id or "", client_secret=decrypt_token(row.client_secret_enc) or "", api_key=decrypt_token(row.api_key_enc) or "", send_path=row.send_path or defaults.get("send_path") or "/api/v1/modules/sms/send", from_sender=row.from_sender or "", route=row.route or "internal", timeout_seconds=float(getattr(row, "timeout_seconds", 15) or 15), ) def encrypt_secret(value: str | None) -> str | None: if not value: return value return encrypt_token(value) def send_sms(config: SmsProviderConfig, to: str, content: str) -> dict[str, Any]: """Send an SMS via the configured provider. Returns provider response payload.""" if not config.is_configured(): raise SmsProviderError(f"SMS provider {config.code} is not configured or disabled") to = (to or "").strip() content = (content or "").strip() if not to: raise SmsProviderError("Destination phone number is required") if not content: raise SmsProviderError("SMS content is required") if len(content) > 480: content = content[:480] url = f"{config.base_url.rstrip('/')}/{config.send_path.lstrip('/')}" headers = { "Accept": "application/json", "Content-Type": "application/json", "X-Client-Id": config.client_id, "X-Client-Secret": config.client_secret, "X-Api-Key": config.api_key, } body: dict[str, Any] = { "to": to, "content": content, "message": content, "phone": to, } if config.from_sender: body["from"] = config.from_sender if config.code == "sms2sms": body["route"] = config.route if config.route in {"internal", "global"} else "internal" try: with httpx.Client(timeout=config.timeout_seconds) as client: response = client.post(url, headers=headers, json=body) except httpx.HTTPError as exc: logger.warning("SMS provider %s HTTP error: %s", config.code, exc) raise SmsProviderError(f"SMS provider unreachable: {exc}") from exc payload: Any try: payload = response.json() except ValueError: payload = {"raw": response.text[:500]} if response.status_code < 200 or response.status_code >= 300: message = "" if isinstance(payload, dict): message = str(payload.get("message") or payload.get("error") or payload.get("detail") or "") raise SmsProviderError( message or f"SMS provider error (HTTP {response.status_code})", status_code=response.status_code, ) if isinstance(payload, dict): # sms2sms may return success:false with HTTP 200 if payload.get("success") is False or payload.get("error"): message = str(payload.get("message") or payload.get("error") or "SMS send failed") raise SmsProviderError(message, status_code=response.status_code) return payload return {"data": payload} def resolve_provider_config(db, code: str | None = None) -> SmsProviderConfig: """Load provider from DB when present, otherwise fall back to env defaults.""" from . import models preferred = (code or "").strip().lower() rows = [] if db is not None: try: query = db.query(models.MfaSmsProvider) if preferred: query = query.filter(models.MfaSmsProvider.code == preferred) rows = query.all() except Exception: rows = [] if preferred and rows: return config_from_row(rows[0]) if preferred: return default_provider_config(preferred) # Prefer first enabled DB row, else first enabled env provider. for row in rows: cfg = config_from_row(row) if cfg.is_configured(): return cfg for code_name in ("crisaleo", "sms2sms"): cfg = default_provider_config(code_name) if cfg.is_configured(): return cfg if rows: return config_from_row(rows[0]) return default_provider_config("crisaleo")