+
+
+
diff --git a/vehicle-map/server.py b/vehicle-map/server.py
index 407e848b..d248811f 100644
--- a/vehicle-map/server.py
+++ b/vehicle-map/server.py
@@ -1,6 +1,10 @@
"""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
@@ -9,6 +13,7 @@ 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
@@ -20,11 +25,134 @@ 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
@@ -73,6 +201,37 @@ def _cached(key, ttl_seconds, loader):
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:
@@ -82,12 +241,7 @@ def _load_dashboard():
mileage_future = executor.submit(
_post_open_platform, "/api/v1/vehicles/mileage/query", {"date": today}
)
- stations_future = executor.submit(
- _cached,
- "stations",
- STATION_CACHE_SECONDS,
- lambda: _post_open_platform("/api/v1/hydrogen-stations/query", {}),
- )
+ stations_future = executor.submit(_load_stations)
vehicles = realtime_future.result()
mileage_rows = mileage_future.result()
stations = stations_future.result()
@@ -138,14 +292,38 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
server_version = "LingniuVehicleMap/1.0"
def do_GET(self):
- if self.path == "/api/health":
+ path = urlparse(self.path).path
+ if path == "/api/health":
self._write_json(200, {
"status": "ok",
"service": "vehicle-map",
"openPlatformConfigured": bool(OPEN_PLATFORM_APP_KEY),
+ "authProvider": AUTH_PROVIDER,
})
return
- if self.path == "/api/dashboard":
+ 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)
@@ -154,11 +332,54 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
self._write_json(502, {
"status": "error",
"code": "UPSTREAM_UNAVAILABLE",
- "message": "车辆数据暂时不可用，请稍后重试",
+ "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 end_headers(self):
if self.path == "/" or self.path.split("?", 1)[0].endswith((".html", ".css", ".js", ".svg")):
self.send_header("Cache-Control", "no-cache")
@@ -167,12 +388,14 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
self.send_header("X-Frame-Options", "SAMEORIGIN")
super().end_headers()
- def _write_json(self, status, payload):
+ 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)
diff --git a/vehicle-map/styles.css b/vehicle-map/styles.css
index 1c156372..34d92650 100644
--- a/vehicle-map/styles.css
+++ b/vehicle-map/styles.css
@@ -290,6 +290,80 @@ body {
box-shadow: 0 2px 6px rgba(0, 0, 0, 0.1);
}
+.segment-btn.is-locked::after {
+ content: '⌁';
+ margin-left: 1px;
+ color: var(--text-sub);
+ font-size: 10px;
+}
+
+.access-trigger {
+ min-height: 28px;
+ padding: 0 10px;
+ border: 1px solid rgba(0, 113, 67, 0.18);
+ border-radius: 8px;
+ background: rgba(0, 113, 67, 0.07);
+ color: var(--accent-primary);
+ font-size: 10px;
+ font-weight: 700;
+ cursor: pointer;
+ transition: background 0.18s ease, border-color 0.18s ease, transform 0.18s ease;
+}
+
+.access-trigger:hover { border-color: rgba(0, 113, 67, 0.34); background: rgba(0, 113, 67, 0.12); transform: translateY(-1px); }
+.access-trigger.is-authorized { border-color: rgba(2, 132, 199, 0.25); background: rgba(2, 132, 199, 0.09); color: var(--accent-blue); }
+
+.access-dialog {
+ position: fixed;
+ inset: 0;
+ margin: auto;
+ width: min(390px, calc(100vw - 32px));
+ max-height: calc(100dvh - 32px);
+ padding: 0;
+ border: 0;
+ border-radius: 20px;
+ background: transparent;
+ color: var(--text-main);
+ box-shadow: 0 24px 80px rgba(15, 23, 42, 0.3);
+}
+
+.access-dialog::backdrop { background: rgba(15, 23, 42, 0.42); backdrop-filter: blur(5px); }
+
+.access-dialog-card {
+ display: grid;
+ gap: 16px;
+ padding: 22px;
+ border: 1px solid var(--glass-border);
+ border-radius: inherit;
+ background: color-mix(in srgb, var(--glass-bg) 98%, white);
+}
+
+.access-dialog-icon {
+ display: grid;
+ width: 36px;
+ height: 36px;
+ place-items: center;
+ border-radius: 12px;
+ background: rgba(0, 113, 67, 0.1);
+ color: var(--accent-primary);
+ font-size: 21px;
+ font-weight: 800;
+}
+
+.access-dialog-heading h2 { margin: 0; font-size: 18px; letter-spacing: -0.02em; }
+.access-dialog-heading p { margin: 6px 0 0; color: var(--text-muted); font-size: 12px; line-height: 1.6; }
+.access-session-note { margin: -7px 0 0; padding: 9px 10px; border-radius: 10px; background: rgba(2, 132, 199, 0.08); color: #0369a1; font-size: 11px; line-height: 1.55; }
+
+.access-code-field { display: grid; gap: 7px; color: var(--text-muted); font-size: 11px; font-weight: 700; }
+.access-code-field input { width: 100%; height: 42px; padding: 0 12px; border: 1px solid var(--glass-border); border-radius: 10px; outline: 0; background: var(--pill-bg); color: var(--text-main); font: 13px/1.2 'JetBrains Mono', monospace; box-sizing: border-box; }
+.access-code-field input:focus { border-color: rgba(2, 132, 199, 0.5); box-shadow: 0 0 0 3px rgba(2, 132, 199, 0.1); }
+.access-dialog-error { margin: -6px 0 0; color: #dc2626; font-size: 11px; }
+.access-dialog-actions { display: flex; justify-content: flex-end; gap: 8px; }
+.access-cancel-btn, .access-submit-btn { min-height: 36px; padding: 0 13px; border-radius: 9px; font-size: 12px; font-weight: 700; cursor: pointer; }
+.access-cancel-btn { border: 1px solid var(--glass-border); background: transparent; color: var(--text-muted); }
+.access-submit-btn { border: 1px solid var(--accent-primary); background: var(--accent-primary); color: white; }
+.access-submit-btn:disabled { opacity: 0.58; cursor: wait; }
+
.bw-svg {
stroke: currentColor;
}
@@ -766,6 +840,7 @@ body {
}
.detail-title-wrap { min-width: 0; }
+.detail-card-actions { display: flex; flex: 0 0 auto; align-items: center; gap: 5px; }
.detail-type-pill {
display: inline-flex;
@@ -818,6 +893,24 @@ body {
cursor: pointer;
}
+.detail-navigate-btn {
+ display: grid;
+ width: 26px;
+ height: 26px;
+ flex: 0 0 26px;
+ place-items: center;
+ border: 1px solid rgba(0, 113, 67, 0.16);
+ border-radius: 50%;
+ background: rgba(0, 113, 67, 0.09);
+ color: var(--accent-primary);
+ cursor: pointer;
+ transition: transform 0.18s ease, background 0.18s ease, border-color 0.18s ease;
+}
+
+.detail-navigate-btn:hover { transform: translateY(-1px); border-color: rgba(0, 113, 67, 0.34); background: rgba(0, 113, 67, 0.15); }
+.detail-navigate-btn:disabled { cursor: default; opacity: 0.45; }
+.detail-navigate-btn svg { width: 13px; height: 13px; fill: none; stroke: currentColor; stroke-linecap: round; stroke-linejoin: round; stroke-width: 1.9; }
+
.detail-grid {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
@@ -1369,11 +1462,16 @@ body {
}
.segment-btn,
- .bw-btn {
+ .bw-btn,
+ .access-trigger {
padding: 4px 8px;
font-size: 10px;
}
+ .access-trigger { min-height: 26px; }
+
+ .access-dialog-card { padding: 18px; }
+
.time-widget {
display: none;
}
diff --git a/vehicle-map/tests/test_app.mjs b/vehicle-map/tests/test_app.mjs
index 7c587f31..b2ccf73c 100644
--- a/vehicle-map/tests/test_app.mjs
+++ b/vehicle-map/tests/test_app.mjs
@@ -80,6 +80,29 @@ assert.equal(points.length, 2);
assert.ok(points.every(node => node.kind === 'vehiclePoint'));
assert.equal(points.find(node => node.nameZh === '粤A00001').online, 1);
+// A selected district is a semantic filter, not a viewport filter. Its full
+// fleet must survive a camera fit even when a valid vehicle sits outside the
+// initial district-centroid viewport.
+currentMode = 'vehicle';
+dashboard = {
+ vehicles: [
+ { vin: 'VIN-BY-1', plateNumber: '粤A10001', activeToday: true, locationAvailable: true, longitude: 113.22, latitude: 23.18 },
+ { vin: 'VIN-BY-2', plateNumber: '粤A10002', activeToday: true, locationAvailable: true, longitude: 113.41, latitude: 23.38 }
+ ]
+};
+vehicleFilterGroups.districts = [{ adcode: '440111', vehicles: dashboard.vehicles }];
+filterState.district = '440111';
+map = {
+ getZoom: () => 11.6,
+ getBounds: () => ({
+ getSouthWest: () => ({ getLng: () => 113.1, getLat: () => 23.1 }),
+ getNorthEast: () => ({ getLng: () => 113.3, getLat: () => 23.3 })
+ })
+};
+assert.equal(hierarchyLevelForZoom(), 'vehicle');
+assert.equal(vehiclePointNodes().length, 2);
+filterState.district = '';
+
dashboard = {
stations: [
{ id: 'GD-1', name: '广州合作站', province: '广东省', city: '广州市', address: '广东省广州市黄埔区开源大道1号', longitude: 113.2, latitude: 23.1, cooperative: true, totalHydrogenKg: 20 },
@@ -110,6 +133,11 @@ map = {
assert.equal(stationNodes(false).length, 3);
assert.equal(stationNodes(true).length, 2);
assert.deepEqual(stationNodes(true).map(node => node.name).sort(), ['佛山外部站', '广州合作站']);
+const stationNavigationUrl = navigationUrlForEntity('station', dashboard.stations[0]);
+assert.ok(stationNavigationUrl.startsWith('https://uri.amap.com/navigation?to=113.200000,23.100000,'));
+assert.match(stationNavigationUrl, /%E5%B9%BF%E5%B7%9E/);
+assert.match(stationNavigationUrl, /coordinate=gaode&callnative=1$/);
+assert.equal(navigationUrlForEntity('vehicle', { plateNumber: '无位置车辆' }), null);
const stationView = stationViewportSummary();
assert.equal(stationView.level, 'station');
assert.equal(stationView.visibleNodes.length, 2);
diff --git a/vehicle-map/tests/test_server.py b/vehicle-map/tests/test_server.py
index abc70e38..3661bb50 100644
--- a/vehicle-map/tests/test_server.py
+++ b/vehicle-map/tests/test_server.py
@@ -1,5 +1,9 @@
import importlib.util
+import hashlib
+import hmac
+import json
from pathlib import Path
+import time
import unittest
from unittest.mock import patch
@@ -41,6 +45,49 @@ class DashboardTest(unittest.TestCase):
self.assertEqual(result["vehicles"][0]["dailyMileageKm"], 12.345)
self.assertEqual(result["vehicles"][2]["dailyMileageKm"], 0)
+ def test_public_station_directory_uses_an_allowlist(self):
+ stations = [{
+ "id": "GD-1", "name": "广州合作站", "shortName": "合作站", "province": "广东省",
+ "city": "广州市", "district": "黄埔区", "address": "开源大道1号", "longitude": 113.2,
+ "latitude": 23.1, "cooperative": True, "monthlyHydrogenKg": 88.8,
+ "totalHydrogenKg": 999.9, "internalOwner": "must-not-leak",
+ }]
+ with patch.object(server, "_post_open_platform", return_value=stations):
+ with patch.object(server, "_cache", {}):
+ result = server._load_public_stations()
+
+ station = result["stations"][0]
+ self.assertEqual(result["access"]["level"], "public")
+ self.assertEqual(station["name"], "广州合作站")
+ self.assertTrue(station["cooperative"])
+ self.assertNotIn("monthlyHydrogenKg", station)
+ self.assertNotIn("totalHydrogenKg", station)
+ self.assertNotIn("internalOwner", station)
+
+ def test_signed_operations_session_requires_the_operations_scope(self):
+ with patch.object(server, "LOCAL_ACCESS_CODE", "LN-map-test-code"), patch.object(server, "SESSION_SECRET", b"test-session-secret"), patch.object(server, "SESSION_TTL_SECONDS", 600):
+ principal = server._authorizer().authorize_access_code("LN-map-test-code")
+ token = server._create_session(principal)
+ session = server._read_session(f"{server.SESSION_COOKIE_NAME}={token}")
+
+ self.assertEqual(session["subject"], "local-access-code")
+ self.assertIn(server.OPERATIONS_READ_SCOPE, session["scopes"])
+ self.assertIsNone(server._read_session(f"{server.SESSION_COOKIE_NAME}={token}tampered"))
+
+ legacy_payload = {"v": 1, "sub": "legacy", "scopes": [server.OPERATIONS_READ_SCOPE], "exp": 4_102_444_800}
+ legacy_encoded = server._b64encode(server._json_bytes(legacy_payload))
+ legacy_signature = server._b64encode(hmac.new(server.SESSION_SECRET, legacy_encoded.encode("ascii"), hashlib.sha256).digest())
+ self.assertIsNone(server._read_session(f"{server.SESSION_COOKIE_NAME}={legacy_encoded}.{legacy_signature}"))
+
+ def test_default_operations_session_expires_after_thirty_minutes(self):
+ self.assertEqual(server.SESSION_TTL_SECONDS, 1800)
+ with patch.object(server, "SESSION_SECRET", b"test-session-secret"), patch.object(server, "SESSION_TTL_SECONDS", 1800):
+ token = server._create_session({"subject": "test", "scopes": [server.OPERATIONS_READ_SCOPE]})
+ encoded, _ = token.split(".", 1)
+ payload = json.loads(server._b64decode(encoded))
+ self.assertGreaterEqual(payload["exp"], int(time.time()) + 1798)
+ self.assertLessEqual(payload["exp"], int(time.time()) + 1801)
+
if __name__ == "__main__":
unittest.main()