106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
"""Outbound SMS.
|
|
|
|
The provider is behind an interface so the platform can be developed and tested
|
|
without credentials, and so switching aggregators does not touch call sites.
|
|
|
|
Nothing here is allowed to raise into a request: a failed SMS must not turn a
|
|
sign-in attempt into a 500, it must simply not deliver.
|
|
"""
|
|
|
|
import logging
|
|
|
|
import httpx
|
|
from django.conf import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SMSError(Exception):
|
|
"""The provider refused or could not be reached."""
|
|
|
|
|
|
def _text(value):
|
|
return value or ""
|
|
|
|
|
|
def send_console(to_number, body):
|
|
"""Print to the server log. The default when no provider is configured."""
|
|
logger.warning("SMS to %s: %s", to_number, body)
|
|
return True
|
|
|
|
|
|
def send_kavenegar(to_number, body):
|
|
"""Kavenegar Verify API.
|
|
|
|
Chosen because it is what the legacy auth service already uses, so a
|
|
migration does not need a second aggregator contract.
|
|
"""
|
|
api_key = _text(getattr(settings, "KAVENEGAR_API_KEY", ""))
|
|
sender = _text(getattr(settings, "KAVENEGAR_SENDER", ""))
|
|
|
|
if not api_key:
|
|
raise SMSError("KAVENEGAR_API_KEY is not configured")
|
|
|
|
try:
|
|
response = httpx.post(
|
|
"https://api.kavenegar.com/v1/"
|
|
f"{api_key}/verify/lookup.json",
|
|
data={"receptor": to_number, "message": body, "sender": sender},
|
|
timeout=float(getattr(settings, "SMS_TIMEOUT", 10)),
|
|
)
|
|
except Exception as exc:
|
|
raise SMSError(f"Provider unreachable: {exc}") from exc
|
|
|
|
if response.status_code >= 400:
|
|
raise SMSError(f"Provider returned {response.status_code}")
|
|
|
|
try:
|
|
payload = response.json()
|
|
except Exception as exc:
|
|
raise SMSError("Provider returned invalid JSON") from exc
|
|
|
|
if payload.get("return", {}).get("status") != 200:
|
|
# Log the code, never the whole payload: it echoes the recipient.
|
|
logger.warning(
|
|
"Kavenegar refused the message for %s: %s",
|
|
to_number,
|
|
payload.get("return", {}).get("message"),
|
|
)
|
|
raise SMSError("Provider refused the message")
|
|
|
|
return True
|
|
|
|
|
|
PROVIDERS = {
|
|
"console": send_console,
|
|
"kavenegar": send_kavenegar,
|
|
}
|
|
|
|
|
|
def send_sms(to_number, body):
|
|
"""Dispatch to the configured provider. Never raises."""
|
|
provider_name = _text(getattr(settings, "SMS_PROVIDER", "")).lower()
|
|
provider = PROVIDERS.get(provider_name)
|
|
|
|
if provider is None:
|
|
if getattr(settings, "DEBUG", False) or _text(
|
|
getattr(settings, "SMS_PROVIDER", "")
|
|
) == "":
|
|
# Unconfigured is only acceptable outside production; settings
|
|
# refuses to boot production without a provider.
|
|
provider = send_console
|
|
else:
|
|
logger.error(
|
|
"SMS_PROVIDER=%s is not a known provider; falling back to console",
|
|
provider_name,
|
|
)
|
|
provider = send_console
|
|
|
|
try:
|
|
return provider(to_number, body)
|
|
except SMSError:
|
|
logger.exception("SMS provider %s failed for %s", provider_name, to_number)
|
|
raise
|
|
except Exception:
|
|
logger.exception("Unexpected SMS failure for %s", to_number)
|
|
raise SMSError("Unexpected provider failure") |