feat: add passkey sign-in
This commit is contained in:
1 parent
15c5f4cf05
commit
c60ff543b5
10 files changed
+724
-4
No files matched your search
+4
-1
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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',
|
||||
})
|
||||
|
||||
+359
@@ -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()}
|
||||
+76
-2
@@ -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}
|
||||
</style></head><body>
|
||||
<div class="card">
|
||||
@@ -2801,6 +2807,7 @@ button:hover{background:rgba(124,185,255,.25)}
|
||||
<form id="login-form" data-invalid-pw="{{LOGIN_INVALID_PW}}" data-conn-failed="{{LOGIN_CONN_FAILED}}">
|
||||
<input type="password" id="pw" placeholder="{{LOGIN_PLACEHOLDER}}" autofocus>
|
||||
<button type="submit">{{LOGIN_BTN}}</button>
|
||||
<button type="button" id="passkey-login" class="passkey-login" style="display:none">Sign in with passkey</button>
|
||||
</form>
|
||||
<div class="err" id="err"></div>
|
||||
</div>
|
||||
@@ -3746,12 +3753,19 @@ def handle_get(handler, parsed) -> bool:
|
||||
|
||||
if parsed.path == "/api/auth/status":
|
||||
from api.auth import is_auth_enabled, parse_cookie, verify_session
|
||||
from api.passkeys import registered_credentials
|
||||
|
||||
logged_in = False
|
||||
if is_auth_enabled():
|
||||
cv = parse_cookie(handler)
|
||||
logged_in = bool(cv and verify_session(cv))
|
||||
return j(handler, {"auth_enabled": is_auth_enabled(), "logged_in": logged_in})
|
||||
passkeys = registered_credentials()
|
||||
return j(handler, {
|
||||
"auth_enabled": is_auth_enabled(),
|
||||
"logged_in": logged_in,
|
||||
"passkeys_enabled": bool(passkeys),
|
||||
"passkeys_count": len(passkeys),
|
||||
})
|
||||
|
||||
if parsed.path in ("/manifest.json", "/manifest.webmanifest"):
|
||||
return _serve_manifest(handler)
|
||||
@@ -6409,6 +6423,66 @@ def handle_post(handler, parsed) -> bool:
|
||||
handler.wfile.write(body)
|
||||
return True
|
||||
|
||||
if parsed.path == "/api/auth/passkey/options":
|
||||
from api.auth import is_auth_enabled
|
||||
from api.passkeys import PasskeyError, authentication_options
|
||||
|
||||
if not is_auth_enabled():
|
||||
return j(handler, {"error": "Auth not enabled"}, status=400)
|
||||
try:
|
||||
return j(handler, {"ok": True, "publicKey": authentication_options(handler)})
|
||||
except PasskeyError as e:
|
||||
return bad(handler, str(e), status=400)
|
||||
|
||||
if parsed.path == "/api/auth/passkey/login":
|
||||
from api.auth import create_session, is_auth_enabled, set_auth_cookie
|
||||
from api.auth import _check_login_rate, _record_login_attempt
|
||||
from api.passkeys import PasskeyError, finish_login
|
||||
|
||||
if not is_auth_enabled():
|
||||
return j(handler, {"error": "Auth not enabled"}, status=400)
|
||||
client_ip = handler.client_address[0]
|
||||
if not _check_login_rate(client_ip):
|
||||
return j(handler, {"error": "Too many attempts. Try again in a minute."}, status=429)
|
||||
try:
|
||||
finish_login(body, handler)
|
||||
except PasskeyError as e:
|
||||
_record_login_attempt(client_ip)
|
||||
return bad(handler, str(e), status=401)
|
||||
cookie_val = create_session()
|
||||
handler.send_response(200)
|
||||
handler.send_header("Content-Type", "application/json")
|
||||
handler.send_header("Cache-Control", "no-store")
|
||||
_security_headers(handler)
|
||||
set_auth_cookie(handler, cookie_val)
|
||||
handler.end_headers()
|
||||
handler.wfile.write(json.dumps({"ok": True}).encode())
|
||||
return True
|
||||
|
||||
if parsed.path == "/api/auth/passkey/register/options":
|
||||
from api.passkeys import registration_options
|
||||
return j(handler, {"ok": True, "publicKey": registration_options(handler)})
|
||||
|
||||
if parsed.path == "/api/auth/passkey/register":
|
||||
from api.passkeys import PasskeyError, finish_registration, registered_credentials
|
||||
try:
|
||||
result = finish_registration(body, handler)
|
||||
result["credentials"] = registered_credentials()
|
||||
return j(handler, result)
|
||||
except PasskeyError as e:
|
||||
return bad(handler, str(e), status=400)
|
||||
|
||||
if parsed.path == "/api/auth/passkey/delete":
|
||||
from api.passkeys import PasskeyError, delete_credential
|
||||
try:
|
||||
return j(handler, delete_credential(str(body.get("id") or "")))
|
||||
except PasskeyError as e:
|
||||
return bad(handler, str(e), status=404)
|
||||
|
||||
if parsed.path == "/api/auth/passkeys":
|
||||
from api.passkeys import registered_credentials
|
||||
return j(handler, {"credentials": registered_credentials()})
|
||||
|
||||
if parsed.path == "/api/auth/logout":
|
||||
from api.auth import clear_auth_cookie, invalidate_session, parse_cookie
|
||||
|
||||
|
||||
+2
-1
@@ -1,4 +1,5 @@
|
||||
# Hermes Web UI -- minimal Python dependencies
|
||||
# The server uses only stdlib + pyyaml.
|
||||
# The server uses PyYAML plus cryptography for optional local passkey/WebAuthn support.
|
||||
# All heavy ML/agent deps live in the Hermes agent venv.
|
||||
pyyaml>=6.0
|
||||
cryptography>=42.0
|
||||
@@ -1267,6 +1267,12 @@
|
||||
<div style="font-size:11px;color:var(--muted);margin-bottom:8px" data-i18n="settings_desc_shutdown">Gracefully stops the local WebUI server. Useful if you launched via <code>./ctl.sh start</code> and want to stop without switching to a terminal. You'll need to relaunch the server (<code>./ctl.sh start</code> or the native app) before the WebUI is reachable again.</div>
|
||||
<button class="sm-btn" id="btnShutdownServer" onclick="shutdownServer()" style="width:100%;padding:8px;font-weight:600;color:#e74c3c;border-color:rgba(231,76,60,.3)" data-i18n="settings_btn_shutdown">Stop server</button>
|
||||
</div>
|
||||
<div class="settings-field" style="margin-top:18px;padding-top:16px;border-top:1px solid var(--border)">
|
||||
<label>Passkeys</label>
|
||||
<div style="font-size:11px;color:var(--muted);margin-bottom:8px">Register this browser or device for passwordless sign-in. Password auth remains the recovery path.</div>
|
||||
<button class="sm-btn" id="btnRegisterPasskey" onclick="registerPasskey()" style="width:100%;padding:8px;font-weight:600">Add passkey</button>
|
||||
<div id="passkeyList" style="margin-top:8px;font-size:12px;color:var(--muted)">No passkeys registered.</div>
|
||||
</div>
|
||||
<div class="settings-field" style="margin-top:18px;padding-top:16px;border-top:1px solid var(--border)">
|
||||
<label for="settingsDashboardMode">Official Hermes Dashboard</label>
|
||||
<div style="font-size:11px;color:var(--muted);margin-bottom:8px">Show a nav-rail link when the official <code>hermes dashboard</code> is reachable. Public reverse-proxy URLs are stored as browser-only links and are never server-probed.</div>
|
||||
|
||||
@@ -5,6 +5,7 @@
|
||||
document.addEventListener('DOMContentLoaded', function () {
|
||||
var form = document.getElementById('login-form');
|
||||
var input = document.getElementById('pw');
|
||||
var passkeyBtn = document.getElementById('passkey-login');
|
||||
|
||||
if (!form || !input) return;
|
||||
|
||||
@@ -60,6 +61,71 @@ document.addEventListener('DOMContentLoaded', function () {
|
||||
|
||||
form.addEventListener('submit', doLogin);
|
||||
|
||||
function b64uToBytes(s) {
|
||||
s = String(s || '').replace(/-/g, '+').replace(/_/g, '/');
|
||||
while (s.length % 4) s += '=';
|
||||
var bin = atob(s);
|
||||
var out = new Uint8Array(bin.length);
|
||||
for (var i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i);
|
||||
return out;
|
||||
}
|
||||
|
||||
function bytesToB64u(buf) {
|
||||
var bytes = new Uint8Array(buf);
|
||||
var bin = '';
|
||||
for (var i = 0; i < bytes.length; i++) bin += String.fromCharCode(bytes[i]);
|
||||
return btoa(bin).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/g, '');
|
||||
}
|
||||
|
||||
async function doPasskeyLogin() {
|
||||
if (!window.PublicKeyCredential || !navigator.credentials) return;
|
||||
hideErr();
|
||||
try {
|
||||
passkeyBtn.disabled = true;
|
||||
var optRes = await fetch('api/auth/passkey/options', { method: 'POST', body: '{}', credentials: 'include' });
|
||||
var optData = await optRes.json();
|
||||
if (!optRes.ok || !optData.publicKey) throw new Error(optData.error || 'Passkey unavailable');
|
||||
var pk = optData.publicKey;
|
||||
pk.challenge = b64uToBytes(pk.challenge);
|
||||
if (Array.isArray(pk.allowCredentials)) {
|
||||
pk.allowCredentials = pk.allowCredentials.map(function (c) { return Object.assign({}, c, { id: b64uToBytes(c.id) }); });
|
||||
}
|
||||
var cred = await navigator.credentials.get({ publicKey: pk });
|
||||
if (!cred) throw new Error('Passkey sign-in cancelled');
|
||||
var payload = {
|
||||
id: cred.id,
|
||||
rawId: bytesToB64u(cred.rawId),
|
||||
type: cred.type,
|
||||
response: {
|
||||
authenticatorData: bytesToB64u(cred.response.authenticatorData),
|
||||
clientDataJSON: bytesToB64u(cred.response.clientDataJSON),
|
||||
signature: bytesToB64u(cred.response.signature),
|
||||
userHandle: cred.response.userHandle ? bytesToB64u(cred.response.userHandle) : null,
|
||||
},
|
||||
};
|
||||
var res = await fetch('api/auth/passkey/login', {
|
||||
method: 'POST', headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify(payload), credentials: 'include',
|
||||
});
|
||||
var data = {};
|
||||
try { data = await res.json(); } catch (_) {}
|
||||
if (res.ok && data.ok) window.location.href = _safeNextPath();
|
||||
else showErr(data.error || invalidPw);
|
||||
} catch (ex) {
|
||||
showErr(ex && ex.message ? ex.message : connFailed);
|
||||
} finally {
|
||||
passkeyBtn.disabled = false;
|
||||
}
|
||||
}
|
||||
|
||||
if (passkeyBtn && window.PublicKeyCredential && navigator.credentials) {
|
||||
fetch('api/auth/status', { credentials: 'include' })
|
||||
.then(function (r) { return r.ok ? r.json() : null; })
|
||||
.then(function (s) { if (s && s.passkeys_enabled) passkeyBtn.style.display = 'block'; })
|
||||
.catch(function () {});
|
||||
passkeyBtn.addEventListener('click', doPasskeyLogin);
|
||||
}
|
||||
|
||||
input.addEventListener('keydown', function (e) {
|
||||
if (e.key === 'Enter') {
|
||||
e.preventDefault();
|
||||
|
||||
@@ -6260,6 +6260,7 @@ async function loadSettingsPanel(){
|
||||
const authStatus=await api('/api/auth/status');
|
||||
_setSettingsAuthButtonsVisible(!!authStatus.auth_enabled);
|
||||
}catch(e){}
|
||||
loadPasskeys();
|
||||
// #1560: env-var-locked password also disables the Disable Auth button —
|
||||
// clearing settings.password_hash is silent no-op when the env var is set,
|
||||
// and the backend now returns 409 anyway, so don't offer the action.
|
||||
@@ -6925,6 +6926,64 @@ function _setSettingsAuthButtonsVisible(active){
|
||||
if(signOutBtn) signOutBtn.style.display=active?'':'none';
|
||||
const disableBtn=$('btnDisableAuth');
|
||||
if(disableBtn) disableBtn.style.display=active?'':'none';
|
||||
const passkeyBtn=$('btnRegisterPasskey');
|
||||
if(passkeyBtn) passkeyBtn.disabled=!active||!window.PublicKeyCredential||!navigator.credentials;
|
||||
}
|
||||
|
||||
function _b64uToBytes(s){
|
||||
s=String(s||'').replace(/-/g,'+').replace(/_/g,'/');
|
||||
while(s.length%4) s+='=';
|
||||
const bin=atob(s), out=new Uint8Array(bin.length);
|
||||
for(let i=0;i<bin.length;i++) out[i]=bin.charCodeAt(i);
|
||||
return out;
|
||||
}
|
||||
function _bytesToB64u(buf){
|
||||
const bytes=new Uint8Array(buf);let bin='';
|
||||
for(let i=0;i<bytes.length;i++) bin+=String.fromCharCode(bytes[i]);
|
||||
return btoa(bin).replace(/\+/g,'-').replace(/\//g,'_').replace(/=+$/g,'');
|
||||
}
|
||||
|
||||
async function loadPasskeys(){
|
||||
const list=$('passkeyList');
|
||||
if(!list) return;
|
||||
if(!window.PublicKeyCredential||!navigator.credentials){
|
||||
list.textContent='Passkeys are not supported by this browser/context.';
|
||||
const btn=$('btnRegisterPasskey'); if(btn) btn.disabled=true;
|
||||
return;
|
||||
}
|
||||
try{
|
||||
const data=await api('/api/auth/passkeys',{method:'POST',body:'{}'});
|
||||
const creds=(data&&data.credentials)||[];
|
||||
if(!creds.length){list.textContent='No passkeys registered.';return;}
|
||||
list.innerHTML=creds.map(c=>`<div style="display:flex;align-items:center;justify-content:space-between;gap:8px;border:1px solid var(--border);border-radius:8px;padding:8px;margin-top:6px"><span>${esc(c.label||'Passkey')}</span><button class="btn-tiny" onclick="deletePasskey('${esc(c.id)}')">Remove</button></div>`).join('');
|
||||
}catch(e){list.textContent='Failed to load passkeys: '+e.message;}
|
||||
}
|
||||
|
||||
async function registerPasskey(){
|
||||
if(!window.PublicKeyCredential||!navigator.credentials){showToast('Passkeys require a supported browser and secure context.');return;}
|
||||
const label=(window.prompt('Name this passkey', 'This device')||'Passkey').trim();
|
||||
try{
|
||||
const optData=await api('/api/auth/passkey/register/options',{method:'POST',body:'{}'});
|
||||
const pk=optData.publicKey;
|
||||
pk.challenge=_b64uToBytes(pk.challenge);
|
||||
pk.user=Object.assign({},pk.user,{id:_b64uToBytes(pk.user.id)});
|
||||
if(Array.isArray(pk.excludeCredentials)) pk.excludeCredentials=pk.excludeCredentials.map(c=>Object.assign({},c,{id:_b64uToBytes(c.id)}));
|
||||
const cred=await navigator.credentials.create({publicKey:pk});
|
||||
if(!cred) throw new Error('Passkey registration cancelled');
|
||||
await api('/api/auth/passkey/register',{method:'POST',body:JSON.stringify({
|
||||
id:cred.id,rawId:_bytesToB64u(cred.rawId),type:cred.type,label,
|
||||
response:{clientDataJSON:_bytesToB64u(cred.response.clientDataJSON),attestationObject:_bytesToB64u(cred.response.attestationObject)}
|
||||
})});
|
||||
showToast('Passkey registered');
|
||||
loadPasskeys();
|
||||
}catch(e){showToast('Passkey registration failed: '+e.message);}
|
||||
}
|
||||
|
||||
async function deletePasskey(id){
|
||||
const ok=await showConfirmDialog({title:'Remove passkey?',message:'This browser/device will no longer be able to sign in with that passkey.',confirmLabel:'Remove',danger:true,focusCancel:true});
|
||||
if(!ok) return;
|
||||
try{await api('/api/auth/passkey/delete',{method:'POST',body:JSON.stringify({id})});showToast('Passkey removed');loadPasskeys();}
|
||||
catch(e){showToast('Failed to remove passkey: '+e.message);}
|
||||
}
|
||||
|
||||
function _applySavedSettingsUi(saved, body, opts){
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
import base64
|
||||
import json
|
||||
import hashlib
|
||||
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
|
||||
|
||||
def b64u(data: bytes) -> str:
|
||||
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
|
||||
|
||||
|
||||
def cbor(obj):
|
||||
if isinstance(obj, int):
|
||||
if obj >= 0:
|
||||
return _cbor_int(0, obj)
|
||||
return _cbor_int(1, -1 - obj)
|
||||
if isinstance(obj, bytes):
|
||||
return _cbor_int(2, len(obj)) + obj
|
||||
if isinstance(obj, str):
|
||||
raw = obj.encode()
|
||||
return _cbor_int(3, len(raw)) + raw
|
||||
if isinstance(obj, list):
|
||||
return _cbor_int(4, len(obj)) + b"".join(cbor(x) for x in obj)
|
||||
if isinstance(obj, dict):
|
||||
return _cbor_int(5, len(obj)) + b"".join(cbor(k) + cbor(v) for k, v in obj.items())
|
||||
if obj is None:
|
||||
return b"\xf6"
|
||||
if obj is False:
|
||||
return b"\xf4"
|
||||
if obj is True:
|
||||
return b"\xf5"
|
||||
raise TypeError(obj)
|
||||
|
||||
|
||||
def _cbor_int(major, value):
|
||||
prefix = major << 5
|
||||
if value < 24:
|
||||
return bytes([prefix | value])
|
||||
if value < 256:
|
||||
return bytes([prefix | 24, value])
|
||||
if value < 65536:
|
||||
return bytes([prefix | 25]) + value.to_bytes(2, "big")
|
||||
return bytes([prefix | 26]) + value.to_bytes(4, "big")
|
||||
|
||||
|
||||
class FakeHeaders(dict):
|
||||
def get(self, key, default=None):
|
||||
return super().get(key, default)
|
||||
|
||||
|
||||
class FakeHandler:
|
||||
headers = FakeHeaders({"Host": "localhost:8787"})
|
||||
|
||||
|
||||
def _set_paths(monkeypatch, tmp_path):
|
||||
import api.passkeys as passkeys
|
||||
monkeypatch.setattr(passkeys, "_CREDENTIALS_FILE", tmp_path / "passkeys.json")
|
||||
monkeypatch.setattr(passkeys, "_CHALLENGES_FILE", tmp_path / ".passkey_challenges.json")
|
||||
return passkeys
|
||||
|
||||
|
||||
def _client_data(kind, challenge, origin="http://localhost:8787"):
|
||||
raw = json.dumps({"type": kind, "challenge": challenge, "origin": origin}).encode()
|
||||
return raw, b64u(raw)
|
||||
|
||||
|
||||
def test_passkey_registration_stores_public_credential_metadata(monkeypatch, tmp_path):
|
||||
passkeys = _set_paths(monkeypatch, tmp_path)
|
||||
key = ec.generate_private_key(ec.SECP256R1())
|
||||
pub = key.public_key().public_numbers()
|
||||
credential_id = b"credential-1"
|
||||
cose_key = {1: 2, 3: -7, -1: 1, -2: pub.x.to_bytes(32, "big"), -3: pub.y.to_bytes(32, "big")}
|
||||
opts = passkeys.registration_options(FakeHandler())
|
||||
client_raw, client_b64 = _client_data("webauthn.create", opts["challenge"])
|
||||
auth_data = (
|
||||
hashlib.sha256(opts["rp"]["id"].encode()).digest()
|
||||
+ bytes([0x41])
|
||||
+ (1).to_bytes(4, "big")
|
||||
+ (b"\0" * 16)
|
||||
+ len(credential_id).to_bytes(2, "big")
|
||||
+ credential_id
|
||||
+ cbor(cose_key)
|
||||
)
|
||||
attestation = cbor({"fmt": "none", "authData": auth_data, "attStmt": {}})
|
||||
|
||||
result = passkeys.finish_registration({
|
||||
"label": "MacBook Touch ID",
|
||||
"response": {"clientDataJSON": client_b64, "attestationObject": b64u(attestation)},
|
||||
}, FakeHandler())
|
||||
|
||||
assert result["ok"] is True
|
||||
creds = passkeys.registered_credentials()
|
||||
assert creds == [{
|
||||
"id": b64u(credential_id),
|
||||
"label": "MacBook Touch ID",
|
||||
"created_at": creds[0]["created_at"],
|
||||
"last_used_at": None,
|
||||
"sign_count": 1,
|
||||
}]
|
||||
assert "public_key_pem" not in creds[0]
|
||||
|
||||
|
||||
def test_passkey_login_verifies_signature_and_updates_usage(monkeypatch, tmp_path):
|
||||
passkeys = _set_paths(monkeypatch, tmp_path)
|
||||
key = ec.generate_private_key(ec.SECP256R1())
|
||||
pub = key.public_key().public_numbers()
|
||||
credential_id = b"credential-2"
|
||||
cose_key = {1: 2, 3: -7, -1: 1, -2: pub.x.to_bytes(32, "big"), -3: pub.y.to_bytes(32, "big")}
|
||||
|
||||
reg_opts = passkeys.registration_options(FakeHandler())
|
||||
_raw, client_b64 = _client_data("webauthn.create", reg_opts["challenge"])
|
||||
reg_auth_data = hashlib.sha256(reg_opts["rp"]["id"].encode()).digest() + bytes([0x41]) + (1).to_bytes(4, "big") + (b"\0" * 16) + len(credential_id).to_bytes(2, "big") + credential_id + cbor(cose_key)
|
||||
passkeys.finish_registration({"response": {"clientDataJSON": client_b64, "attestationObject": b64u(cbor({"fmt": "none", "authData": reg_auth_data, "attStmt": {}}))}}, FakeHandler())
|
||||
|
||||
login_opts = passkeys.authentication_options(FakeHandler())
|
||||
client_raw, login_client_b64 = _client_data("webauthn.get", login_opts["challenge"])
|
||||
login_auth_data = hashlib.sha256(login_opts["rpId"].encode()).digest() + bytes([0x01]) + (2).to_bytes(4, "big")
|
||||
signature = key.sign(login_auth_data + hashlib.sha256(client_raw).digest(), ec.ECDSA(hashes.SHA256()))
|
||||
|
||||
result = passkeys.finish_login({
|
||||
"id": b64u(credential_id),
|
||||
"response": {
|
||||
"clientDataJSON": login_client_b64,
|
||||
"authenticatorData": b64u(login_auth_data),
|
||||
"signature": b64u(signature),
|
||||
},
|
||||
}, FakeHandler())
|
||||
|
||||
assert result == {"ok": True, "credential_id": b64u(credential_id)}
|
||||
[cred] = passkeys.registered_credentials()
|
||||
assert cred["sign_count"] == 2
|
||||
assert cred["last_used_at"] is not None
|
||||
|
||||
|
||||
def test_auth_status_reports_passkey_availability_source_contract():
|
||||
src = open("api/routes.py", encoding="utf-8").read()
|
||||
assert '"passkeys_enabled"' in src
|
||||
assert '"passkeys_count"' in src
|
||||
assert 'registered_credentials()' in src
|
||||
|
||||
|
||||
def test_login_page_has_default_hidden_passkey_button_and_script_wiring():
|
||||
routes = open("api/routes.py", encoding="utf-8").read()
|
||||
login_js = open("static/login.js", encoding="utf-8").read()
|
||||
assert 'id="passkey-login"' in routes
|
||||
assert 'style="display:none"' in routes
|
||||
assert "api/auth/passkey/options" in login_js
|
||||
assert "navigator.credentials.get" in login_js
|
||||
Reference in new issue
Block a user