ske087/rdm-card-reader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Serve a read-only status page on the bridge port

Browse Source 
GET / now returns a dashboard for the reader: mode, port, uptime,
card presence, the current card, event counters and a live event log
seeded from /api/taps and updated over the /api/events SSE stream.

The page is read-only on purpose - it has no claim button, because
POST /api/card/claim spends the tap and merely viewing the status
would otherwise consume a real login.

Serving the page requires allowing same-origin browser requests: a
browser sends Origin even for a request to the same origin, so
_cors_ok() used to reject the page's own API calls with 403 unless
--allow-origin was passed. Loopback and same-host origins are now
accepted, and the Access-Control-Allow-Origin echo is applied
consistently in one helper instead of four copies.

--no-page reverts POST-free deployments to API-only.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
scheianu
2026-10-05 21:34:17 +03:00
parent c720f4ca67
commit 882b89d6da
4 changed files with 436 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/rdm_card_reader/bridge.py
+71 -14
View File
@@ -55,8 +55,10 @@ import time
from collections import deque
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any, Callable, Deque, Dict, List, Optional
from urllib.parse import urlsplit
from .card import EVENT_INVALID, EVENT_INSERTED, EVENT_REMOVED, CardEvent
from .webui import render_page
#: Reader modes. ``login`` publishes inserts only (a tap = an authentication
#: attempt). ``monitor`` publishes inserts and removals both.
@@ -325,14 +327,31 @@ class _Handler(BaseHTTPRequestHandler):
# -- helpers ----------------------------------------------------------
def _origin_is_loopback(self, origin: str) -> bool:
"""True when *origin* is this server itself or a loopback address.
A browser sends ``Origin`` even for same-origin requests, so without
this the status page served from this very port could not call its own
API. A remote site cannot forge ``Origin: http://127.0.0.1``, so
trusting loopback here does not widen what an attacker can reach.
"""
host = self.headers.get("Host")
if host and origin in ("http://" + host, "https://" + host):
return True
try:
hostname = (urlsplit(origin).hostname or "").lower()
except ValueError:
return False
return hostname in ("127.0.0.1", "localhost", "::1", "0.0.0.0")
def _cors_ok(self) -> bool:
"""Refuse cross-origin reads unless the origin was allow-listed."""
origin = self.headers.get("Origin")
if not origin:
return True # same-origin fetch or a non-browser client
if self._origin_is_loopback(origin):
return True
allowed: List[str] = getattr(self.server, "allowed_origins", [])
if not allowed:
return False # browsers only; opt out entirely
return origin in allowed
def _token_ok(self) -> bool:
@@ -351,26 +370,52 @@ class _Handler(BaseHTTPRequestHandler):
supplied = pair.split("=", 1)[1]
return secrets.compare_digest(supplied, required)
def _allowed_origin(self) -> Optional[str]:
"""The Origin to echo in ``Access-Control-Allow-Origin``, or None.
Same-origin and loopback callers are echoed so the dashboard can read
its own API; any other origin must be on the allow-list.
"""
origin = self.headers.get("Origin")
if not origin:
return None
if self._origin_is_loopback(origin):
return origin
if origin in getattr(self.server, "allowed_origins", []):
return origin
return None
def _send_json(self, payload: Any, status: int = 200) -> None:
body = json.dumps(payload).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
origin = self.headers.get("Origin")
if origin and origin in getattr(self.server, "allowed_origins", []):
self.send_header("Access-Control-Allow-Origin", origin)
allowed = self._allowed_origin()
if allowed:
self.send_header("Access-Control-Allow-Origin", allowed)
self.send_header("Vary", "Origin")
self.end_headers()
self.wfile.write(body)
def _send_bytes(self, body: bytes, content_type: str, status: int = 200) -> None:
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def _send_page(self) -> None:
self._send_bytes(render_page(), "text/html; charset=utf-8")
def _send_no_content(self) -> None:
self.send_response(204)
self.send_header("Cache-Control", "no-store")
self.send_header("Content-Length", "0")
origin = self.headers.get("Origin")
if origin and origin in getattr(self.server, "allowed_origins", []):
self.send_header("Access-Control-Allow-Origin", origin)
allowed = self._allowed_origin()
if allowed:
self.send_header("Access-Control-Allow-Origin", allowed)
self.send_header("Vary", "Origin")
self.end_headers()
@@ -388,6 +433,16 @@ class _Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
path = self.path.split("?", 1)[0].rstrip("/") or "/"
# The status page is served before the CORS and token checks, on
# purpose: it is static, reveals nothing on its own, and has to be able
# to load so it can prompt for a token when the API demands one.
if path == "/":
if getattr(self.server, "serve_page", True):
self._send_page()
else:
self._send_json({"error": "not found", "path": "/"}, status=404)
return
if path == "/api/health":
self._send_json(self._hub.health())
return
@@ -476,10 +531,10 @@ class _Handler(BaseHTTPRequestHandler):
self._send_json({"mode": mode})
def do_OPTIONS(self) -> None: # noqa: N802
origin = self.headers.get("Origin")
self.send_response(204)
if origin and origin in getattr(self.server, "allowed_origins", []):
self.send_header("Access-Control-Allow-Origin", origin)
allowed = self._allowed_origin()
if allowed:
self.send_header("Access-Control-Allow-Origin", allowed)
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type, X-Auth-Token")
self.send_header("Vary", "Origin")
@@ -495,9 +550,9 @@ class _Handler(BaseHTTPRequestHandler):
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-store")
self.send_header("Connection", "keep-alive")
origin = self.headers.get("Origin")
if origin and origin in getattr(self.server, "allowed_origins", []):
self.send_header("Access-Control-Allow-Origin", origin)
allowed = self._allowed_origin()
if allowed:
self.send_header("Access-Control-Allow-Origin", allowed)
self.send_header("Vary", "Origin")
self.end_headers()
@@ -532,11 +587,13 @@ def create_server(
allowed_origins: Optional[List[str]] = None,
token: Optional[str] = None,
logger: Optional[Callable[[str], None]] = None,
serve_page: bool = True,
) -> ThreadingHTTPServer:
"""Build (but do not start) the HTTP server around *hub*."""
server = ThreadingHTTPServer((host, port), _Handler)
server.daemon_threads = True
server.hub = hub # type: ignore[attr-defined]
server.serve_page = serve_page # type: ignore[attr-defined]
server.allowed_origins = list(allowed_origins or []) # type: ignore[attr-defined]
server.token = token or os.environ.get("RDM_BRIDGE_TOKEN") or None # type: ignore[attr-defined]
server.logger = logger # type: ignore[attr-defined]