feat(acr-wedge+kiosk): SSE bridge for ACR122U / PC-SC readers

macOS keystroke injection from a CLI-launched Python hits multiple
TCC permission walls (Accessibility AND Automation, both attaching
to identities macOS often can't resolve cleanly). After bouncing
through Quartz, AppleScript, and pyautogui fallbacks, none of them
worked reliably in our test environment.

Switch to a proper IPC channel instead of pretending to be a
keyboard.

Daemon (wedge.py):
  - Adds a ThreadingHTTPServer on 127.0.0.1:8765 exposing /events
  - SSE stream pushes each detected UID as one event
  - 30s keep-alive comments to keep idle connections open
  - CORS: Access-Control-Allow-Origin: * (kiosk page may be on any
    client-domain HTTPS origin; SSE source is always localhost)
  - Keystroke injection kept as best-effort fallback for non-SSE
    clients

Kiosk JS (fusion_clock_nfc_kiosk.js):
  - Adds startWedgeSseListener() that opens EventSource to
    http://localhost:8765/events on setup
  - On message: same handleTap()/_onEnrollTap() flow as Web NFC + HID
  - EventSource auto-reconnects; first error is logged then silenced
  - http://localhost is a "potentially trustworthy origin" so this
    works from https:// pages without mixed-content blocking

Result: ACR122U + wedge.py daemon now drives the kiosk with zero
macOS permission prompts and no focused-window dependency. Same
input plumbing as Web NFC and HID — penalty/photo/activity log
fire identically.

Bump fusion_clock to 19.0.3.3.0.
This commit is contained in:
gsinghpal
2026-05-15 20:10:40 -04:00
parent d75198be9f
commit 00981a502a
3 changed files with 161 additions and 9 deletions

View File

@@ -49,7 +49,9 @@ Usage:
import argparse
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
try:
from smartcard.System import readers
@@ -62,6 +64,81 @@ except ImportError:
print(" (macOS may also need: brew install swig)", file=sys.stderr)
sys.exit(1)
# ────────────────────────────────────────────────────────────────────
# Server-Sent Events (SSE) endpoint — primary IPC to the kiosk page.
#
# The kiosk JS connects to http://localhost:8765/events via EventSource
# and gets pushed each UID as it's detected. Chrome treats http://localhost
# as a secure origin, so this works from an HTTPS kiosk page without
# mixed-content warnings. No keystroke injection, no Accessibility prompt,
# no focus dependency.
# ────────────────────────────────────────────────────────────────────
SSE_HOST = "127.0.0.1"
SSE_PORT = 8765
_sse_lock = threading.Lock()
_sse_latest_uid = None
_sse_latest_ts = 0.0
def _sse_publish(uid):
"""Make ``uid`` available to all connected SSE clients."""
global _sse_latest_uid, _sse_latest_ts
with _sse_lock:
_sse_latest_uid = uid
_sse_latest_ts = time.time()
class _TapSSEHandler(BaseHTTPRequestHandler):
# Quiet the default per-request access log.
def log_message(self, format, *args):
return
def do_GET(self):
if self.path != "/events":
self.send_response(404)
self.end_headers()
return
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.send_header("Connection", "keep-alive")
# Allow ANY origin to connect — the kiosk page lives on
# https://<client-domain>/fusion_clock/kiosk/nfc and the
# daemon runs on the same machine as the kiosk browser.
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
last_seen_ts = 0.0
try:
while True:
with _sse_lock:
uid = _sse_latest_uid
ts = _sse_latest_ts
if uid and ts > last_seen_ts:
self.wfile.write(f"data: {uid}\n\n".encode("utf-8"))
self.wfile.flush()
last_seen_ts = ts
else:
# 30s keep-alive comment line so proxies/network stacks
# don't kill an idle connection.
self.wfile.write(b": keep-alive\n\n")
self.wfile.flush()
time.sleep(0.1)
except (BrokenPipeError, ConnectionResetError, OSError):
pass # client disconnected
def start_sse_server(verbose=False):
"""Start the SSE server in a daemon thread. Returns the server."""
server = ThreadingHTTPServer((SSE_HOST, SSE_PORT), _TapSSEHandler)
t = threading.Thread(target=server.serve_forever, daemon=True)
t.start()
if verbose:
print(f"[wedge] SSE server listening on http://{SSE_HOST}:{SSE_PORT}/events")
return server
# Standard PC/SC APDU to get the card UID. Works on ACR122U, ACR1252U,
# and any other PC/SC reader following the ACS extension spec.
GET_UID_APDU = [0xFF, 0xCA, 0x00, 0x00, 0x00]
@@ -144,12 +221,23 @@ def _emit_other(uid):
def emit_uid(uid, type_keys=True, verbose=False):
"""Type UID + Enter into the focused window."""
"""Publish UID via SSE (primary) and optionally type it (fallback)."""
if verbose:
print(f"[wedge] TAP {uid}")
# Primary path: push to all SSE clients.
_sse_publish(uid)
if not type_keys:
return
# Optional secondary path: keystroke injection. Useful for testing
# in TextEdit / other apps, or for browsers/devices that don't
# connect to the SSE endpoint (Safari, old Chrome, etc.).
#
# On macOS this often hits TCC permission walls (Accessibility,
# Automation). The SSE path above is the reliable production
# mechanism — keystroke injection is best-effort.
if sys.platform == "darwin":
try:
_emit_macos(uid)
@@ -161,13 +249,15 @@ def emit_uid(uid, type_keys=True, verbose=False):
try:
_emit_other(uid)
except ImportError:
print(
"[wedge] pyautogui is not installed; cannot type. "
"Run: pip3 install pyautogui (or use --no-type for detect-only)",
file=sys.stderr,
)
if verbose:
print(
"[wedge] pyautogui is not installed; SSE-only mode. "
"Run: pip3 install pyautogui (for keystroke fallback)",
file=sys.stderr,
)
except Exception as e:
print(f"[wedge] type error: {e}", file=sys.stderr)
if verbose:
print(f"[wedge] type error: {e}", file=sys.stderr)
def run_loop(args):
@@ -241,7 +331,14 @@ def main():
print("[wedge] fusion_clock ACR wedge starting. Ctrl+C to stop.")
if args.no_type:
print("[wedge] --no-type: tags will be detected but not typed.")
print("[wedge] --no-type: keystroke injection disabled (SSE only).")
# Always start the SSE server — it's the primary IPC channel.
try:
start_sse_server(verbose=True)
except OSError as e:
print(f"[wedge] WARNING: could not start SSE server on port {SSE_PORT}: {e}")
print("[wedge] Daemon will continue in keystroke-only mode.")
try:
run_loop(args)