"""Vehicle Map static server and server-side proxy for the vehicle open platform.""" from concurrent.futures import ThreadPoolExecutor import base64 import binascii import hashlib import hmac from http.server import HTTPServer, SimpleHTTPRequestHandler import json import os from pathlib import Path import threading import time from socketserver import ThreadingMixIn from urllib.error import HTTPError, URLError from urllib.parse import urlparse from urllib.request import Request, urlopen ROOT = Path(__file__).resolve().parent def _static_asset_version(): """Content fingerprint used to bypass upstream static-asset caches on release.""" digest = hashlib.sha256() for filename in ("app.js", "styles.css"): digest.update((ROOT / filename).read_bytes()) return digest.hexdigest()[:16] STATIC_ASSET_VERSION = _static_asset_version() HOST = os.getenv("VEHICLE_MAP_HOST", "0.0.0.0") PORT = int(os.getenv("VEHICLE_MAP_PORT", "20800")) OPEN_PLATFORM_BASE_URL = os.getenv("OPEN_PLATFORM_BASE_URL", "https://open.d.lnoneos.com").rstrip("/") OPEN_PLATFORM_APP_KEY = os.getenv("OPEN_PLATFORM_APP_KEY", "").strip() UPSTREAM_TIMEOUT_SECONDS = float(os.getenv("UPSTREAM_TIMEOUT_SECONDS", "15")) DASHBOARD_CACHE_SECONDS = int(os.getenv("DASHBOARD_CACHE_SECONDS", "5")) STATION_CACHE_SECONDS = int(os.getenv("STATION_CACHE_SECONDS", "3600")) AUTH_PROVIDER = os.getenv("VEHICLE_MAP_AUTH_PROVIDER", "local").strip().lower() LOCAL_ACCESS_CODE = os.getenv("VEHICLE_MAP_ACCESS_CODE", "").strip() SESSION_SECRET = os.getenv("VEHICLE_MAP_SESSION_SECRET", "").encode("utf-8") SESSION_TTL_SECONDS = int(os.getenv("VEHICLE_MAP_SESSION_TTL_SECONDS", "1800")) COOKIE_SECURE = os.getenv("VEHICLE_MAP_COOKIE_SECURE", "true").strip().lower() not in {"0", "false", "no"} AUTH_CENTER_INTROSPECTION_URL = os.getenv("VEHICLE_MAP_AUTH_CENTER_INTROSPECTION_URL", "").strip() AUTH_CENTER_CLIENT_TOKEN = os.getenv("VEHICLE_MAP_AUTH_CENTER_CLIENT_TOKEN", "").strip() OPERATIONS_READ_SCOPE = "operations:read" SESSION_COOKIE_NAME = "ln_map_session" _cache_lock = threading.Lock() _cache = {} class AuthenticationError(RuntimeError): """The request does not carry a valid operations-read session.""" class AuthenticationUnavailable(RuntimeError): """The selected authentication provider is not configured.""" class AccessAuthorizer: """Small provider boundary: replace this class when the auth center contract is ready.""" def authorize_access_code(self, access_code): raise NotImplementedError class LocalAccessCodeAuthorizer(AccessAuthorizer): def authorize_access_code(self, access_code): if not LOCAL_ACCESS_CODE or not SESSION_SECRET: raise AuthenticationUnavailable("local access-code authentication is not configured") if not hmac.compare_digest(str(access_code or ""), LOCAL_ACCESS_CODE): raise AuthenticationError("invalid access code") return {"subject": "local-access-code", "scopes": [OPERATIONS_READ_SCOPE]} class AuthCenterAccessCodeAuthorizer(AccessAuthorizer): """Adapter for an auth-center access-code introspection endpoint. The endpoint contract is deliberately small: it receives accessCode and audience, and returns {active, subject, scopes}. No frontend change is needed when enabled. """ def authorize_access_code(self, access_code): if not AUTH_CENTER_INTROSPECTION_URL: raise AuthenticationUnavailable("auth-center introspection URL is not configured") headers = {"Content-Type": "application/json", "Accept": "application/json"} if AUTH_CENTER_CLIENT_TOKEN: headers["Authorization"] = "Bearer " + AUTH_CENTER_CLIENT_TOKEN request = Request( AUTH_CENTER_INTROSPECTION_URL, data=_json_bytes({"accessCode": str(access_code or ""), "audience": "vehicle-map"}), method="POST", headers=headers, ) try: with urlopen(request, timeout=UPSTREAM_TIMEOUT_SECONDS) as response: payload = json.load(response) except (HTTPError, URLError, ValueError) as exc: raise AuthenticationUnavailable("auth-center is unavailable") from exc scopes = payload.get("scopes") or [] if not payload.get("active") or OPERATIONS_READ_SCOPE not in scopes: raise AuthenticationError("access code does not grant operations read") return {"subject": str(payload.get("subject") or "auth-center"), "scopes": scopes} def _authorizer(): if AUTH_PROVIDER == "local": return LocalAccessCodeAuthorizer() if AUTH_PROVIDER in {"auth-center", "auth_center"}: return AuthCenterAccessCodeAuthorizer() raise AuthenticationUnavailable("unknown authentication provider") def _b64encode(value): return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=") def _b64decode(value): padded = str(value) + "=" * (-len(str(value)) % 4) return base64.urlsafe_b64decode(padded.encode("ascii")) def _create_session(principal): if not SESSION_SECRET: raise AuthenticationUnavailable("session signing secret is not configured") payload = { # Incrementing the session format invalidates earlier long-lived sessions. "v": 2, "sub": principal["subject"], "scopes": principal.get("scopes") or [], "exp": int(time.time()) + max(300, SESSION_TTL_SECONDS), } encoded = _b64encode(_json_bytes(payload)) signature = _b64encode(hmac.new(SESSION_SECRET, encoded.encode("ascii"), hashlib.sha256).digest()) return encoded + "." + signature def _read_session(cookie_header): cookies = {} for item in str(cookie_header or "").split(";"): if "=" in item: key, value = item.strip().split("=", 1) cookies[key] = value token = cookies.get(SESSION_COOKIE_NAME, "") try: encoded, signature = token.split(".", 1) expected = _b64encode(hmac.new(SESSION_SECRET, encoded.encode("ascii"), hashlib.sha256).digest()) if not SESSION_SECRET or not hmac.compare_digest(signature, expected): return None payload = json.loads(_b64decode(encoded)) if payload.get("v") != 2 or int(payload.get("exp", 0)) <= int(time.time()): return None if OPERATIONS_READ_SCOPE not in (payload.get("scopes") or []): return None return {"subject": str(payload.get("sub") or ""), "scopes": payload["scopes"]} except (ValueError, TypeError, UnicodeDecodeError, binascii.Error, json.JSONDecodeError): return None def _session_cookie(value, max_age): secure = "; Secure" if COOKIE_SECURE else "" return f"{SESSION_COOKIE_NAME}={value}; Path=/; Max-Age={max_age}; HttpOnly; SameSite=Lax{secure}" class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True allow_reuse_address = True def _json_bytes(value): return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8") def _post_open_platform(path, body): if not OPEN_PLATFORM_APP_KEY: raise RuntimeError("OPEN_PLATFORM_APP_KEY is not configured") request = Request( OPEN_PLATFORM_BASE_URL + path, data=_json_bytes(body), method="POST", headers={ "Authorization": "Bearer " + OPEN_PLATFORM_APP_KEY, "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "lingniu-vehicle-map/1.0", }, ) try: with urlopen(request, timeout=UPSTREAM_TIMEOUT_SECONDS) as response: payload = json.load(response) except HTTPError as exc: detail = exc.read(2048).decode("utf-8", errors="replace") raise RuntimeError(f"open platform returned HTTP {exc.code}: {detail}") from exc except URLError as exc: raise RuntimeError(f"open platform unavailable: {exc.reason}") from exc if payload.get("code") != "SUCCESS": raise RuntimeError(f"open platform rejected request: {payload.get('code')} {payload.get('message')}") return payload.get("data") or [] def _cached(key, ttl_seconds, loader): now = time.time() with _cache_lock: cached = _cache.get(key) if cached and now - cached[0] < ttl_seconds: return cached[1] value = loader() with _cache_lock: _cache[key] = (now, value) return value def _load_stations(): return _cached( "stations", STATION_CACHE_SECONDS, lambda: _post_open_platform("/api/v1/hydrogen-stations/query", {}), ) def _public_station(station): """Public station directory contract. Keep this a positive field allowlist.""" fields = ( "id", "name", "shortName", "province", "city", "district", "address", "longitude", "latitude", "cooperative", ) return {field: station.get(field) for field in fields if station.get(field) is not None} def _load_public_stations(): stations = _load_stations() return { "status": "ok", "access": {"level": "public", "stations": "directory", "vehicles": "restricted", "hydrogen": "restricted"}, "asOf": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), "summary": { "totalStations": len(stations), "cooperativeStations": sum(1 for station in stations if station.get("cooperative")), }, "stations": [_public_station(station) for station in stations], } def _load_dashboard(): today = time.strftime("%Y-%m-%d", time.localtime()) with ThreadPoolExecutor(max_workers=3) as executor: realtime_future = executor.submit( _post_open_platform, "/api/v1/vehicles/realtime/query", {} ) mileage_future = executor.submit( _post_open_platform, "/api/v1/vehicles/mileage/query", {"date": today} ) stations_future = executor.submit(_load_stations) vehicles = realtime_future.result() mileage_rows = mileage_future.result() stations = stations_future.result() mileage_by_vin = { row.get("vin"): row for row in mileage_rows if row.get("vin") and row.get("status") == "NORMAL" } # The operations card is a daily activity view: every vehicle that reported # today is placed in exactly one bucket using its latest selected speed. active_vehicles = [item for item in vehicles if item.get("activeToday")] driving = sum(1 for item in active_vehicles if float(item.get("speedKmh") or 0) > 3) idle = len(active_vehicles) - driving offline = sum(1 for item in vehicles if item.get("motionStatus") == "offline") active_today = len(active_vehicles) daily_mileage = round( sum(float(row.get("dailyMileageKm") or 0) for row in mileage_rows), 3 ) monthly_hydrogen = round( sum(float(station.get("monthlyHydrogenKg") or 0) for station in stations), 3 ) for vehicle in vehicles: mileage = mileage_by_vin.get(vehicle.get("vin"), {}) vehicle["dailyMileageKm"] = mileage.get("dailyMileageKm", 0) return { "status": "ok", "asOf": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), "date": today, "summary": { "totalVehicles": len(vehicles), "onlineVehicles": active_today, "drivingVehicles": driving, "idleVehicles": idle, "offlineVehicles": offline, "todayMileageKm": daily_mileage, "monthlyHydrogenKg": monthly_hydrogen, "totalStations": len(stations), "cooperativeStations": sum(1 for item in stations if item.get("cooperative")), }, "vehicles": vehicles, "stations": stations, } class VehicleMapHandler(SimpleHTTPRequestHandler): server_version = "LingniuVehicleMap/1.0" def do_GET(self): path = urlparse(self.path).path if path in {"/", "/index.html"}: self._serve_index() return if path == "/api/health": self._write_json(200, { "status": "ok", "service": "vehicle-map", "openPlatformConfigured": bool(OPEN_PLATFORM_APP_KEY), "authProvider": AUTH_PROVIDER, }) return if path == "/api/public/stations": try: payload = _cached("public-stations", DASHBOARD_CACHE_SECONDS, _load_public_stations) self._write_json(200, payload) except Exception as exc: # keep upstream details server-side only self.log_error("public station refresh failed: %s", exc) self._write_json(502, { "status": "error", "code": "UPSTREAM_UNAVAILABLE", "message": "加氢站目录暂时不可用，请稍后重试", }) return if path == "/api/session": principal = _read_session(self.headers.get("Cookie")) self._write_json(200, { "status": "ok", "authorized": bool(principal), "principal": principal, }) return if path == "/api/dashboard": if not self._require_operations_access(): return try: payload = _cached("dashboard", DASHBOARD_CACHE_SECONDS, _load_dashboard) self._write_json(200, payload) except Exception as exc: # keep upstream details server-side only self.log_error("dashboard refresh failed: %s", exc) self._write_json(502, { "status": "error", "code": "UPSTREAM_UNAVAILABLE", "message": "运营数据暂时不可用，请稍后重试", }) return super().do_GET() def do_POST(self): path = urlparse(self.path).path if path != "/api/session": self._write_json(404, {"status": "error", "code": "NOT_FOUND"}) return try: content_length = int(self.headers.get("Content-Length", "0")) if content_length < 1 or content_length > 4096: raise ValueError("invalid request length") payload = json.loads(self.rfile.read(content_length)) principal = _authorizer().authorize_access_code(payload.get("accessCode")) session = _create_session(principal) except AuthenticationError: self._write_json(401, {"status": "error", "code": "ACCESS_DENIED", "message": "访问码无效"}) return except AuthenticationUnavailable: self._write_json(503, {"status": "error", "code": "AUTH_UNAVAILABLE", "message": "授权服务暂不可用"}) return except (ValueError, TypeError, json.JSONDecodeError): self._write_json(400, {"status": "error", "code": "INVALID_REQUEST", "message": "访问码格式不正确"}) return self._write_json(200, { "status": "ok", "authorized": True, "principal": principal, }, headers={"Set-Cookie": _session_cookie(session, max(300, SESSION_TTL_SECONDS))}) def do_DELETE(self): if urlparse(self.path).path != "/api/session": self._write_json(404, {"status": "error", "code": "NOT_FOUND"}) return self._write_json(200, {"status": "ok", "authorized": False}, headers={"Set-Cookie": _session_cookie("", 0)}) def _require_operations_access(self): if _read_session(self.headers.get("Cookie")): return True self._write_json(401, { "status": "error", "code": "AUTH_REQUIRED", "message": "运营数据需要授权访问", }) return False def _serve_index(self): try: template = (ROOT / "index.html").read_text(encoding="utf-8") body = template.replace("__ASSET_VERSION__", STATIC_ASSET_VERSION).encode("utf-8") except OSError as exc: self.log_error("index template unavailable: %s", exc) self.send_error(500, "Map application is temporarily unavailable") return self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(body) def end_headers(self): if self.path == "/" or self.path.split("?", 1)[0].endswith((".html", ".css", ".js", ".svg")): self.send_header("Cache-Control", "no-cache") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Referrer-Policy", "strict-origin-when-cross-origin") self.send_header("X-Frame-Options", "SAMEORIGIN") super().end_headers() def _write_json(self, status, payload, headers=None): body = _json_bytes(payload) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) for name, value in (headers or {}).items(): self.send_header(name, value) self.end_headers() self.wfile.write(body) if __name__ == "__main__": server = ThreadingHTTPServer((HOST, PORT), VehicleMapHandler) print(f"Lingniu Vehicle Map listening on http://{HOST}:{PORT}") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close()