Spaces:
Sleeping
Sleeping
File size: 5,851 Bytes
b2be963 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """
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),
})
|