@@ -235,26 +243,6 @@
-
-
diff --git a/vehicle-map/server.py b/vehicle-map/server.py
index 7df0d764..aa498558 100644
--- a/vehicle-map/server.py
+++ b/vehicle-map/server.py
@@ -1,10 +1,7 @@
"""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
@@ -36,134 +33,10 @@ 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
@@ -220,29 +93,6 @@ def _load_stations():
)
-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:
@@ -262,7 +112,7 @@ def _load_dashboard():
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
+ # The daily activity view places 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)
@@ -312,32 +162,9 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
"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)
@@ -346,54 +173,11 @@ 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 _serve_index(self):
try:
template = (ROOT / "index.html").read_text(encoding="utf-8")
diff --git a/vehicle-map/styles.css b/vehicle-map/styles.css
index 34d92650..a1f409aa 100644
--- a/vehicle-map/styles.css
+++ b/vehicle-map/styles.css
@@ -290,80 +290,6 @@ 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;
}
@@ -795,6 +721,55 @@ body {
height: 100%;
}
+.navigation-launch-overlay {
+ position: absolute;
+ z-index: 28;
+ inset: 0;
+ display: grid;
+ place-items: center;
+ padding: 22px;
+ background: rgba(15, 23, 42, 0.14);
+ backdrop-filter: blur(1.5px);
+ -webkit-backdrop-filter: blur(1.5px);
+ opacity: 0;
+ pointer-events: none;
+ transition: opacity 0.18s ease;
+}
+
+.navigation-launch-overlay[hidden] { display: none; }
+.navigation-launch-overlay.is-visible { opacity: 1; pointer-events: auto; }
+
+.navigation-launch-panel {
+ display: flex;
+ align-items: center;
+ gap: 11px;
+ max-width: min(280px, calc(100vw - 48px));
+ padding: 13px 15px;
+ border: 1px solid rgba(255, 255, 255, 0.64);
+ border-radius: 16px;
+ background: color-mix(in srgb, var(--glass-bg) 94%, white);
+ box-shadow: 0 18px 42px rgba(15, 23, 42, 0.2);
+ transform: translateY(8px) scale(0.97);
+ transition: transform 0.22s cubic-bezier(.2, .75, .25, 1);
+}
+
+.navigation-launch-overlay.is-visible .navigation-launch-panel { transform: translateY(0) scale(1); }
+.navigation-launch-panel strong, .navigation-launch-panel span { display: block; }
+.navigation-launch-panel strong { color: var(--text-main); font-size: 12px; line-height: 1.35; }
+.navigation-launch-panel div > span { margin-top: 2px; color: var(--text-muted); font-size: 10px; line-height: 1.45; }
+
+.navigation-launch-spinner {
+ width: 20px;
+ height: 20px;
+ flex: 0 0 20px;
+ border: 2px solid rgba(0, 113, 67, 0.18);
+ border-top-color: var(--accent-primary);
+ border-radius: 50%;
+ animation: navigation-spin 0.78s linear infinite;
+}
+
+@keyframes navigation-spin { to { transform: rotate(360deg); } }
+
.map-detail-card {
position: absolute;
left: 14px;
@@ -910,6 +885,17 @@ body {
.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-navigate-btn.is-launching { transform: scale(0.92); background: rgba(0, 113, 67, 0.17); }
+.detail-navigate-btn.is-launching svg { animation: navigation-pulse 0.7s ease-in-out infinite alternate; }
+
+@keyframes navigation-pulse { to { transform: scale(1.15); } }
+
+@media (prefers-reduced-motion: reduce) {
+ .navigation-launch-overlay,
+ .navigation-launch-panel { transition: none; }
+ .navigation-launch-spinner,
+ .detail-navigate-btn.is-launching svg { animation: none; }
+}
.detail-grid {
display: grid;
@@ -1462,16 +1448,11 @@ body {
}
.segment-btn,
- .bw-btn,
- .access-trigger {
+ .bw-btn {
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 b2ccf73c..0ddd81c4 100644
--- a/vehicle-map/tests/test_app.mjs
+++ b/vehicle-map/tests/test_app.mjs
@@ -138,6 +138,22 @@ assert.ok(stationNavigationUrl.startsWith('https://uri.amap.com/navigation?to=11
assert.match(stationNavigationUrl, /%E5%B9%BF%E5%B7%9E/);
assert.match(stationNavigationUrl, /coordinate=gaode&callnative=1$/);
assert.equal(navigationUrlForEntity('vehicle', { plateNumber: '无位置车辆' }), null);
+const launchClassSet = new Set();
+const launchOverlay = { hidden: true, classList: { add(name) { launchClassSet.add(name); }, remove(name) { launchClassSet.delete(name); } } };
+const launchButton = { disabled: false, dataset: { navigationUrl: stationNavigationUrl }, classList: { add(name) { launchClassSet.add('button:' + name); }, remove(name) { launchClassSet.delete('button:' + name); }, contains(name) { return launchClassSet.has('button:' + name); } } };
+const launchTimers = [];
+document.getElementById = id => id === 'navigationLaunchOverlay' ? launchOverlay : id === 'detailNavigateBtn' ? launchButton : null;
+window.requestAnimationFrame = callback => callback();
+window.setTimeout = (callback, delay) => { launchTimers.push({ callback, delay }); return launchTimers.length; };
+window.clearTimeout = () => {};
+window.location = { assign() {} };
+selectedEntity = { mode: 'station', entity: dashboard.stations[0] };
+navigateToSelectedEntity();
+assert.equal(launchOverlay.hidden, false);
+assert.ok(launchClassSet.has('is-visible'));
+assert.ok(launchClassSet.has('button:is-launching'));
+assert.equal(launchButton.disabled, true);
+assert.deepEqual(launchTimers.map(timer => timer.delay), [140, 4000]);
const stationView = stationViewportSummary();
assert.equal(stationView.level, 'station');
assert.equal(stationView.visibleNodes.length, 2);
@@ -170,5 +186,7 @@ const sandbox = {
setInterval() {},
clearInterval() {}
};
+assert.match(source, /navigationLaunchOverlay/);
+assert.match(source, /正在打开高德地图/);
vm.runInNewContext(`${source}\n${checks}`, sandbox, { filename: 'app.js' });
console.log('vehicle hierarchy aggregation tests: ok');
diff --git a/vehicle-map/tests/test_server.py b/vehicle-map/tests/test_server.py
index 1296319c..f46d2f9d 100644
--- a/vehicle-map/tests/test_server.py
+++ b/vehicle-map/tests/test_server.py
@@ -1,9 +1,5 @@
import importlib.util
-import hashlib
-import hmac
-import json
from pathlib import Path
-import time
import unittest
from unittest.mock import patch
@@ -14,7 +10,7 @@ SPEC.loader.exec_module(server)
class DashboardTest(unittest.TestCase):
- def test_dashboard_aggregates_authorized_vehicle_and_station_data(self):
+ def test_dashboard_aggregates_public_vehicle_and_station_data(self):
def fake_post(path, body):
if path.endswith("realtime/query"):
return [
@@ -45,55 +41,11 @@ 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_static_asset_version_is_a_content_fingerprint(self):
self.assertEqual(len(server.STATIC_ASSET_VERSION), 16)
self.assertRegex(server.STATIC_ASSET_VERSION, r"^[0-9a-f]{16}$")
index_template = (Path(__file__).parents[1] / "index.html").read_text(encoding="utf-8")
self.assertEqual(index_template.count("__ASSET_VERSION__"), 2)
- 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()