From c60ff543b5921175825eedf56df82f82b4c9888c Mon Sep 17 00:00:00 2001 From: AJV20 <24819659+AJV20@users.noreply.github.com> Date: Sun, 24 May 2026 13:29:02 -0400 Subject: [PATCH] feat: add passkey sign-in --- CHANGELOG.md | 5 +- README.md | 2 + api/auth.py | 1 + api/passkeys.py | 359 +++++++++++++++++++++++++++++++++++++ api/routes.py | 78 +++++++- requirements.txt | 3 +- static/index.html | 6 + static/login.js | 66 +++++++ static/panels.js | 59 ++++++ tests/test_passkey_auth.py | 149 +++++++++++++++ 10 files changed, 724 insertions(+), 4 deletions(-) create mode 100644 api/passkeys.py create mode 100644 tests/test_passkey_auth.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c964954..c6bf8b03 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,10 @@ ## [Unreleased] +### Added + +- Optional passkey/WebAuthn sign-in for password-protected WebUI instances. Authenticated users can register/remove passkeys from Settings -> System, and `/login` shows a passwordless sign-in button only after a passkey exists. Password auth remains the default-off bootstrap and recovery path. + ### Fixed - Clarify `Response interrupted` recovery markers so they report that the live response stream stopped instead of asserting that the WebUI process restarted. The recovery path now records distinct interruption causes for real process restarts, stream/run split-brain, and lost worker bookkeeping; browser-side SSE transport failures show a separate `Connection interrupted` message, client-side `BrokenPipeError` disconnects no longer get logged as server 500s, and chat/gateway SSE errors emit rate-limited, body-capped, sanitized client diagnostics to `/api/client-events/log` for future root-cause checks. The stream-status `terminal_state` value for lost-worker bookkeeping changes from `stale-from-restart` to `lost-worker-bookkeeping`, matching the new non-restart wording. @@ -145,7 +149,6 @@ - UX evidence for #2812 captured at 1280/1440/1920/mobile (iPhone 14 emulation); Telegram-approved. - File a follow-up issue for pdeathsig-on-supervisor-thread hardening (#2854 deferred Option B) and French-locale `open_in_vscode` parity gap (predates this batch, Opus advisor flagged). - ## [v0.51.126] — 2026-05-24 — Release CX (stage-batch8 — 2-PR low-risk batch — kanban markdown + live activity timeline) ### Added diff --git a/README.md b/README.md index ccaca241..56af9e10 100644 --- a/README.md +++ b/README.md @@ -467,6 +467,8 @@ Production data and real cron jobs are never touched. Current snapshot: ### Authentication and security - Optional password auth -- off by default, zero friction for localhost - Enable via `HERMES_WEBUI_PASSWORD` env var or Settings panel +- Optional passkeys/WebAuthn -- register from Settings -> System after signing in with a password; the login page only shows passkey sign-in after at least one passkey exists +- Password auth remains the bootstrap/recovery path; passkeys are same-origin and stored locally in the WebUI state directory - Signed HMAC HTTP-only cookie with 24h TTL - Minimal dark-themed login page at `/login` - Security headers on all responses (X-Content-Type-Options, X-Frame-Options, Referrer-Policy) diff --git a/api/auth.py b/api/auth.py index 5a49516f..9232e2c7 100644 --- a/api/auth.py +++ b/api/auth.py @@ -49,6 +49,7 @@ def _resolve_session_ttl() -> int: PUBLIC_PATHS = frozenset({ '/login', '/health', '/favicon.ico', '/sw.js', '/api/auth/login', '/api/auth/status', + '/api/auth/passkey/options', '/api/auth/passkey/login', '/manifest.json', '/manifest.webmanifest', '/session/manifest.json', '/session/manifest.webmanifest', }) diff --git a/api/passkeys.py b/api/passkeys.py new file mode 100644 index 00000000..dbfbb65d --- /dev/null +++ b/api/passkeys.py @@ -0,0 +1,359 @@ +"""Passkey/WebAuthn helpers for Hermes WebUI. + +Default-off: passkeys are only advertised after an authenticated user registers +one from Settings. Password auth remains the bootstrap/recovery mechanism. +""" +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import os +import secrets +import tempfile +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from api.config import STATE_DIR + +try: # optional at import-time; endpoints return a clear error if unavailable + from cryptography.exceptions import InvalidSignature + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import ec +except Exception: # pragma: no cover - exercised by source tests instead + InvalidSignature = Exception # type: ignore[assignment] + hashes = serialization = ec = None # type: ignore[assignment] + +_CREDENTIALS_FILE = STATE_DIR / "passkeys.json" +_CHALLENGES_FILE = STATE_DIR / ".passkey_challenges.json" +_CHALLENGE_TTL = 300 +_RP_NAME = "Hermes WebUI" + + +class PasskeyError(ValueError): + """Raised for user-correctable WebAuthn failures.""" + + +def _b64u(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _b64u_decode(value: str | bytes) -> bytes: + if isinstance(value, bytes): + value = value.decode("ascii") + value = str(value).strip() + value += "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(value.encode("ascii")) + + +def _json_load(path: Path, default: Any) -> Any: + try: + if path.exists(): + return json.loads(path.read_text(encoding="utf-8")) + except Exception: + return default + return default + + +def _atomic_write_json(path: Path, payload: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2, sort_keys=True) + os.chmod(tmp, 0o600) + os.replace(tmp, path) + except Exception: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def _load_credentials() -> list[dict[str, Any]]: + data = _json_load(_CREDENTIALS_FILE, []) + if not isinstance(data, list): + return [] + return [c for c in data if isinstance(c, dict) and isinstance(c.get("id"), str)] + + +def _save_credentials(creds: list[dict[str, Any]]) -> None: + _atomic_write_json(_CREDENTIALS_FILE, creds) + + +def registered_credentials() -> list[dict[str, Any]]: + """Return public credential metadata only; never expose public keys.""" + out = [] + for c in _load_credentials(): + out.append({ + "id": c.get("id"), + "label": c.get("label") or "Passkey", + "created_at": c.get("created_at"), + "last_used_at": c.get("last_used_at"), + "sign_count": c.get("sign_count", 0), + }) + return out + + +def passkeys_available() -> bool: + return bool(_load_credentials()) + + +def _load_challenges() -> dict[str, dict[str, Any]]: + raw = _json_load(_CHALLENGES_FILE, {}) + if not isinstance(raw, dict): + return {} + now = time.time() + clean = { + k: v for k, v in raw.items() + if isinstance(k, str) and isinstance(v, dict) and now - float(v.get("ts", 0)) < _CHALLENGE_TTL + } + if clean != raw: + _atomic_write_json(_CHALLENGES_FILE, clean) + return clean + + +def _store_challenge(challenge: str, kind: str, rp_id: str, origin: str) -> None: + data = _load_challenges() + data[challenge] = {"kind": kind, "rp_id": rp_id, "origin": origin, "ts": time.time()} + _atomic_write_json(_CHALLENGES_FILE, data) + + +def _consume_challenge(challenge: str, kind: str) -> dict[str, Any]: + data = _load_challenges() + entry = data.pop(challenge, None) + _atomic_write_json(_CHALLENGES_FILE, data) + if not entry or entry.get("kind") != kind: + raise PasskeyError("Passkey challenge expired. Try again.") + return entry + + +def _host_without_port(host: str) -> str: + host = (host or "localhost").strip().split(",", 1)[0] + if host.startswith("[") and "]" in host: + return host[1:host.index("]")] + return host.rsplit(":", 1)[0] if ":" in host else host + + +def rp_context(handler) -> tuple[str, str]: + host = _host_without_port(handler.headers.get("Host", "localhost")) + proto = handler.headers.get("X-Forwarded-Proto", "").split(",", 1)[0].strip().lower() + if proto not in {"http", "https"}: + try: + from api.auth import _is_secure_context + proto = "https" if _is_secure_context(handler) else "http" + except AttributeError: + proto = "http" + return host, f"{proto}://{handler.headers.get('Host', host)}" + + +def registration_options(handler) -> dict[str, Any]: + rp_id, _origin = rp_context(handler) + challenge = _b64u(secrets.token_bytes(32)) + _store_challenge(challenge, "register", rp_id, _origin) + return { + "challenge": challenge, + "rp": {"name": _RP_NAME, "id": rp_id}, + "user": {"id": _b64u(hashlib.sha256(rp_id.encode()).digest()[:16]), "name": "Hermes WebUI", "displayName": "Hermes WebUI"}, + "pubKeyCredParams": [{"type": "public-key", "alg": -7}], + "authenticatorSelection": {"residentKey": "preferred", "userVerification": "preferred"}, + "timeout": 60000, + "attestation": "none", + "excludeCredentials": [{"type": "public-key", "id": c["id"]} for c in registered_credentials()], + } + + +def authentication_options(handler) -> dict[str, Any]: + creds = registered_credentials() + if not creds: + raise PasskeyError("No passkeys are registered.") + rp_id, origin = rp_context(handler) + challenge = _b64u(secrets.token_bytes(32)) + _store_challenge(challenge, "login", rp_id, origin) + return { + "challenge": challenge, + "rpId": rp_id, + "allowCredentials": [{"type": "public-key", "id": c["id"]} for c in creds], + "timeout": 60000, + "userVerification": "preferred", + } + + +@dataclass +class _Cbor: + data: bytes + pos: int = 0 + + def read(self, n: int) -> bytes: + if self.pos + n > len(self.data): + raise PasskeyError("Malformed CBOR data") + out = self.data[self.pos:self.pos + n] + self.pos += n + return out + + def item(self) -> Any: + initial = self.read(1)[0] + major, addl = initial >> 5, initial & 0x1F + val = self._val(addl) + if major == 0: + return val + if major == 1: + return -1 - val + if major == 2: + return self.read(val) + if major == 3: + return self.read(val).decode("utf-8") + if major == 4: + return [self.item() for _ in range(val)] + if major == 5: + return {self.item(): self.item() for _ in range(val)} + if major == 7: + if val == 20: + return False + if val == 21: + return True + if val == 22: + return None + raise PasskeyError("Unsupported CBOR data") + + def _val(self, addl: int) -> int: + if addl < 24: + return addl + if addl == 24: + return self.read(1)[0] + if addl == 25: + return int.from_bytes(self.read(2), "big") + if addl == 26: + return int.from_bytes(self.read(4), "big") + if addl == 27: + return int.from_bytes(self.read(8), "big") + raise PasskeyError("Indefinite CBOR values are not supported") + + +def _cbor_loads(data: bytes) -> Any: + parser = _Cbor(data) + value = parser.item() + if parser.pos != len(data): + raise PasskeyError("Trailing CBOR data") + return value + + +def _client_data(encoded: str, expected_type: str, challenge_kind: str) -> tuple[dict[str, Any], dict[str, Any], bytes]: + raw = _b64u_decode(encoded) + try: + data = json.loads(raw.decode("utf-8")) + except Exception as exc: + raise PasskeyError("Malformed client data") from exc + if data.get("type") != expected_type: + raise PasskeyError("Unexpected passkey response type") + challenge = data.get("challenge") + if not isinstance(challenge, str): + raise PasskeyError("Missing passkey challenge") + entry = _consume_challenge(challenge, challenge_kind) + if data.get("origin") != entry.get("origin"): + raise PasskeyError("Passkey origin mismatch") + return data, entry, raw + + +def _parse_auth_data(auth_data: bytes, rp_id: str) -> dict[str, Any]: + if len(auth_data) < 37: + raise PasskeyError("Malformed authenticator data") + rp_hash = auth_data[:32] + expected = hashlib.sha256(rp_id.encode("idna")).digest() + if not hmac.compare_digest(rp_hash, expected): + raise PasskeyError("Passkey RP ID mismatch") + flags = auth_data[32] + if not (flags & 0x01): + raise PasskeyError("Passkey user presence was not verified") + sign_count = int.from_bytes(auth_data[33:37], "big") + return {"flags": flags, "sign_count": sign_count, "rest": auth_data[37:]} + + +def _public_key_from_cose(cose: dict[Any, Any]): + if ec is None or serialization is None: + raise PasskeyError("Passkey support requires the cryptography package") + alg = cose.get(3) + kty = cose.get(1) + crv = cose.get(-1) + x = cose.get(-2) + y = cose.get(-3) + if alg != -7 or kty != 2 or crv != 1 or not isinstance(x, bytes) or not isinstance(y, bytes): + raise PasskeyError("Only ES256 passkeys are supported") + numbers = ec.EllipticCurvePublicNumbers(int.from_bytes(x, "big"), int.from_bytes(y, "big"), ec.SECP256R1()) + return numbers.public_key() + + +def finish_registration(payload: dict[str, Any], handler) -> dict[str, Any]: + response = payload.get("response") or {} + _client, entry, _client_raw = _client_data(response.get("clientDataJSON", ""), "webauthn.create", "register") + att_obj = _cbor_loads(_b64u_decode(response.get("attestationObject", ""))) + if not isinstance(att_obj, dict) or not isinstance(att_obj.get("authData"), bytes): + raise PasskeyError("Malformed attestation object") + parsed = _parse_auth_data(att_obj["authData"], entry["rp_id"]) + if not (parsed["flags"] & 0x40): + raise PasskeyError("Passkey credential data missing") + rest = parsed["rest"] + if len(rest) < 18: + raise PasskeyError("Malformed credential data") + cred_len = int.from_bytes(rest[16:18], "big") + credential_id = rest[18:18 + cred_len] + cose_bytes = rest[18 + cred_len:] + cose_key = _cbor_loads(cose_bytes) + public_key = _public_key_from_cose(cose_key) + pem = public_key.public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo).decode("ascii") + cred_id = _b64u(credential_id) + label = str(payload.get("label") or "Passkey").strip()[:80] or "Passkey" + creds = [c for c in _load_credentials() if c.get("id") != cred_id] + creds.append({ + "id": cred_id, + "label": label, + "public_key_pem": pem, + "sign_count": parsed["sign_count"], + "created_at": time.time(), + "last_used_at": None, + }) + _save_credentials(creds) + return {"ok": True, "credential": {"id": cred_id, "label": label}} + + +def finish_login(payload: dict[str, Any], handler) -> dict[str, Any]: + if serialization is None or hashes is None: + raise PasskeyError("Passkey support requires the cryptography package") + response = payload.get("response") or {} + cred_id = payload.get("id") or payload.get("rawId") + if not isinstance(cred_id, str): + raise PasskeyError("Missing passkey credential id") + creds = _load_credentials() + idx = next((i for i, c in enumerate(creds) if c.get("id") == cred_id), -1) + if idx < 0: + raise PasskeyError("Unknown passkey") + _client, entry, client_raw = _client_data(response.get("clientDataJSON", ""), "webauthn.get", "login") + auth_data = _b64u_decode(response.get("authenticatorData", "")) + parsed = _parse_auth_data(auth_data, entry["rp_id"]) + signature = _b64u_decode(response.get("signature", "")) + public_key = serialization.load_pem_public_key(str(creds[idx].get("public_key_pem", "")).encode("ascii")) + signed = auth_data + hashlib.sha256(client_raw).digest() + try: + public_key.verify(signature, signed, ec.ECDSA(hashes.SHA256())) + except InvalidSignature as exc: + raise PasskeyError("Passkey signature verification failed") from exc + old_count = int(creds[idx].get("sign_count") or 0) + if parsed["sign_count"] and old_count and parsed["sign_count"] <= old_count: + raise PasskeyError("Passkey sign counter did not advance") + creds[idx]["sign_count"] = parsed["sign_count"] or old_count + creds[idx]["last_used_at"] = time.time() + _save_credentials(creds) + return {"ok": True, "credential_id": cred_id} + + +def delete_credential(credential_id: str) -> dict[str, Any]: + creds = _load_credentials() + kept = [c for c in creds if c.get("id") != credential_id] + if len(kept) == len(creds): + raise PasskeyError("Passkey not found") + _save_credentials(kept) + return {"ok": True, "credentials": registered_credentials()} diff --git a/api/routes.py b/api/routes.py index c482a385..b6e030e1 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1278,7 +1278,12 @@ def _is_browser_unsafe_request(handler) -> bool: def _csrf_exempt_path(path: str) -> bool: """Paths that cannot or must not carry a session CSRF token.""" - return path in {"/api/auth/login", "/api/csp-report"} + return path in { + "/api/auth/login", + "/api/auth/passkey/options", + "/api/auth/passkey/login", + "/api/csp-report", + } _CSRF_FAILURE_ATTR = "_hermes_csrf_failure_reason" @@ -2792,6 +2797,7 @@ button{width:100%;padding:10px;border-radius:10px;border:none;background:rgba(12 border:1px solid rgba(124,185,255,.3);color:#7cb9ff;font-size:14px;font-weight:600;cursor:pointer; transition:all .15s} button:hover{background:rgba(124,185,255,.25)} +.passkey-login{margin-top:10px;background:rgba(255,255,255,.04);border-color:rgba(232,160,48,.35);color:#e8a030} .err{color:#e94560;font-size:12px;margin-top:10px;display:none}
./ctl.sh start and want to stop without switching to a terminal. You'll need to relaunch the server (./ctl.sh start or the native app) before the WebUI is reachable again.hermes dashboard is reachable. Public reverse-proxy URLs are stored as browser-only links and are never server-probed.