""" MetadataService — Modular client tracking service. Extracts and enriches client metadata from HTTP requests for fraud detection & analytics. Components: 1. parse_user_agent() — Browser / OS / Device detection from User-Agent string 2. get_geoip_info() — Async GeoIP lookup via ip-api.com (free tier) 3. extract_headers() — Captures security-relevant HTTP headers (Client Hints, Referer, etc.) 4. build_metadata() — Merges frontend browser_meta with server-side headers into one JSON blob """ import json from typing import Optional, Dict, Any import httpx from fastapi import Request # ─── 1. User-Agent Parsing ──────────────────────────────────────────────────── def parse_user_agent(ua: Optional[str]) -> Dict[str, str]: """ Parse a User-Agent string into browser, OS, and device type. Returns {"browser": ..., "os": ..., "device_type": ...} """ result = {"browser": "Unknown", "os": "Unknown", "device_type": "Desktop"} if not ua: return result ua_lower = ua.lower() # ── Device ── if "mobi" in ua_lower: result["device_type"] = "Mobile" elif "tablet" in ua_lower or "ipad" in ua_lower: result["device_type"] = "Tablet" # ── Browser (order matters: Edge contains "chrome", so check Edge first) ── if "edg" in ua_lower: result["browser"] = "Edge" elif "opr" in ua_lower or "opera" in ua_lower: result["browser"] = "Opera" elif "chrome" in ua_lower and "chromium" not in ua_lower: result["browser"] = "Chrome" elif "firefox" in ua_lower: result["browser"] = "Firefox" elif "safari" in ua_lower: result["browser"] = "Safari" # ── OS (order matters: iPhone/iPad UAs contain "Mac OS X", so check them first) ── if "iphone" in ua_lower: result["os"] = "iOS" result["device_type"] = "Mobile" elif "ipad" in ua_lower: result["os"] = "iPadOS" result["device_type"] = "Tablet" elif "android" in ua_lower: result["os"] = "Android" elif "windows" in ua_lower: result["os"] = "Windows" elif "mac os" in ua_lower or "macintosh" in ua_lower: result["os"] = "MacOS" elif "cros" in ua_lower: result["os"] = "ChromeOS" elif "linux" in ua_lower: result["os"] = "Linux" return result # ─── 2. GeoIP Lookup ────────────────────────────────────────────────────────── async def get_geoip_info(ip: Optional[str]) -> Optional[str]: """ Look up IP geolocation via ip-api.com. Returns JSON string or None. Skips localhost / private-range IPs. """ if not ip or ip in ("127.0.0.1", "localhost", "::1", "0.0.0.0"): return None # Skip common private ranges if ip and ip.startswith(("10.", "172.16.", "192.168.")): return None try: async with httpx.AsyncClient(timeout=3.0) as client: resp = await client.get(f"http://ip-api.com/json/{ip}?fields=66846719") if resp.status_code == 200: data = resp.json() if data.get("status") == "success": return json.dumps({ "country": data.get("country"), "countryCode": data.get("countryCode"), "region": data.get("regionName"), "city": data.get("city"), "zip": data.get("zip"), "lat": data.get("lat"), "lon": data.get("lon"), "timezone": data.get("timezone"), "isp": data.get("isp"), "org": data.get("org"), "as": data.get("as"), "mobile": data.get("mobile"), "proxy": data.get("proxy"), "hosting": data.get("hosting"), }) except Exception: pass # Non-critical — order proceeds without geo data return None # ─── 3. Header Extraction ───────────────────────────────────────────────────── def extract_headers(request: Request) -> Dict[str, Any]: """ Pull security/analytics-relevant headers from the incoming HTTP request. """ h = request.headers return { "referer": h.get("Referer"), "accept_language": h.get("Accept-Language"), "sec_ch_ua": h.get("Sec-CH-UA"), "sec_ch_ua_mobile": h.get("Sec-CH-UA-Mobile"), "sec_ch_ua_platform": h.get("Sec-CH-UA-Platform"), "sec_fetch_site": h.get("Sec-Fetch-Site"), "x_forwarded_for": h.get("X-Forwarded-For"), } # ─── 4. Build Merged Metadata ───────────────────────────────────────────────── def build_metadata(frontend_json: Optional[str], request: Request) -> str: """ Merge frontend-supplied browser metadata with server-extracted headers into a single JSON string for storage. """ browser_meta = {} if frontend_json: try: browser_meta = json.loads(frontend_json) except (json.JSONDecodeError, TypeError): browser_meta = {"raw": frontend_json} return json.dumps({ "browser_meta": browser_meta, "headers": extract_headers(request), })