mteehan
6f728cf853
- Raise ValueError on missing request body in auth_logout - Add username check in verify_authentication to prevent credential reuse - Replace N+1 queries in list_users with single JOIN query
410 lines
12 KiB
Python
410 lines
12 KiB
Python
"""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 is_enabled() -> bool:
|
|
"""Return whether WebAuthn is enabled in config."""
|
|
cfg = _get_webauthn_config()
|
|
return cfg.get("enabled", True)
|
|
|
|
|
|
def get_rp_name() -> str:
|
|
"""Return the Relying Party name from config."""
|
|
return _get_webauthn_config().get("rp_name", "Vacuum Wall")
|
|
|
|
|
|
def get_management_domains() -> list[str]:
|
|
"""Return domain names eligible for WebAuthn.
|
|
|
|
Delegates to lib.nginx.get_management_domains() to read the
|
|
live proxy config and discover which domains serve the management UI.
|
|
"""
|
|
from lib.nginx import get_management_domains as _resolve
|
|
|
|
return _resolve()
|
|
|
|
|
|
def is_domain_valid(domain: str) -> bool:
|
|
"""Check if *domain* is eligible for WebAuthn."""
|
|
return domain in get_management_domains()
|
|
|
|
|
|
def check_webauthn_config() -> None:
|
|
"""Validate WebAuthn config at startup."""
|
|
enabled = is_enabled()
|
|
if not enabled:
|
|
logger.info("WebAuthn is disabled in config")
|
|
return
|
|
try:
|
|
domains = get_management_domains()
|
|
if not domains:
|
|
logger.warning(
|
|
"WebAuthn is enabled but no management domains are configured. "
|
|
"Add a domain with backend 'webui' to nginx config, or disable WebAuthn."
|
|
)
|
|
else:
|
|
logger.info("WebAuthn enabled for domains: %s", domains)
|
|
except Exception as exc:
|
|
logger.error("Failed to resolve management domains for WebAuthn: %s", exc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def create_registration_options(
|
|
username: str,
|
|
origin: str,
|
|
rp_id: str,
|
|
) -> dict[str, Any]:
|
|
"""Create WebAuthn registration options for a new credential.
|
|
|
|
Args:
|
|
username: The user registering the credential.
|
|
origin: WebAuthn origin (must match the request origin).
|
|
rp_id: Relying Party ID (must match the request domain).
|
|
|
|
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=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 = "",
|
|
origin: str = "",
|
|
rp_id: 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.
|
|
origin: WebAuthn origin for verification.
|
|
rp_id: Relying Party ID for verification.
|
|
|
|
Returns:
|
|
Dict with ``id``, ``name``, ``transports``, ``sign_count``.
|
|
"""
|
|
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=origin,
|
|
expected_rp_id=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,
|
|
rp_id: str,
|
|
) -> dict[str, Any] | None:
|
|
"""Create authentication options for a user.
|
|
|
|
Args:
|
|
username: The user authenticating.
|
|
rp_id: Relying Party ID (must match the request domain).
|
|
|
|
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=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],
|
|
origin: str = "",
|
|
rp_id: str = "",
|
|
) -> 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``.
|
|
origin: WebAuthn origin for verification.
|
|
rp_id: Relying Party ID for verification.
|
|
|
|
Returns:
|
|
True on successful verification.
|
|
|
|
Raises:
|
|
ValueError: On verification failure.
|
|
"""
|
|
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]
|
|
if cred_row["username"] != username:
|
|
raise ValueError("Credential not found")
|
|
|
|
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=origin,
|
|
expected_rp_id=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(f"Credential {credential_id!r} not found")
|
|
|
|
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}
|