feat: add auth subsystem with WebAuthn passkeys support
New modules: lib/auth, lib/auth_users, lib/db, lib/db_sqlite, lib/password, lib/webauthn, daemon/handlers/auth, scripts/bootstrap_auth, tests/test_auth Frontend: webui/api/auth, hoover/components/auth, pages/login, passkeys, users Updates: daemon/iface and server, lib/common and nginx, pyproject.toml deps, install script, server.py, app.js, and websocket/api clients
This commit is contained in:
+355
@@ -0,0 +1,355 @@
|
||||
"""WebAuthn passkey support for Vacuum Wall.
|
||||
|
||||
Uses the Duo Labs webauthn library (v3) to handle the FIDO2/WebAuthn ceremony:
|
||||
registration, authentication, and credential management.
|
||||
|
||||
Credential data is stored in the database webauthn_creds table.
|
||||
Configuration comes from config/auth/config.json (webauthn section).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from webauthn import (
|
||||
generate_authentication_options,
|
||||
generate_registration_options,
|
||||
options_to_json,
|
||||
verify_authentication_response,
|
||||
verify_registration_response,
|
||||
)
|
||||
from webauthn.helpers.cose import COSEAlgorithmIdentifier
|
||||
from webauthn.helpers.structs import (
|
||||
AttestationConveyancePreference,
|
||||
AuthenticatorSelectionCriteria,
|
||||
PublicKeyCredentialDescriptor,
|
||||
ResidentKeyRequirement,
|
||||
UserVerificationRequirement,
|
||||
)
|
||||
|
||||
from lib.auth import AUTH_CONFIG_PATH
|
||||
from lib.common import load_json
|
||||
from lib.db import (
|
||||
Q_DELETE_WEBAUTHN,
|
||||
Q_INSERT_WEBAUTHN,
|
||||
Q_SELECT_WEBAUTHN_COUNTS,
|
||||
Q_SELECT_WEBAUTHN_ID,
|
||||
Q_SELECT_WEBAUTHN_USER,
|
||||
Q_UPDATE_WEBAUTHN_SIGN_COUNT,
|
||||
get_db,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Crypto algorithms we support
|
||||
_SUPPORTED_ALGS = [
|
||||
COSEAlgorithmIdentifier.ECDSA_SHA_256,
|
||||
COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_256,
|
||||
COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_384,
|
||||
COSEAlgorithmIdentifier.ECDSA_SHA_512,
|
||||
COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_512,
|
||||
COSEAlgorithmIdentifier.EDDSA,
|
||||
]
|
||||
|
||||
|
||||
# b64url helpers
|
||||
def b64u_encode(b: bytes) -> str:
|
||||
return base64.urlsafe_b64encode(b).decode("ascii").rstrip("=")
|
||||
|
||||
|
||||
def b64u_decode(s: str) -> bytes:
|
||||
padding = 4 - len(s) % 4
|
||||
if padding != 4:
|
||||
s += "=" * padding
|
||||
return base64.urlsafe_b64decode(s)
|
||||
|
||||
|
||||
def _get_webauthn_config() -> dict[str, Any]:
|
||||
"""Load WebAuthn configuration from auth config."""
|
||||
raw = load_json(AUTH_CONFIG_PATH)
|
||||
return raw.get("webauthn", {})
|
||||
|
||||
|
||||
def get_rp_id() -> str:
|
||||
"""Return the Relying Party ID from config."""
|
||||
return _get_webauthn_config().get("rp_id", "localhost")
|
||||
|
||||
|
||||
def get_rp_name() -> str:
|
||||
"""Return the Relying Party name from config."""
|
||||
return _get_webauthn_config().get("rp_name", "Vacuum Wall")
|
||||
|
||||
|
||||
def get_origin() -> str:
|
||||
"""Return the WebAuthn origin from config."""
|
||||
return _get_webauthn_config().get("origin", "http://localhost")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Registration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def create_registration_options(username: str) -> dict[str, Any]:
|
||||
"""Create WebAuthn registration options for a new credential.
|
||||
|
||||
Returns a dict serializable to JSON, matching the format expected by
|
||||
``navigator.credentials.create()``.
|
||||
"""
|
||||
# Load existing credential IDs to exclude
|
||||
db = get_db()
|
||||
rows = db.query(Q_SELECT_WEBAUTHN_USER, (username,))
|
||||
exclude_credentials = [
|
||||
PublicKeyCredentialDescriptor(id=b64u_decode(row["credential_id"]))
|
||||
for row in rows
|
||||
]
|
||||
|
||||
# Pad username to >= 8 bytes (required for user_id)
|
||||
user_id = username.encode("utf-8")
|
||||
if len(user_id) < 8:
|
||||
user_id = user_id + b"\x00" * (8 - len(user_id))
|
||||
|
||||
options = generate_registration_options(
|
||||
rp_id=get_rp_id(),
|
||||
rp_name=get_rp_name(),
|
||||
user_name=username,
|
||||
user_display_name=username,
|
||||
user_id=user_id,
|
||||
attestation=AttestationConveyancePreference.NONE,
|
||||
authenticator_selection=AuthenticatorSelectionCriteria(
|
||||
resident_key=ResidentKeyRequirement.PREFERRED,
|
||||
user_verification=UserVerificationRequirement.PREFERRED,
|
||||
),
|
||||
supported_pub_key_algs=_SUPPORTED_ALGS,
|
||||
exclude_credentials=exclude_credentials,
|
||||
)
|
||||
|
||||
# Serialize using the library's built-in function
|
||||
return json.loads(options_to_json(options))
|
||||
|
||||
|
||||
def verify_registration(
|
||||
username: str,
|
||||
credential_response: dict[str, Any],
|
||||
registration_options: dict[str, Any],
|
||||
credential_name: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Verify a registration response and persist the credential.
|
||||
|
||||
Args:
|
||||
username: The user registering the credential.
|
||||
credential_response: Browser response from ``credentials.create()``.
|
||||
registration_options: The options dict from ``create_registration_options``.
|
||||
credential_name: Optional human-readable label.
|
||||
|
||||
Returns:
|
||||
Dict with ``id``, ``name``, ``transports``, ``sign_count``.
|
||||
"""
|
||||
expected_origin = get_origin()
|
||||
expected_rp_id = get_rp_id()
|
||||
challenge = b64u_decode(registration_options["challenge"])
|
||||
|
||||
# The library accepts the credential response as a JSON-serializable dict
|
||||
# We pass it directly — the library handles the parsing
|
||||
col = verify_registration_response(
|
||||
credential=credential_response,
|
||||
expected_challenge=challenge,
|
||||
expected_origin=expected_origin,
|
||||
expected_rp_id=expected_rp_id,
|
||||
require_user_verification=False,
|
||||
)
|
||||
|
||||
new_cred = col.credential
|
||||
new_credential_id = b64u_encode(new_cred.id)
|
||||
new_public_key = b64u_encode(new_cred.public_key)
|
||||
new_sign_count = new_cred.sign_count
|
||||
|
||||
transports = []
|
||||
if hasattr(new_cred.response, "transports") and new_cred.response.transports:
|
||||
transports = [str(t) for t in new_cred.response.transports]
|
||||
if not transports:
|
||||
transports_raw = credential_response.get("response", {}).get("transports", [])
|
||||
transports = [
|
||||
t
|
||||
for t in transports_raw
|
||||
if t
|
||||
in (
|
||||
"internal",
|
||||
"hybrid",
|
||||
"nfc",
|
||||
"ble",
|
||||
"usb",
|
||||
"smart-card",
|
||||
)
|
||||
] or ["internal"]
|
||||
|
||||
db = get_db()
|
||||
db.run(
|
||||
Q_INSERT_WEBAUTHN,
|
||||
(
|
||||
username,
|
||||
new_credential_id,
|
||||
new_public_key,
|
||||
new_sign_count,
|
||||
credential_name,
|
||||
json.dumps(transports),
|
||||
),
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"WebAuthn credential registered for %s: %s",
|
||||
username,
|
||||
credential_name or new_credential_id[:16],
|
||||
)
|
||||
|
||||
return {
|
||||
"id": new_credential_id,
|
||||
"name": credential_name,
|
||||
"transports": transports,
|
||||
"sign_count": new_sign_count,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Authentication
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def create_authentication_options(username: str) -> dict[str, Any] | None:
|
||||
"""Create authentication options for a user.
|
||||
|
||||
Returns a dict serializable to JSON (for ``navigator.credentials.get()``),
|
||||
or ``None`` if the user has no registered credentials.
|
||||
"""
|
||||
db = get_db()
|
||||
rows = db.query(Q_SELECT_WEBAUTHN_USER, (username,))
|
||||
if not rows:
|
||||
return None
|
||||
|
||||
allow_credentials = [
|
||||
PublicKeyCredentialDescriptor(id=b64u_decode(row["credential_id"]))
|
||||
for row in rows
|
||||
]
|
||||
|
||||
options = generate_authentication_options(
|
||||
rp_id=get_rp_id(),
|
||||
allow_credentials=allow_credentials,
|
||||
user_verification=UserVerificationRequirement.PREFERRED,
|
||||
)
|
||||
|
||||
return json.loads(options_to_json(options))
|
||||
|
||||
|
||||
def verify_authentication(
|
||||
username: str,
|
||||
assertion_response: dict[str, Any],
|
||||
auth_options: dict[str, Any],
|
||||
) -> bool:
|
||||
"""Verify an authentication assertion.
|
||||
|
||||
Args:
|
||||
username: The user authenticating.
|
||||
assertion_response: Browser response from ``credentials.get()``.
|
||||
auth_options: The options dict from ``create_authentication_options``.
|
||||
|
||||
Returns:
|
||||
True on successful verification.
|
||||
|
||||
Raises:
|
||||
ValueError: On verification failure.
|
||||
"""
|
||||
expected_origin = get_origin()
|
||||
expected_rp_id = get_rp_id()
|
||||
challenge = b64u_decode(auth_options["challenge"])
|
||||
cred_id_str = assertion_response["id"]
|
||||
|
||||
# Load credential from DB
|
||||
db = get_db()
|
||||
cred_rows = db.query(Q_SELECT_WEBAUTHN_ID, (cred_id_str,))
|
||||
if not cred_rows:
|
||||
raise ValueError("Credential not found")
|
||||
|
||||
cred_row = cred_rows[0]
|
||||
public_key = b64u_decode(cred_row["public_key"])
|
||||
old_sign_count = cred_row["sign_count"]
|
||||
|
||||
col = verify_authentication_response(
|
||||
credential=assertion_response,
|
||||
expected_challenge=challenge,
|
||||
expected_origin=expected_origin,
|
||||
expected_rp_id=expected_rp_id,
|
||||
credential_public_key=public_key,
|
||||
credential_current_sign_count=old_sign_count,
|
||||
require_user_verification=False,
|
||||
)
|
||||
|
||||
new_sign_count = col.credential_sign_count
|
||||
if new_sign_count > old_sign_count:
|
||||
db.run(
|
||||
Q_UPDATE_WEBAUTHN_SIGN_COUNT,
|
||||
(new_sign_count, cred_id_str),
|
||||
)
|
||||
|
||||
logger.info("WebAuthn assertion verified for %s", username)
|
||||
return True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Credential management
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def list_credentials(username: str) -> list[dict[str, Any]]:
|
||||
"""List all registered credentials for a user."""
|
||||
db = get_db()
|
||||
rows = db.query(Q_SELECT_WEBAUTHN_USER, (username,))
|
||||
|
||||
result = []
|
||||
for row in rows:
|
||||
transports_str = row.get("transports", "[]")
|
||||
try:
|
||||
transports = json.loads(transports_str)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
transports = []
|
||||
|
||||
result.append(
|
||||
{
|
||||
"id": row["credential_id"],
|
||||
"name": row.get("name") or "",
|
||||
"transports": transports,
|
||||
"sign_count": row.get("sign_count", 0),
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def remove_credential(username: str, credential_id: str) -> bool:
|
||||
"""Remove a credential.
|
||||
|
||||
Raises:
|
||||
ValueError: If credential not found or doesn't belong to user.
|
||||
"""
|
||||
db = get_db()
|
||||
rows = db.query(Q_SELECT_WEBAUTHN_ID, (credential_id,))
|
||||
if not rows:
|
||||
raise ValueError(f"Credential {credential_id!r} not found")
|
||||
if rows[0]["username"] != username:
|
||||
raise ValueError("Credential does not belong to this user")
|
||||
|
||||
db.run(Q_DELETE_WEBAUTHN, (credential_id,))
|
||||
logger.info("WebAuthn credential removed: %s, %s", username, credential_id)
|
||||
return True
|
||||
|
||||
|
||||
def get_all_credential_counts() -> dict[str, int]:
|
||||
"""Return credential counts for all users.
|
||||
|
||||
Returns:
|
||||
Dict mapping usernames to credential counts.
|
||||
"""
|
||||
db = get_db()
|
||||
rows = db.query(Q_SELECT_WEBAUTHN_COUNTS, ())
|
||||
return {row["username"]: row["cred_count"] for row in rows}
|
||||
Reference in New Issue
Block a user