236 lines
7 KiB
Python
236 lines
7 KiB
Python
import hashlib
|
|
import logging
|
|
import re
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import httpx
|
|
from fastapi import HTTPException, status
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
settings = get_settings()
|
|
|
|
INDIA_COUNTRY_CODE = "91"
|
|
OTP_LENGTH = 6
|
|
OTP_TTL_MINUTES = 10
|
|
OTP_RESEND_SECONDS = 60
|
|
MAX_VERIFY_ATTEMPTS = 5
|
|
|
|
INDIAN_MOBILE_RE = re.compile(r"^[6-9]\d{9}$")
|
|
|
|
|
|
def phone_key(country_code: str, phone: str) -> str:
|
|
return f"{country_code.strip()}{phone.strip()}"
|
|
|
|
|
|
def format_phone_e164(country_code: str, phone: str) -> str:
|
|
return f"+{phone_key(country_code, phone)}"
|
|
|
|
|
|
def parse_india_phone(phone_number: str) -> tuple[str, str]:
|
|
digits = re.sub(r"\D", "", phone_number or "")
|
|
if digits.startswith("91") and len(digits) == 12:
|
|
local = digits[2:]
|
|
elif len(digits) == 10:
|
|
local = digits
|
|
else:
|
|
raise ValueError(
|
|
"Multi-factor authentication requires an Indian (+91) mobile number on your account."
|
|
)
|
|
if not INDIAN_MOBILE_RE.fullmatch(local):
|
|
raise ValueError(
|
|
"Multi-factor authentication requires a valid Indian mobile number on your account."
|
|
)
|
|
return INDIA_COUNTRY_CODE, local
|
|
|
|
|
|
def _hash_otp(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _generate_otp() -> str:
|
|
return f"{secrets.randbelow(10 ** OTP_LENGTH):0{OTP_LENGTH}d}"
|
|
|
|
|
|
async def send_phone_otp(
|
|
database,
|
|
country_code: str,
|
|
phone: str,
|
|
*,
|
|
purpose: str,
|
|
purpose_key: str,
|
|
) -> tuple[str, dict]:
|
|
if country_code != INDIA_COUNTRY_CODE:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Phone verification is only available for India (+91).",
|
|
)
|
|
local = phone.strip()
|
|
if not INDIAN_MOBILE_RE.fullmatch(local):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Enter a valid 10-digit Indian mobile number.",
|
|
)
|
|
if not settings.urmi_otp_api_key:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Phone verification is temporarily unavailable.",
|
|
)
|
|
|
|
key = phone_key(country_code, local)
|
|
now = datetime.now(timezone.utc)
|
|
existing = await database.signup_otps.find_one(
|
|
{
|
|
"phone_key": key,
|
|
"purpose": purpose,
|
|
"purpose_key": purpose_key,
|
|
"consumed": False,
|
|
},
|
|
sort=[("created_at", -1)],
|
|
)
|
|
if existing and existing.get("sent_at"):
|
|
retry_at = existing["sent_at"] + timedelta(seconds=OTP_RESEND_SECONDS)
|
|
if retry_at > now:
|
|
wait = int((retry_at - now).total_seconds())
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail=f"Wait {wait}s before requesting another code.",
|
|
)
|
|
|
|
otp = _generate_otp()
|
|
payload = {
|
|
"project_name": settings.urmi_otp_project_name,
|
|
"template_type": settings.urmi_otp_template_type,
|
|
"to": key,
|
|
"variables": {
|
|
"otp": otp,
|
|
"validity_minutes": str(OTP_TTL_MINUTES),
|
|
},
|
|
}
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
"X-Api-Key": settings.urmi_otp_api_key,
|
|
}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=20.0) as client:
|
|
response = await client.post(
|
|
settings.urmi_otp_api_url,
|
|
json=payload,
|
|
headers=headers,
|
|
)
|
|
except httpx.HTTPError as error:
|
|
logger.exception("Urmi OTP request failed for %s", key)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail="Could not send verification code. Try again shortly.",
|
|
) from error
|
|
|
|
if response.status_code >= 400:
|
|
logger.warning(
|
|
"Urmi OTP rejected request for %s: %s %s",
|
|
key,
|
|
response.status_code,
|
|
response.text[:300],
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail="Could not send verification code. Try again shortly.",
|
|
)
|
|
|
|
expires_at = now + timedelta(minutes=OTP_TTL_MINUTES)
|
|
await database.signup_otps.update_many(
|
|
{
|
|
"phone_key": key,
|
|
"purpose": purpose,
|
|
"purpose_key": purpose_key,
|
|
"consumed": False,
|
|
},
|
|
{"$set": {"consumed": True}},
|
|
)
|
|
await database.signup_otps.insert_one(
|
|
{
|
|
"phone_key": key,
|
|
"purpose": purpose,
|
|
"purpose_key": purpose_key,
|
|
"otp_hash": _hash_otp(otp),
|
|
"attempts": 0,
|
|
"consumed": False,
|
|
"sent_at": now,
|
|
"created_at": now,
|
|
"expires_at": expires_at,
|
|
}
|
|
)
|
|
return otp, {
|
|
"message": "Verification code sent.",
|
|
"expires_in_minutes": OTP_TTL_MINUTES,
|
|
}
|
|
|
|
|
|
async def send_signup_otp(database, country_code: str, phone: str) -> dict:
|
|
key = phone_key(country_code, phone.strip())
|
|
_, meta = await send_phone_otp(
|
|
database,
|
|
country_code,
|
|
phone,
|
|
purpose="signup",
|
|
purpose_key=key,
|
|
)
|
|
return meta
|
|
|
|
|
|
async def verify_signup_otp(database, country_code: str, phone: str, otp: str) -> None:
|
|
if country_code != INDIA_COUNTRY_CODE:
|
|
return
|
|
|
|
local = phone.strip()
|
|
code = otp.strip()
|
|
if not INDIAN_MOBILE_RE.fullmatch(local):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Enter a valid 10-digit Indian mobile number.",
|
|
)
|
|
if not re.fullmatch(rf"\d{{{OTP_LENGTH}}}", code):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Enter the 6-digit verification code.",
|
|
)
|
|
|
|
key = phone_key(country_code, local)
|
|
now = datetime.now(timezone.utc)
|
|
record = await database.signup_otps.find_one(
|
|
{
|
|
"phone_key": key,
|
|
"purpose": "signup",
|
|
"purpose_key": key,
|
|
"consumed": False,
|
|
},
|
|
sort=[("created_at", -1)],
|
|
)
|
|
if record is None or record.get("expires_at") <= now:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Verification code expired. Request a new one.",
|
|
)
|
|
if record.get("attempts", 0) >= MAX_VERIFY_ATTEMPTS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Too many incorrect attempts. Request a new code.",
|
|
)
|
|
|
|
if _hash_otp(code) != record["otp_hash"]:
|
|
await database.signup_otps.update_one(
|
|
{"_id": record["_id"]},
|
|
{"$inc": {"attempts": 1}},
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Incorrect verification code.",
|
|
)
|
|
|
|
await database.signup_otps.update_one(
|
|
{"_id": record["_id"]},
|
|
{"$set": {"consumed": True, "verified_at": now}},
|
|
)
|