""" RouteCheck — local routing web service with image captcha. Endpoints: GET / — web UI GET /captcha/image/{id} — PNG captcha image POST /api/captcha/new — create captcha, return {id} POST /api/captcha/solve — {id, answer} → {token} or 400 GET /api/route — ?from=lat,lon&to=lat,lon&token=... token = solved captcha token OR INTERNAL_TOKEN env var """ import io import math import os import random import string import time import uuid from typing import Optional import httpx from fastapi import FastAPI, HTTPException, Query from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse from PIL import Image, ImageDraw, ImageFilter, ImageFont from pydantic import BaseModel app = FastAPI(title="RouteCheck") # ── Config ───────────────────────────────────────────────────────────────────── YANDEX_KEY = os.getenv("YANDEX_ROUTING_KEY", "") INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "") HTTPS_PROXY = os.getenv("HTTPS_PROXY", "") CAPTCHA_TTL = 300 # seconds a captcha is valid TOKEN_TTL = 3600 # seconds a solved token is valid # ── In-memory captcha store ──────────────────────────────────────────────────── _captchas: dict[str, dict] = {} # id → {answer, token, expires} _tokens: dict[str, float] = {} # token → expires def _purge(): now = time.time() for k in list(_captchas.keys()): if _captchas[k]["expires"] < now: del _captchas[k] for k in list(_tokens.keys()): if _tokens[k] < now: del _tokens[k] # ── Captcha image generation ─────────────────────────────────────────────────── def _rand_color(dark=False): if dark: return tuple(random.randint(0, 100) for _ in range(3)) return tuple(random.randint(140, 255) for _ in range(3)) def _make_captcha_image(text: str) -> bytes: W, H = 220, 80 img = Image.new("RGB", (W, H), color=_rand_color()) draw = ImageDraw.Draw(img) # Background noise: random lines for _ in range(8): x1, y1 = random.randint(0, W), random.randint(0, H) x2, y2 = random.randint(0, W), random.randint(0, H) draw.line([(x1, y1), (x2, y2)], fill=_rand_color(dark=True), width=2) # Background noise: random dots for _ in range(300): x, y = random.randint(0, W), random.randint(0, H) draw.point((x, y), fill=_rand_color(dark=True)) # Draw each character with slight random offset and rotation try: font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 36) except Exception: font = ImageFont.load_default() char_w = W // (len(text) + 2) for i, ch in enumerate(text): x = char_w + i * char_w + random.randint(-4, 4) y = (H - 40) // 2 + random.randint(-6, 6) # Draw shadow draw.text((x + 2, y + 2), ch, font=font, fill=_rand_color(dark=True)) draw.text((x, y), ch, font=font, fill=_rand_color(dark=True)) # Wavy distortion via pixel manipulation pixels = img.load() for x in range(W): shift = int(4 * math.sin(x / 15.0)) col = [pixels[x, y] for y in range(H)] for y in range(H): pixels[x, y] = col[(y - shift) % H] img = img.filter(ImageFilter.SMOOTH) buf = io.BytesIO() img.save(buf, format="PNG") return buf.getvalue() def _generate_problem() -> tuple[str, int]: """Return (display_text, answer).""" ops = [ lambda a, b: (f"{a} + {b} = ?", a + b), lambda a, b: (f"{a} × {b} = ?", a * b), lambda a, b: (f"{max(a,b)} − {min(a,b)} = ?", max(a, b) - min(a, b)), ] op = random.choice(ops) a, b = random.randint(2, 9), random.randint(2, 9) text, answer = op(a, b) return text, answer # ── Routes ───────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def index(): return HTML_PAGE @app.get("/captcha/image/{captcha_id}") async def captcha_image(captcha_id: str): _purge() entry = _captchas.get(captcha_id) if not entry: raise HTTPException(404, "Captcha not found or expired") png = _make_captcha_image(entry["problem"]) return StreamingResponse(io.BytesIO(png), media_type="image/png", headers={"Cache-Control": "no-store"}) class CaptchaNewResponse(BaseModel): id: str @app.post("/api/captcha/new") async def captcha_new(): _purge() problem_text, answer = _generate_problem() cid = str(uuid.uuid4()) _captchas[cid] = { "problem": problem_text, "answer": answer, "expires": time.time() + CAPTCHA_TTL, } return {"id": cid} class SolveRequest(BaseModel): id: str answer: int @app.post("/api/captcha/solve") async def captcha_solve(req: SolveRequest): _purge() entry = _captchas.get(req.id) if not entry: raise HTTPException(400, "Captcha expired or not found") if entry["answer"] != req.answer: raise HTTPException(400, "Wrong answer") token = str(uuid.uuid4()) _tokens[token] = time.time() + TOKEN_TTL del _captchas[req.id] return {"token": token} @app.get("/api/route") async def route( from_coords: str = Query(..., alias="from", description="lat,lon"), to_coords: str = Query(..., alias="to", description="lat,lon"), token: str = Query(...), ): _purge() # Auth: internal service token or valid captcha token if token != INTERNAL_TOKEN: if token not in _tokens: raise HTTPException(401, "Invalid or expired token — solve captcha first") if not YANDEX_KEY: raise HTTPException(503, "YANDEX_ROUTING_KEY not configured") # Parse coords try: from_lat, from_lon = map(float, from_coords.split(",")) to_lat, to_lon = map(float, to_coords.split(",")) except ValueError: raise HTTPException(400, "coords must be lat,lon") # Yandex Routing API expects lon,lat order waypoints = f"{from_lon},{from_lat}|{to_lon},{to_lat}" transport = httpx.AsyncHTTPTransport(proxy=HTTPS_PROXY) if HTTPS_PROXY else None async with httpx.AsyncClient(timeout=15, transport=transport) as client: try: r = await client.get( "https://api.routing.yandex.net/v2/route", params={"apikey": YANDEX_KEY, "waypoints": waypoints, "mode": "driving"}, ) except Exception as e: raise HTTPException(502, f"Yandex API unreachable: {e}") if r.status_code != 200: raise HTTPException(502, f"Yandex API error {r.status_code}: {r.text[:200]}") data = r.json() try: leg = data["route"]["legs"][0] duration_s = leg["duration"] duration_traffic_s = leg.get("duration_in_traffic", duration_s) distance_m = leg["distance"] except (KeyError, IndexError) as e: raise HTTPException(502, f"Unexpected Yandex response: {e} — {str(data)[:200]}") return { "duration_min": round(duration_s / 60), "duration_traffic_min": round(duration_traffic_s / 60), "distance_km": round(distance_m / 1000, 1), } # ── HTML ─────────────────────────────────────────────────────────────────────── HTML_PAGE = """ RouteCheck

RouteCheck

Real-time driving time with Yandex traffic data

captcha
↻ New challenge
Wrong answer, try again.
with current traffic
without traffic
distance
"""