import hashlib import json import re from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional, Tuple from uuid import uuid4 from urllib.parse import urlparse from fastapi import HTTPException, Request import requests from app.core.config import ( SUPABASE_URL, SUPABASE_ANON_KEY, SUPABASE_SERVICE_ROLE_KEY, SUPABASE_TIMEOUT_SECONDS, PASSWORD_RESET_REDIRECT_URL, INTERNAL_LOGIN_EMAIL_SUFFIX, MOBILE_REDIRECT_SCHEME_PATTERN, PARKING_LOCATION_PATTERN, CAMERA_SOURCE_PATTERN, REDIRECT_ALLOWED_HTTP_HOSTS, REDIRECT_ALLOWED_DEEPLINK_SCHEMES, SUPPORTED_EVENT_TYPES, SUPPORTED_OTP_CHANNELS, PHONE_PATTERN, PROFILE_SELECT_FIELDS, VEHICLE_SELECT_FIELDS, supabase_configured, LOGGER, ) from app.security.auth import ( _normalize_spaces, _supabase_headers, _extract_response_error_message, is_valid_redirect_url, _looks_like_jwt, ) TZ_UTC = timezone.utc def _normalize_location_key_for_config(value: Optional[str]) -> Optional[str]: if value is None: return None raw = str(value).strip().upper() if not raw: return None return "".join(raw.split()) or None def _normalize_parking_location(value: Optional[str]) -> Optional[str]: raw = _normalize_spaces(value or "") if not raw: return None if not PARKING_LOCATION_PATTERN.fullmatch(raw): raise HTTPException(status_code=400, detail="Invalid parking_location format.") parts = [_normalize_location_key_for_config(part) for part in raw.split(",")] parts = [p for p in parts if p] return ",".join(parts) if parts else None def _normalize_camera_source(value: Optional[str]) -> Optional[str]: raw = _normalize_spaces(value or "") if not raw: return None if not CAMERA_SOURCE_PATTERN.fullmatch(raw): raise HTTPException(status_code=400, detail="Invalid camera_source. Use alphanumeric, underscore, or dash (max 64 chars).") return raw def _normalize_name_part(value: Any) -> str: return _normalize_spaces(str(value or "")) def _compose_full_name(first_name: Any, last_name: Any) -> str: return _normalize_spaces(f"{_normalize_name_part(first_name)} {_normalize_name_part(last_name)}") def _split_full_name_parts(full_name: Any) -> Tuple[str, str]: normalized = _normalize_name_part(full_name) if not normalized: return "", "" parts = normalized.split(" ", 1) if len(parts) == 1: return parts[0], "" return parts[0], _normalize_spaces(parts[1]) def _resolve_profile_name_fields(payload: Optional[Dict[str, Any]], *, fallback_full_name: Any = None, fallback_first_name: Any = None, fallback_last_name: Any = None) -> Tuple[str, str, str]: body = payload if isinstance(payload, dict) else {} first_name = _normalize_name_part(body.get("first_name") or body.get("firstName") or fallback_first_name) last_name = _normalize_name_part(body.get("last_name") or body.get("lastName") or fallback_last_name) legacy_full_name = _normalize_name_part(body.get("full_name") or body.get("name") or fallback_full_name) if not first_name and not last_name and legacy_full_name: first_name, last_name = _split_full_name_parts(legacy_full_name) full_name = _compose_full_name(first_name, last_name) if not full_name and legacy_full_name: full_name = legacy_full_name if not first_name and not last_name: first_name, last_name = _split_full_name_parts(legacy_full_name) return first_name, last_name, full_name def _resolve_public_base_url(request: Request) -> str: forwarded_proto = _normalize_spaces((request.headers.get("x-forwarded-proto") or "").split(",")[0]) forwarded_host = _normalize_spaces((request.headers.get("x-forwarded-host") or "").split(",")[0]) if forwarded_proto and forwarded_host: return f"{forwarded_proto}://{forwarded_host}".rstrip("/") return str(request.base_url).rstrip("/") def _resolve_default_password_reset_redirect_url(request: Request) -> str: configured = _normalize_spaces(PASSWORD_RESET_REDIRECT_URL) if configured: parsed = urlparse(configured) if parsed.scheme and parsed.scheme not in {"http", "https"}: return f"{_resolve_public_base_url(request)}/auth/reset-password-bridge" return configured return f"{_resolve_public_base_url(request)}/auth/reset-password-page" def _resolve_password_reset_deep_link_url() -> str: configured = _normalize_spaces(PASSWORD_RESET_REDIRECT_URL) if is_valid_redirect_url(configured): parsed = urlparse(configured) if parsed.scheme not in {"http", "https"}: return configured return "ainelaql://auth/reset" def _split_arabic_plate(plate_text_ar: str) -> Tuple[Optional[str], Optional[str]]: raw = _normalize_spaces(plate_text_ar) if not raw or raw.upper() == "N/A" or "|" not in raw: return None, None letters, numbers = [part.strip() for part in raw.split("|", 1)] letters = _normalize_spaces(letters) numbers = _normalize_spaces(numbers) if not letters or not numbers: return None, None return letters, numbers def _split_english_plate(plate_text_en: str) -> Tuple[Optional[str], Optional[str]]: raw = _normalize_spaces(plate_text_en) if not raw or raw.upper() == "N/A" or "|" not in raw: return None, None letters, numbers = [part.strip() for part in raw.split("|", 1)] letters = _normalize_spaces(letters) numbers = numbers.replace(" ", "") if not letters and not numbers: return None, None return letters or None, numbers or None def _is_plate_detected(plate_info: Dict[str, Any]) -> bool: if not isinstance(plate_info, dict): return False arabic = str(plate_info.get("arabic") or "").strip().upper() english = str(plate_info.get("english") or "").strip().upper() return arabic not in {"", "N/A"} or english not in {"", "N/A"} def _build_plate_payload(plate_info: Dict[str, Any]) -> Dict[str, Any]: arabic_text = str(plate_info.get("arabic") or "N/A") english_text = str(plate_info.get("english") or "N/A") ar_letters, ar_numbers = _split_arabic_plate(arabic_text) en_letters, en_numbers = _split_english_plate(english_text) return { "arabic_text": arabic_text, "english_text": english_text, "arabic": {"letters": ar_letters, "numbers": ar_numbers}, "english": {"letters": en_letters, "numbers": en_numbers}, } def _normalize_event_type(value: Optional[str]) -> str: raw = (value or "ocr_scan").strip().lower() if raw not in SUPPORTED_EVENT_TYPES: raise HTTPException(status_code=400, detail=f"Invalid event_type '{raw}'. Allowed: {sorted(SUPPORTED_EVENT_TYPES)}") return raw def _normalize_otp_channel(raw_value: Optional[str]) -> str: raw = _normalize_spaces(str(raw_value or "")).lower() if raw in {"email", "mail"}: return "email" if raw in {"phone", "sms", "phone_number"}: return "phone" raise HTTPException(status_code=400, detail="channel must be 'email' or 'phone'.") def _missing_schema_hint(raw_error: str) -> str: err = (raw_error or "").strip() if "42703" in err: return "Database schema is missing new auth/profile columns. Run supabase SQL migrations first." return err[:280] if err else "Schema validation failed." def _ensure_auth_profile_schema_ready() -> None: if not supabase_configured(): raise HTTPException(status_code=503, detail="Supabase is not configured.") response = requests.get( f"{SUPABASE_URL}/rest/v1/profiles", params={"select": "id,first_name,last_name,full_name,username,national_id,phone_number,email,role", "limit": "1"}, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY), timeout=SUPABASE_TIMEOUT_SECONDS, ) if response.status_code != 200: raise HTTPException(status_code=503, detail=f"Auth schema check failed: {_missing_schema_hint(response.text)}") def _is_internal_login_email(email: Optional[str]) -> bool: normalized = _normalize_spaces(str(email or "")).lower() if not normalized: return False return normalized.endswith(INTERNAL_LOGIN_EMAIL_SUFFIX) def _build_internal_login_email(username: str, phone_number: str) -> str: seed = re.sub(r"[^a-z0-9]+", "", f"{username}-{phone_number}".lower()) if not seed: seed = uuid4().hex[:12] return f"{seed[:24]}-{uuid4().hex[:8]}@ainel-aql.local" def _is_supabase_duplicate_auth_user_error(error_message: Optional[str]) -> bool: normalized = _normalize_spaces(str(error_message or "")).lower() if not normalized: return False duplicate_hints = ["already registered", "already exists", "already in use", "user already registered", "email address already in use", "duplicate key value"] return any(hint in normalized for hint in duplicate_hints) def _find_supabase_auth_user_by_email(email: str) -> Optional[Dict[str, Any]]: if not supabase_configured(): return None normalized_email = _normalize_spaces(email).lower() if not normalized_email or "@" not in normalized_email: return None max_pages = 20 per_page = 200 for page in range(1, max_pages + 1): try: response = requests.get( f"{SUPABASE_URL}/auth/v1/admin/users", params={"page": str(page), "per_page": str(per_page)}, headers={"apikey": SUPABASE_SERVICE_ROLE_KEY, "Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}"}, timeout=SUPABASE_TIMEOUT_SECONDS, ) except requests.RequestException: return None if response.status_code != 200: return None try: payload = response.json() except ValueError: return None users = payload.get("users") if isinstance(payload, dict) else payload if not isinstance(users, list): return None for entry in users: if not isinstance(entry, dict): continue candidate_email = _normalize_spaces(str(entry.get("email") or "")).lower() if candidate_email == normalized_email: return entry total_count = None if isinstance(payload, dict): try: total_count = int(payload.get("total", 0)) except (TypeError, ValueError): pass if total_count is not None and page * per_page >= total_count: break if len(users) < per_page: break return None def _update_supabase_auth_user(user_id: str, patch_payload: Dict[str, Any]) -> None: if not user_id or not patch_payload: return try: response = requests.put( f"{SUPABASE_URL}/auth/v1/admin/users/{user_id}", json=patch_payload, headers={"apikey": SUPABASE_SERVICE_ROLE_KEY, "Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}", "Content-Type": "application/json"}, timeout=SUPABASE_TIMEOUT_SECONDS, ) except requests.RequestException as exc: raise RuntimeError(f"Supabase auth update failed: {exc}") from exc if response.status_code not in {200, 201}: raise RuntimeError(f"Supabase auth update failed ({response.status_code}): {_extract_response_error_message(response)}") def _extract_supabase_session_tokens(payload: Dict[str, Any]) -> Dict[str, Any]: session_obj = payload.get("session") if isinstance(payload.get("session"), dict) else {} access_token = _normalize_spaces(str(payload.get("access_token") or session_obj.get("access_token") or "")) refresh_token = _normalize_spaces(str(payload.get("refresh_token") or session_obj.get("refresh_token") or "")) expires_in_raw = payload.get("expires_in") if payload.get("expires_in") is not None else session_obj.get("expires_in") return {"access_token": access_token, "refresh_token": refresh_token, "expires_in": expires_in_raw} def _extract_registration_vehicle_payload(payload: Dict[str, Any]) -> Optional[Dict[str, str]]: raw_vehicle = payload.get("vehicle") if isinstance(payload.get("vehicle"), dict) else {} def _pick(*keys: str) -> Optional[str]: for key in keys: value = raw_vehicle.get(key) if isinstance(raw_vehicle, dict) else None if value is None: value = payload.get(key) if value is None: continue normalized_value = _normalize_spaces(str(value)) if normalized_value: return normalized_value return None plate_letters_ar = _pick("plate_letters_ar", "plate_letters") plate_numbers_ar = _pick("plate_numbers_ar", "plate_numbers") plate_letters_en = _pick("plate_letters_en") plate_numbers_en = _pick("plate_numbers_en") car_name = _pick("car_name", "nickname") car_model = _pick("car_model", "model") car_color = _pick("car_color", "color") has_any = any([plate_letters_ar, plate_numbers_ar, plate_letters_en, plate_numbers_en, car_name, car_model, car_color]) if not has_any: return None if not plate_letters_ar or not plate_numbers_ar: raise HTTPException(status_code=400, detail="Vehicle plate details are incomplete. Provide plate_letters_ar and plate_numbers_ar.") vehicle_payload: Dict[str, str] = {"plate_letters_ar": plate_letters_ar, "plate_numbers_ar": plate_numbers_ar} if plate_letters_en: vehicle_payload["plate_letters_en"] = plate_letters_en if plate_numbers_en: vehicle_payload["plate_numbers_en"] = plate_numbers_en if car_name: vehicle_payload["car_name"] = car_name if car_model: vehicle_payload["car_model"] = car_model if car_color: vehicle_payload["car_color"] = car_color return vehicle_payload def _create_vehicle_for_owner(*, owner_id: str, vehicle_payload: Dict[str, str]) -> Optional[Dict[str, Any]]: if not owner_id or not vehicle_payload: return None payload: Dict[str, Any] = {"owner_id": owner_id, "plate_letters_ar": vehicle_payload["plate_letters_ar"], "plate_numbers_ar": vehicle_payload["plate_numbers_ar"]} for optional_key in ["plate_letters_en", "plate_numbers_en", "car_name", "car_model", "car_color"]: optional_value = vehicle_payload.get(optional_key) if optional_value: payload[optional_key] = optional_value response = requests.post( f"{SUPABASE_URL}/rest/v1/vehicles", json=payload, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json", prefer="return=representation"), timeout=SUPABASE_TIMEOUT_SECONDS, ) if response.status_code not in {200, 201}: raise HTTPException(status_code=400, detail=f"Failed to save vehicle: {_missing_schema_hint(response.text)}") rows = response.json() if isinstance(rows, list) and rows and isinstance(rows[0], dict): return rows[0] return None def _upsert_profile_after_register(*, user_id: str, first_name: str, last_name: str, username: str, phone_number: str, email: str, national_id: Optional[str], role: str = "user") -> Optional[Dict[str, Any]]: resolved_first = _normalize_name_part(first_name) resolved_last = _normalize_name_part(last_name) resolved_full = _compose_full_name(resolved_first, resolved_last) or _normalize_name_part(username) payload: Dict[str, Any] = {"id": user_id, "first_name": resolved_first, "last_name": resolved_last, "full_name": resolved_full, "username": username, "phone_number": phone_number, "email": email} if national_id: payload["national_id"] = national_id if role: payload["role"] = role.lower() response = requests.post( f"{SUPABASE_URL}/rest/v1/profiles", json=payload, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json", prefer="resolution=merge-duplicates,return=representation"), timeout=SUPABASE_TIMEOUT_SECONDS, ) if response.status_code not in {200, 201}: raise RuntimeError(f"Failed to upsert profile ({response.status_code}): {_missing_schema_hint(response.text)}") rows = response.json() if isinstance(rows, list) and rows and isinstance(rows[0], dict): return rows[0] from app.services.supabase_client import supabase_get_profile_by_user_id return supabase_get_profile_by_user_id(user_id) def _extract_created_by_user_id(request_user: Optional[Dict[str, Any]]) -> Optional[str]: if not isinstance(request_user, dict): return None if str(request_user.get("auth_type") or "").strip().lower() != "supabase": return None user_id = str(request_user.get("id") or "").strip() return user_id or None def _is_valid_http_url(url: str) -> bool: if not url: return False parsed = urlparse(url) return parsed.scheme in {"http", "https"} and bool(parsed.netloc) def _resolve_otp_identifier(*, channel: str, payload: Dict[str, Any], profile: Optional[Dict[str, Any]]) -> str: if channel not in SUPPORTED_OTP_CHANNELS: raise HTTPException(status_code=400, detail="Unsupported OTP channel.") if channel == "email": candidate = _normalize_spaces(str(payload.get("email") or (profile or {}).get("email") or "")).lower() if not candidate or "@" not in candidate: raise HTTPException(status_code=400, detail="A valid email is required for email OTP.") return candidate candidate = _normalize_spaces(str(payload.get("phone") or payload.get("phone_number") or (profile or {}).get("phone_number") or "")) if not candidate or not PHONE_PATTERN.fullmatch(candidate): raise HTTPException(status_code=400, detail="A valid phone number is required for phone OTP.") return candidate def _request_supabase_otp(*, channel: str, identifier: str, create_user: bool, redirect_to: Optional[str], purpose: Optional[str]) -> None: request_payload = {"create_user": bool(create_user)} if channel == "email": request_payload["email"] = identifier if redirect_to: if not _is_valid_redirect_url(redirect_to): raise HTTPException(status_code=400, detail="redirect_to must be a valid URL or deep-link.") request_payload["email_redirect_to"] = redirect_to request_payload["options"] = {"emailRedirectTo": redirect_to} else: request_payload["phone"] = identifier request_payload["channel"] = "sms" try: response = requests.post(f"{SUPABASE_URL}/auth/v1/otp", json=request_payload, headers={"apikey": SUPABASE_ANON_KEY, "Content-Type": "application/json"}, timeout=SUPABASE_TIMEOUT_SECONDS) except requests.RequestException as exc: raise HTTPException(status_code=503, detail=f"Supabase OTP request failed: {exc}") from exc if response.status_code not in {200, 201}: error_data = response.json() if response.text else {} error_msg = "" if isinstance(error_data, dict): error_msg = str(error_data.get("message") or error_data.get("error_description") or error_data.get("msg") or "").strip() if not error_msg: error_msg = response.text[:220] normalized_error = _normalize_spaces(error_msg).lower() normalized_purpose = _normalize_spaces(str(purpose or "")).lower() if "otp_disabled" in normalized_error or "signups not allowed for otp" in normalized_error: if normalized_purpose in {"register", "signup"}: raise HTTPException(status_code=422, detail="OTP signup is disabled in backend auth configuration (otp_disabled). purpose=register cannot proceed with OTP until OTP signups are enabled.") raise HTTPException(status_code=422, detail="OTP provider is disabled in backend auth configuration (otp_disabled).") raise HTTPException(status_code=response.status_code, detail=f"OTP request failed: {error_msg}") def _verify_supabase_otp(*, channel: str, identifier: str, token: str, verify_type: Optional[str], purpose: Optional[str] = None) -> Dict[str, Any]: import time token_value = _normalize_spaces(token) if not token_value: raise HTTPException(status_code=400, detail="otp_token is required.") primary_type = _normalize_spaces(str(verify_type or "")).lower() resolved_purpose = _normalize_spaces(str(purpose or "")).lower() if not primary_type: if resolved_purpose in {"register", "signup"} and channel == "email": primary_type = "signup" else: primary_type = "email" if channel == "email" else "sms" types_to_try = [primary_type] if resolved_purpose in {"register", "signup"} and "signup" not in types_to_try: types_to_try.insert(0, "signup") for fallback in ["email", "sms", "signup", "magiclink", "recovery", "invite"]: if fallback not in types_to_try: types_to_try.append(fallback) last_error = "Unknown error" for current_type in types_to_try: time.sleep(0.5) request_payload = {"token": token_value, "type": current_type, "email" if channel == "email" else "phone": identifier} try: LOGGER.info("Verifying OTP for %s with type: %s", identifier, current_type) response = requests.post(f"{SUPABASE_URL}/auth/v1/verify", json=request_payload, headers={"apikey": SUPABASE_ANON_KEY, "Content-Type": "application/json"}, timeout=SUPABASE_TIMEOUT_SECONDS) if response.status_code == 200: return response.json() error_code = "" try: body = response.json() last_error = body.get("error_description") or body.get("msg") or response.text error_code = body.get("error_code") or "" except Exception: last_error = response.text if error_code == "otp_expired" or "expired" in last_error.lower(): LOGGER.warning("OTP rejected for %s with type %s: %s", identifier, current_type, last_error) if response.status_code not in {400, 403}: break except requests.RequestException as exc: LOGGER.error("OTP Request failed for type %s: %s", current_type, exc) last_error = str(exc) raise HTTPException(status_code=401, detail=f"OTP verification failed: {last_error}") def _supabase_upsert_profile(*, user_id: str, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: if not user_id: return None if not isinstance(payload, dict): payload = {} if "phone_number" not in payload and payload.get("phone"): payload["phone_number"] = payload.get("phone") allowed_keys = {"id", "first_name", "last_name", "full_name", "username", "phone_number", "email", "national_id", "role", "staff_id"} upsert_payload = {"id": user_id} for key in allowed_keys: if key == "id": continue value = payload.get(key) if value is None: continue if isinstance(value, str): value = _normalize_spaces(value) upsert_payload[key] = value if len(upsert_payload) == 1: from app.services.supabase_client import supabase_get_profile_by_user_id return supabase_get_profile_by_user_id(user_id) response = requests.post(f"{SUPABASE_URL}/rest/v1/profiles", json=upsert_payload, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json", prefer="resolution=merge-duplicates,return=representation"), timeout=SUPABASE_TIMEOUT_SECONDS) if response.status_code not in {200, 201}: raise HTTPException(status_code=502, detail=f"Failed to upsert profile ({response.status_code}): {_missing_schema_hint(response.text)}") rows = response.json() if response.text else [] if isinstance(rows, list) and rows and isinstance(rows[0], dict): return rows[0] from app.services.supabase_client import supabase_get_profile_by_user_id return supabase_get_profile_by_user_id(user_id)