"""Server-side proxy to the Google Maps Geocoding API (INDL-49, INDL-49g). Security posture: - The `GOOGLE_MAPS_API_KEY` never leaves the backend; the frontend calls only our own proxy route. - `address` is validated as plain text (no URL schemes, IPs, `@`, or CRLF) BEFORE any outbound request, defeating SSRF (A10/API7/S-09). - `address` is passed as a `params` dict to httpx — never string-interpolated into the URL (A03/S-03). - Redirects are not followed; a 5s timeout is enforced. - Google's raw error bodies are never surfaced to the client. """ from __future__ import annotations import ipaddress import logging import re from typing import Optional import httpx from src.core.config import settings from src.core.exceptions import ( NotFoundError, ServiceUnavailableError, ValidationError, ) logger = logging.getLogger(__name__) _GOOGLE_URL = "https://maps.googleapis.com/maps/api/geocode/json" _TIMEOUT_SECONDS = 5.0 _MIN_QUERY_LEN = 3 _MAX_QUERY_LEN = 500 # Reject anything that looks like a URL/scheme or contains dangerous chars. _SCHEME_RE = re.compile(r"[a-zA-Z][a-zA-Z0-9+.\-]*://") _FORBIDDEN_CHARS = ("@", "\r", "\n", "\t", "\x00") _GEOCODE_CACHE_TTL = 3600 # seconds — dedupe identical repeat searches (API6) def _validate_address(address: str) -> str: if not isinstance(address, str): raise ValidationError("A search address is required") addr = address.strip() if len(addr) < _MIN_QUERY_LEN: raise ValidationError("Enter at least 3 characters to search") if len(addr) > _MAX_QUERY_LEN: raise ValidationError("Search address is too long") if _SCHEME_RE.search(addr) or any(ch in addr for ch in _FORBIDDEN_CHARS): raise ValidationError("Invalid characters in search address") # Reject bare IP addresses (SSRF to metadata / internal hosts). candidate = addr.split("/")[0].split(":")[0].strip() try: ipaddress.ip_address(candidate) raise ValidationError("Invalid characters in search address") except ValueError: pass # not an IP — good return addr def _normalize(address: str) -> str: return re.sub(r"\s+", " ", address.strip().lower()) class GeocodingService: def __init__(self, db=None): self.db = db @property def enabled(self) -> bool: return bool(settings.GOOGLE_MAPS_API_KEY) async def _call_google(self, address: str) -> dict: """Perform the outbound HTTP call. Isolated so tests can stub it and so the SSRF validation can be asserted to run first.""" async with httpx.AsyncClient( timeout=_TIMEOUT_SECONDS, follow_redirects=False ) as client: resp = await client.get( _GOOGLE_URL, params={"address": address, "key": settings.GOOGLE_MAPS_API_KEY}, ) resp.raise_for_status() if "application/json" not in resp.headers.get("content-type", ""): raise ServiceUnavailableError( "Address search is temporarily unavailable — you can still " "pan/zoom the map manually" ) return resp.json() async def _cache_get(self, norm: str) -> Optional[dict]: from src.database.session import get_redis client = await get_redis() if client is None: return None try: import json raw = await client.get(f"geocode:{norm}") return json.loads(raw) if raw else None except Exception: return None finally: try: await client.aclose() except Exception: pass async def _cache_set(self, norm: str, value: dict) -> None: from src.database.session import get_redis client = await get_redis() if client is None: return try: import json await client.set(f"geocode:{norm}", json.dumps(value), ex=_GEOCODE_CACHE_TTL) except Exception: pass finally: try: await client.aclose() except Exception: pass async def geocode(self, address: str) -> dict: """Resolve an address to {lat, lng, formatted_address}. Raises: 404 (no match / geocoding disabled), 422 (bad input), 503 (upstream unavailable).""" if not self.enabled: # Key not configured — behave as "not found" (frontend hides the box). raise NotFoundError("Address search is not configured") addr = _validate_address(address) # SSRF guard — runs before any network I/O norm = _normalize(addr) cached = await self._cache_get(norm) if cached is not None: return cached try: payload = await self._call_google(addr) except (httpx.TimeoutException, httpx.TransportError) as exc: logger.warning("geocode upstream error: %s", type(exc).__name__) raise ServiceUnavailableError( "Address search is temporarily unavailable — you can still " "pan/zoom the map manually" ) except httpx.HTTPStatusError as exc: logger.warning("geocode upstream HTTP %s", exc.response.status_code) raise ServiceUnavailableError( "Address search is temporarily unavailable — you can still " "pan/zoom the map manually" ) status = payload.get("status") if status == "ZERO_RESULTS": raise NotFoundError( "No matching address found — try a more specific search" ) if status != "OK": # OVER_QUERY_LIMIT / REQUEST_DENIED / INVALID_REQUEST / UNKNOWN_ERROR logger.warning("geocode non-OK status: %s", status) raise ServiceUnavailableError( "Address search is temporarily unavailable — you can still " "pan/zoom the map manually" ) results = payload.get("results") or [] if not isinstance(results, list) or not results: raise NotFoundError( "No matching address found — try a more specific search" ) first = results[0] try: loc = first["geometry"]["location"] lat = float(loc["lat"]) lng = float(loc["lng"]) formatted = str(first.get("formatted_address") or addr) except (KeyError, TypeError, ValueError): raise ServiceUnavailableError( "Address search is temporarily unavailable — you can still " "pan/zoom the map manually" ) # Validate the coordinates are sane finite values in range (API10). if not (-90.0 <= lat <= 90.0 and -180.0 <= lng <= 180.0): raise ServiceUnavailableError( "Address search is temporarily unavailable — you can still " "pan/zoom the map manually" ) result = {"lat": lat, "lng": lng, "formatted_address": formatted[:500]} await self._cache_set(norm, result) return result