feat(auth): passwordless sign-in with passkeys for local accounts
"Sign in with a passkey" on the login page replaces password and TOTP. User verification is required, so the passkey is two factors on its own; password login stays alongside. Users add, list and remove passkeys under Settings -> Profile after confirming their password. WebAuthn checks (challenge, origin, RP ID, signature, sign count) come from py_webauthn; the browser side uses the native PublicKeyCredential JSON helpers. Off unless WEBAUTHN_ORIGIN is set; its hostname is the RP ID. Local accounts only: a passkey sign-in never asks the directory, so an LDAP user disabled or demoted in AD would keep access and role. Deactivated and permanently locked users are refused. Challenges are 5-minute signed tokens burnt on first use (token_revocation.claim, check and revoke under one lock). The admin "Reset MFA" also removes the user's passkeys. New table passkeys (migration 067). Rules live in app/auth/passkey.py, endpoints under /auth/passkeys in app/routers/auth_passkey.py.
This commit is contained in:
@@ -183,6 +183,13 @@ AUTH_JIT_DEFAULT_ROLE=readonly
|
||||
# CHANGE THIS — losing the key invalidates all stored TOTP secrets.
|
||||
AUTH_PROVIDER_CRYPTO_KEY=CHANGE-ME-FERNET-KEY
|
||||
|
||||
# Passkeys (passwordless sign-in for local accounts). Off while empty.
|
||||
# Exactly the URL users open in the browser, no trailing path; its
|
||||
# hostname becomes the WebAuthn RP ID. Changing the hostname later
|
||||
# invalidates every registered passkey.
|
||||
WEBAUTHN_ORIGIN=
|
||||
# WEBAUTHN_ORIGIN=https://truevuln.company.com
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# LDAPS (Active Directory by default)
|
||||
|
||||
+1
-1
@@ -366,7 +366,7 @@ status). Not yet shipped — placeholder for future ops work.
|
||||
See `README.DEV.md` "Out of scope" sections for the live backlog. Big-ticket
|
||||
items currently parked:
|
||||
|
||||
- WebAuthn / FIDO2 second factor
|
||||
- Passkeys for LDAP users (passwordless passkeys exist for local accounts)
|
||||
- Per-framework URS breakdown (URS_CIS / URS_ISO27001 / URS_NIS2)
|
||||
- SCIM v2 endpoint for IdP-driven provisioning
|
||||
- endoflife.date integration for EOL/EOS package detection
|
||||
|
||||
+56
-3
@@ -398,10 +398,61 @@ Secrets are stored Fernet-encrypted in `users.totp_secret`. Key in env:
|
||||
Next.js needs an explicit `route.ts` for every backend path it forwards.
|
||||
A catch-all at `frontend/app/auth/[...path]/route.ts` covers the new
|
||||
`/auth/mfa/*`, `/auth/providers`, `/auth/oidc/*`, `/auth/saml/*` endpoints
|
||||
in one file. Specific routes (e.g. `/auth/login`, `/auth/me`,
|
||||
and `/auth/passkeys/*` in one file. Specific routes (e.g. `/auth/login`, `/auth/me`,
|
||||
`/auth/users/{id}/reset-password`, `/auth/users/{id}/reset-mfa`) keep
|
||||
priority — Next.js prefers more specific segments over catch-alls.
|
||||
|
||||
## Passkeys (passwordless sign-in)
|
||||
|
||||
**Sign in with a passkey** on the login page replaces password *and* TOTP:
|
||||
user verification (Touch ID / Windows Hello / PIN) is required, so the passkey
|
||||
is two factors on its own. Password login stays alongside. Code:
|
||||
`app/auth/passkey.py` (rules, WebAuthn via `py_webauthn`),
|
||||
`app/routers/auth_passkey.py`, table `passkeys` (migration 067).
|
||||
|
||||
- **Off unless `WEBAUTHN_ORIGIN` is set** — exactly the URL users open
|
||||
(`https://truevuln.company.com`). Its hostname is the WebAuthn RP ID;
|
||||
changing the hostname later invalidates every registered passkey.
|
||||
- **Local accounts only.** A passkey sign-in never asks the directory, so an
|
||||
LDAP user disabled or demoted in AD would keep access and role. SSO users
|
||||
get their factors from the IdP. A user later moved off `local` can no
|
||||
longer sign in with an old passkey.
|
||||
- **Discoverable credentials**: the login needs no username; the browser offers
|
||||
the passkeys it has for this site.
|
||||
- **MFA policy**: a passkey sign-in skips the TOTP step and
|
||||
`mfa_enforced` forced setup (it already is two factors). It refuses
|
||||
deactivated and permanently `locked` users, but not the temporary lockout —
|
||||
that window counts password guesses, which a passkey does not make.
|
||||
- **Challenges** are 5-minute signed tokens burnt on first use
|
||||
(`token_revocation.claim`), so a captured response cannot be replayed. The
|
||||
burn list is in memory: one worker, as for refresh-token revocation.
|
||||
|
||||
### Self-service flow (UI)
|
||||
|
||||
`/settings` → Profile → **Passkeys** card: **Add passkey** → name + current
|
||||
password → browser prompt. Each passkey shows created / last used and can be
|
||||
removed. Uses the browser's native `PublicKeyCredential.parse…FromJSON` /
|
||||
`toJSON()`; browsers without them don't see the button.
|
||||
|
||||
### Admin reset (lost or stolen device)
|
||||
|
||||
The existing **Reset MFA** button also removes all of the user's passkeys.
|
||||
Sessions already open stay valid until their tokens expire.
|
||||
|
||||
### Endpoints
|
||||
|
||||
| Endpoint | Auth | Purpose |
|
||||
|---|---|---|
|
||||
| `POST /auth/passkeys/register/options` | self | Verify current password → creation options + challenge token |
|
||||
| `POST /auth/passkeys/register` | self | Verify attestation, store passkey |
|
||||
| `GET /auth/passkeys` | self | List own passkeys |
|
||||
| `DELETE /auth/passkeys/{id}` | self | Remove own passkey |
|
||||
| `POST /auth/passkeys/login/options` | public | Request options + challenge token |
|
||||
| `POST /auth/passkeys/login` | public | Verify assertion → full session (cookies + tokens) |
|
||||
|
||||
Audit: sign-in as `LOGIN_SUCCESS` / `LOGIN_FAILED` ("via passkey"), add /
|
||||
remove as `MFA_ENABLED` / `MFA_DISABLED` with "Passkey …" in the text.
|
||||
|
||||
## Just-In-Time provisioning + auto-link
|
||||
|
||||
On first SSO/LDAP login, the orchestrator:
|
||||
@@ -466,7 +517,8 @@ Endpoint config still lives in env (restart required for those changes).
|
||||
| `POST` | `/auth/mfa/activate` | self | Confirm enrolment |
|
||||
| `POST` | `/auth/mfa/disable` | self | Disable enrolment |
|
||||
| `POST` | `/auth/mfa/verify` | login flow | Second factor after `/auth/login` |
|
||||
| `POST` | `/auth/users/{user_id}/reset-mfa` | admin | Force-reset MFA for any user (lost-device escape hatch) |
|
||||
| `POST` | `/auth/users/{user_id}/reset-mfa` | admin | Force-reset MFA and passkeys for any user (lost-device escape hatch) |
|
||||
| `*` | `/auth/passkeys/*` | self / public | Passkey enrolment and passwordless sign-in (see Passkeys) |
|
||||
| `GET` | `/auth/oidc/login` | public | Start OIDC SP-initiated flow |
|
||||
| `GET` | `/auth/oidc/callback` | IdP-redirect | Complete OIDC flow |
|
||||
| `GET` | `/auth/saml/login` | public | Start SAML SP-initiated flow |
|
||||
@@ -2389,7 +2441,8 @@ matched by a second scanner post-creation.
|
||||
|
||||
- LDAP password change flow (currently read-only — users change pw in AD)
|
||||
- SCIM v2 endpoint for IdP-driven provisioning (push instead of JIT pull)
|
||||
- WebAuthn / FIDO2 second factor as alternative to TOTP
|
||||
- WebAuthn / FIDO2 as second factor after the password (passwordless passkeys exist)
|
||||
- Passkeys for LDAP users — needs a directory check (disabled / group change) at sign-in
|
||||
- Risk-based MFA (require TOTP only from unknown IP / unknown device)
|
||||
- LDAP referral chasing across multiple forests
|
||||
- Encrypted SAML assertions (currently only signed)
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
"""Add passkeys for passwordless sign-in
|
||||
|
||||
Revision ID: 067
|
||||
Revises: 066
|
||||
Create Date: 2026-11-03 10:00:00.000000
|
||||
|
||||
One row per WebAuthn discoverable credential (app/routers/auth_passkey.py).
|
||||
Rows go with their user (ON DELETE CASCADE). Idempotent, same shape as 035.
|
||||
"""
|
||||
from alembic import op
|
||||
|
||||
|
||||
revision = "067"
|
||||
down_revision = "066"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute("""
|
||||
CREATE TABLE IF NOT EXISTS passkeys (
|
||||
id SERIAL PRIMARY KEY,
|
||||
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
credential_id BYTEA NOT NULL UNIQUE,
|
||||
public_key BYTEA NOT NULL,
|
||||
sign_count INTEGER NOT NULL DEFAULT 0,
|
||||
name VARCHAR(100) NOT NULL,
|
||||
created_at TIMESTAMP NOT NULL DEFAULT now(),
|
||||
last_used_at TIMESTAMP
|
||||
);
|
||||
""")
|
||||
op.execute("CREATE INDEX IF NOT EXISTS ix_passkeys_user_id ON passkeys (user_id);")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute("DROP TABLE IF EXISTS passkeys;")
|
||||
@@ -42,6 +42,7 @@ TOKEN_TYPE_ACCESS = "access"
|
||||
TOKEN_TYPE_REFRESH = "refresh"
|
||||
TOKEN_TYPE_MFA_CHALLENGE = "mfa_challenge"
|
||||
TOKEN_TYPE_MFA_SETUP = "mfa_setup"
|
||||
TOKEN_TYPE_PASSKEY_CHALLENGE = "passkey_challenge"
|
||||
|
||||
# Password Hashing (bcrypt mit 12 Rounds)
|
||||
BCRYPT_ROUNDS = 12
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
"""
|
||||
Passkeys (WebAuthn discoverable credentials) for passwordless sign-in.
|
||||
|
||||
A passkey replaces password AND TOTP: user verification (biometric / PIN on
|
||||
the device) is required, so it is two factors on its own. Sign-in therefore
|
||||
skips the TOTP step and `mfa_policy.requires_setup`, and honours only the
|
||||
permanent lock — the temporary lockout counts password guesses, which a
|
||||
passkey does not make.
|
||||
|
||||
Off unless WEBAUTHN_ORIGIN is set (the URL users open, e.g.
|
||||
https://truevuln.company.com); the RP ID is its hostname.
|
||||
|
||||
LOCAL accounts only. A passkey sign-in never asks the directory, so an LDAP
|
||||
user disabled or demoted in AD would keep access and role; SSO users get
|
||||
their factors from the IdP.
|
||||
|
||||
Challenges travel as short-lived signed tokens and are burnt on first use
|
||||
(token_revocation.claim), so a captured response cannot be replayed.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from enum import Enum
|
||||
from typing import Optional
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
from webauthn import (
|
||||
generate_authentication_options,
|
||||
generate_registration_options,
|
||||
options_to_json,
|
||||
verify_authentication_response,
|
||||
verify_registration_response,
|
||||
)
|
||||
from webauthn.helpers import base64url_to_bytes, bytes_to_base64url, parse_authentication_credential_json
|
||||
from webauthn.helpers.exceptions import WebAuthnException
|
||||
from webauthn.helpers.structs import (
|
||||
AuthenticatorSelectionCriteria,
|
||||
PublicKeyCredentialDescriptor,
|
||||
ResidentKeyRequirement,
|
||||
UserVerificationRequirement,
|
||||
)
|
||||
|
||||
from app.auth.jwt_handler import TOKEN_TYPE_PASSKEY_CHALLENGE, create_access_token, decode_token
|
||||
from app.auth.token_revocation import claim
|
||||
from app.models.passkey import Passkey
|
||||
from app.models.user import AuthProvider, User
|
||||
|
||||
|
||||
class Purpose(str, Enum):
|
||||
REGISTER = "register"
|
||||
LOGIN = "login"
|
||||
|
||||
|
||||
class PasskeyError(Exception):
|
||||
"""`str(e)` is for the audit log; never show it to an unauthenticated client."""
|
||||
|
||||
|
||||
class ChallengeExpired(PasskeyError):
|
||||
pass
|
||||
|
||||
|
||||
def enabled() -> bool:
|
||||
return bool(os.getenv("WEBAUTHN_ORIGIN", "").strip())
|
||||
|
||||
|
||||
def relying_party() -> tuple[str, str]:
|
||||
"""(RP ID, origin). Call only when enabled()."""
|
||||
origin = os.getenv("WEBAUTHN_ORIGIN", "").strip().rstrip("/")
|
||||
return urlparse(origin).hostname, origin
|
||||
|
||||
|
||||
def eligible(user: User) -> bool:
|
||||
return user.auth_provider == AuthProvider.LOCAL
|
||||
|
||||
|
||||
def _issue(challenge: bytes, purpose: Purpose, username: str = "") -> str:
|
||||
return create_access_token(
|
||||
{"sub": username, "purpose": purpose.value, "challenge": bytes_to_base64url(challenge)},
|
||||
expires_delta=timedelta(minutes=5),
|
||||
token_type=TOKEN_TYPE_PASSKEY_CHALLENGE,
|
||||
)
|
||||
|
||||
|
||||
def _redeem(token: str, purpose: Purpose, username: str = "") -> bytes:
|
||||
"""Challenge bytes from a token issued for `purpose`; burns the token."""
|
||||
claims = decode_token(token)
|
||||
if (
|
||||
not claims
|
||||
or claims.get("type") != TOKEN_TYPE_PASSKEY_CHALLENGE
|
||||
or claims.get("purpose") != purpose.value
|
||||
or claims.get("sub") != username
|
||||
or not claim(claims.get("jti"), claims.get("exp"))
|
||||
):
|
||||
raise ChallengeExpired("challenge expired or already used")
|
||||
return base64url_to_bytes(claims["challenge"])
|
||||
|
||||
|
||||
def registration_options(db: Session, user: User) -> dict:
|
||||
rp_id, _ = relying_party()
|
||||
existing = db.query(Passkey.credential_id).filter(Passkey.user_id == user.id).all()
|
||||
options = generate_registration_options(
|
||||
rp_id=rp_id,
|
||||
rp_name="TrueVuln",
|
||||
user_name=user.username,
|
||||
user_id=str(user.id).encode(),
|
||||
exclude_credentials=[PublicKeyCredentialDescriptor(id=cid) for (cid,) in existing],
|
||||
authenticator_selection=AuthenticatorSelectionCriteria(
|
||||
resident_key=ResidentKeyRequirement.REQUIRED,
|
||||
user_verification=UserVerificationRequirement.REQUIRED,
|
||||
),
|
||||
)
|
||||
return {
|
||||
"options": json.loads(options_to_json(options)),
|
||||
"challenge_token": _issue(options.challenge, Purpose.REGISTER, user.username),
|
||||
}
|
||||
|
||||
|
||||
def register(user: User, challenge_token: str, credential: dict, name: str) -> Passkey:
|
||||
"""Verified, unsaved Passkey row. Raises PasskeyError."""
|
||||
rp_id, origin = relying_party()
|
||||
challenge = _redeem(challenge_token, Purpose.REGISTER, user.username)
|
||||
try:
|
||||
verified = verify_registration_response(
|
||||
credential=credential,
|
||||
expected_challenge=challenge,
|
||||
expected_rp_id=rp_id,
|
||||
expected_origin=origin,
|
||||
require_user_verification=True,
|
||||
)
|
||||
except (WebAuthnException, ValueError) as e:
|
||||
raise PasskeyError(str(e)[:200]) from e
|
||||
return Passkey(
|
||||
user_id=user.id,
|
||||
credential_id=verified.credential_id,
|
||||
public_key=verified.credential_public_key,
|
||||
sign_count=verified.sign_count,
|
||||
name=name.strip()[:100] or "Passkey",
|
||||
)
|
||||
|
||||
|
||||
def authentication_options() -> dict:
|
||||
rp_id, _ = relying_party()
|
||||
# No allowCredentials: the browser offers whichever passkey the user has
|
||||
# for this site, so nobody has to type a username first.
|
||||
options = generate_authentication_options(rp_id=rp_id, user_verification=UserVerificationRequirement.REQUIRED)
|
||||
return {
|
||||
"options": json.loads(options_to_json(options)),
|
||||
"challenge_token": _issue(options.challenge, Purpose.LOGIN),
|
||||
}
|
||||
|
||||
|
||||
def sign_in(db: Session, challenge_token: str, credential: dict) -> tuple[Passkey, User]:
|
||||
"""
|
||||
The passkey and its user after a verified assertion; bumps sign count and
|
||||
last use (caller commits). Raises PasskeyError, with `.user` set once the
|
||||
passkey is known.
|
||||
"""
|
||||
rp_id, origin = relying_party()
|
||||
challenge = _redeem(challenge_token, Purpose.LOGIN)
|
||||
user: Optional[User] = None
|
||||
try:
|
||||
parsed = parse_authentication_credential_json(credential)
|
||||
row = db.query(Passkey).filter(Passkey.credential_id == parsed.raw_id).first()
|
||||
user = db.query(User).filter(User.id == row.user_id).first() if row else None
|
||||
if not user or not user.is_active or user.locked or not eligible(user):
|
||||
raise WebAuthnException("unknown passkey or user cannot sign in")
|
||||
verified = verify_authentication_response(
|
||||
credential=parsed,
|
||||
expected_challenge=challenge,
|
||||
expected_rp_id=rp_id,
|
||||
expected_origin=origin,
|
||||
credential_public_key=row.public_key,
|
||||
credential_current_sign_count=row.sign_count,
|
||||
require_user_verification=True,
|
||||
)
|
||||
except (WebAuthnException, ValueError) as e: # ValueError: rawId not base64url
|
||||
err = PasskeyError(str(e)[:200]) # may quote the client's origin
|
||||
err.user = user
|
||||
raise err from e
|
||||
row.sign_count = verified.new_sign_count
|
||||
row.last_used_at = datetime.now()
|
||||
return row, user
|
||||
@@ -27,6 +27,18 @@ def revoke(jti: str, exp_epoch: Optional[float]) -> None:
|
||||
_revoked[jti] = expiry
|
||||
|
||||
|
||||
def claim(jti: str, exp_epoch: Optional[float]) -> bool:
|
||||
"""Revoke `jti` and return True, or False if it was already revoked —
|
||||
check and revoke under one lock, so a one-shot token is used once."""
|
||||
if not jti or is_revoked(jti):
|
||||
return False
|
||||
with _lock:
|
||||
if jti in _revoked:
|
||||
return False
|
||||
_revoked[jti] = exp_epoch if exp_epoch is not None else time.time() + 7 * 24 * 3600
|
||||
return True
|
||||
|
||||
|
||||
def is_revoked(jti: str) -> bool:
|
||||
"""Return True if the JTI was revoked and the revocation has not expired."""
|
||||
if not jti:
|
||||
|
||||
+2
-1
@@ -29,7 +29,7 @@ from slowapi.errors import RateLimitExceeded
|
||||
|
||||
|
||||
|
||||
from app.routers import auth, auth_admin, vulnerabilities, assets, policies, scans, settings, notifications, groups, reports, audit, nessus, compliance, intune, vcenter, igel, netdisco, dockhand, xenserver, advisories, trust_store, https_certificate
|
||||
from app.routers import auth, auth_admin, auth_passkey, vulnerabilities, assets, policies, scans, settings, notifications, groups, reports, audit, nessus, compliance, intune, vcenter, igel, netdisco, dockhand, xenserver, advisories, trust_store, https_certificate
|
||||
try:
|
||||
from app.routers import auth_oidc
|
||||
_HAS_OIDC = True
|
||||
@@ -362,6 +362,7 @@ def root(request: Request):
|
||||
|
||||
app.include_router(auth.router)
|
||||
app.include_router(auth_admin.router)
|
||||
app.include_router(auth_passkey.router)
|
||||
if _HAS_OIDC and auth_oidc is not None:
|
||||
app.include_router(auth_oidc.router)
|
||||
if _HAS_SAML and auth_saml is not None:
|
||||
|
||||
@@ -3,6 +3,7 @@ Database Models für VulnManager
|
||||
"""
|
||||
from app.models.group import Group, user_groups # Must be imported before User
|
||||
from app.models.user import User
|
||||
from app.models.passkey import Passkey
|
||||
from app.models.asset import Asset
|
||||
from app.models.vulnerability import Vulnerability
|
||||
from app.models.vulnerability_package import VulnerabilityPackage
|
||||
@@ -24,6 +25,7 @@ from app.models.compliance import (
|
||||
__all__ = [
|
||||
"Group",
|
||||
"User",
|
||||
"Passkey",
|
||||
"Asset",
|
||||
"Vulnerability",
|
||||
"VulnerabilityPackage",
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
"""
|
||||
Passkey (WebAuthn discoverable credential) for passwordless sign-in.
|
||||
"""
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import Column, DateTime, ForeignKey, Integer, LargeBinary, String
|
||||
|
||||
from app.models.base import Base
|
||||
|
||||
|
||||
class Passkey(Base):
|
||||
__tablename__ = "passkeys"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
|
||||
credential_id = Column(LargeBinary, nullable=False, unique=True)
|
||||
public_key = Column(LargeBinary, nullable=False) # COSE, as the authenticator sent it
|
||||
sign_count = Column(Integer, nullable=False, default=0)
|
||||
name = Column(String(100), nullable=False)
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.now)
|
||||
last_used_at = Column(DateTime, nullable=True)
|
||||
+11
-5
@@ -15,6 +15,7 @@ from pydantic import BaseModel, EmailStr
|
||||
|
||||
from app.database import get_db
|
||||
from app.models.user import User, UserRole
|
||||
from app.models.passkey import Passkey
|
||||
from app.models.audit_log import AuditLog, AuditEventType
|
||||
from app.auth.jwt_handler import (
|
||||
hash_password,
|
||||
@@ -33,6 +34,7 @@ from app.services.audit_trail import record_change
|
||||
from app.auth.orchestrator import AuthOrchestrator
|
||||
from app.auth.strategies.base import AuthError
|
||||
from app.auth import totp as totp_helper
|
||||
from app.auth.passkey import enabled as passkeys_enabled
|
||||
from app.auth.token_revocation import revoke as revoke_token
|
||||
from app.models.user import AuthProvider
|
||||
from app.rate_limiter import limiter
|
||||
@@ -416,6 +418,8 @@ def list_auth_providers():
|
||||
"""
|
||||
enabled = [p.strip().lower() for p in os.getenv("AUTH_PROVIDERS", "local").split(",") if p.strip()]
|
||||
return {
|
||||
# Passwordless sign-in button (app/auth/passkey.py)
|
||||
"passkeys": passkeys_enabled(),
|
||||
"providers": [
|
||||
{
|
||||
"name": p,
|
||||
@@ -972,12 +976,12 @@ def admin_reset_mfa(
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Admin force-reset of a user's TOTP/MFA enrolment.
|
||||
Admin force-reset of a user's TOTP/MFA enrolment and passkeys.
|
||||
|
||||
Use when a user lost their authenticator device and can no longer
|
||||
complete the second-factor step at login. Clears both totp_enabled
|
||||
and totp_secret — the user must re-enrol via /auth/mfa/setup on
|
||||
their next login.
|
||||
complete the second-factor step at login, or a device with a passkey
|
||||
was lost or stolen. Clears totp_enabled, totp_secret and every
|
||||
passkey — the user re-enrols via /settings on their next login.
|
||||
"""
|
||||
user = db.query(User).filter(User.id == user_id).first()
|
||||
if not user:
|
||||
@@ -988,12 +992,14 @@ def admin_reset_mfa(
|
||||
had_mfa = bool(user.totp_enabled or user.totp_secret)
|
||||
user.totp_enabled = False
|
||||
user.totp_secret = None
|
||||
passkeys_removed = db.query(Passkey).filter(Passkey.user_id == user.id).delete()
|
||||
db.commit()
|
||||
|
||||
log_audit_event(
|
||||
db,
|
||||
AuditEventType.MFA_DISABLED,
|
||||
f"MFA force-reset for {user.username} by admin {current_user.username}",
|
||||
f"MFA force-reset for {user.username} by admin {current_user.username}"
|
||||
+ (f" ({passkeys_removed} passkey(s) removed)" if passkeys_removed else ""),
|
||||
user_id=current_user.id,
|
||||
request=request,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
"""
|
||||
# Passkey Router
|
||||
#
|
||||
# Passwordless sign-in with a passkey; rules and WebAuthn checks live in
|
||||
# app/auth/passkey.py. 404 on every route while WEBAUTHN_ORIGIN is unset.
|
||||
"""
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.auth import passkey as passkeys
|
||||
from app.auth.dependencies import get_current_user
|
||||
from app.database import get_db
|
||||
from app.models.audit_log import AuditEventType
|
||||
from app.models.passkey import Passkey
|
||||
from app.models.user import User
|
||||
from app.rate_limiter import limiter
|
||||
from app.routers.auth import _issue_session_response, _verify_user_password, log_audit_event
|
||||
|
||||
router = APIRouter(prefix="/auth/passkeys", tags=["Authentication"])
|
||||
|
||||
|
||||
class RegisterOptionsRequest(BaseModel):
|
||||
password: str # confirm the current password, same as TOTP setup
|
||||
|
||||
|
||||
class RegisterRequest(BaseModel):
|
||||
challenge_token: str
|
||||
credential: dict
|
||||
name: str = "Passkey"
|
||||
|
||||
|
||||
class LoginRequest(BaseModel):
|
||||
challenge_token: str
|
||||
credential: dict
|
||||
|
||||
|
||||
def _require_enabled():
|
||||
if not passkeys.enabled():
|
||||
raise HTTPException(status_code=404, detail="Passkeys are not enabled")
|
||||
|
||||
|
||||
def _require_eligible(user: User):
|
||||
if not passkeys.eligible(user):
|
||||
raise HTTPException(status_code=403, detail="Passkeys are available for local accounts only")
|
||||
|
||||
|
||||
def _channel(request: Request) -> str:
|
||||
return "Web-UI" if request.headers.get("sec-fetch-mode") else "API"
|
||||
|
||||
|
||||
# ============================================
|
||||
# Enrolment (signed-in user)
|
||||
# ============================================
|
||||
|
||||
@router.post("/register/options", dependencies=[Depends(_require_enabled)])
|
||||
@limiter.limit("10/minute")
|
||||
def register_options(
|
||||
request: Request,
|
||||
payload: RegisterOptionsRequest,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
_require_eligible(current_user)
|
||||
if not _verify_user_password(db, current_user, payload.password):
|
||||
log_audit_event(db, AuditEventType.MFA_FAILED, "Passkey setup denied: bad password",
|
||||
user_id=current_user.id, request=request)
|
||||
raise HTTPException(status_code=403, detail="Invalid password")
|
||||
return passkeys.registration_options(db, current_user)
|
||||
|
||||
|
||||
@router.post("/register", dependencies=[Depends(_require_enabled)])
|
||||
@limiter.limit("10/minute")
|
||||
def register(
|
||||
request: Request,
|
||||
payload: RegisterRequest,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
_require_eligible(current_user)
|
||||
try:
|
||||
row = passkeys.register(current_user, payload.challenge_token, payload.credential, payload.name)
|
||||
except passkeys.PasskeyError as e:
|
||||
raise HTTPException(status_code=400, detail=f"Passkey registration failed: {e}")
|
||||
db.add(row)
|
||||
try:
|
||||
db.commit()
|
||||
except IntegrityError:
|
||||
db.rollback()
|
||||
raise HTTPException(status_code=409, detail="This passkey is already registered")
|
||||
log_audit_event(db, AuditEventType.MFA_ENABLED, f"Passkey '{row.name}' added for {current_user.username}",
|
||||
user_id=current_user.id, request=request)
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
@router.get("")
|
||||
def list_passkeys(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
||||
rows = db.query(Passkey).filter(Passkey.user_id == current_user.id).order_by(Passkey.created_at).all()
|
||||
return [
|
||||
{"id": p.id, "name": p.name, "created_at": p.created_at, "last_used_at": p.last_used_at}
|
||||
for p in rows
|
||||
]
|
||||
|
||||
|
||||
@router.delete("/{passkey_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_passkey(
|
||||
passkey_id: int,
|
||||
request: Request,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
row = db.query(Passkey).filter(Passkey.id == passkey_id, Passkey.user_id == current_user.id).first()
|
||||
if not row:
|
||||
raise HTTPException(status_code=404, detail="Passkey not found")
|
||||
name = row.name
|
||||
db.delete(row)
|
||||
db.commit()
|
||||
log_audit_event(db, AuditEventType.MFA_DISABLED, f"Passkey '{name}' removed for {current_user.username}",
|
||||
user_id=current_user.id, request=request)
|
||||
|
||||
|
||||
# ============================================
|
||||
# Sign-in (public)
|
||||
# ============================================
|
||||
|
||||
@router.post("/login/options", dependencies=[Depends(_require_enabled)])
|
||||
@limiter.limit("20/minute")
|
||||
def login_options(request: Request):
|
||||
return passkeys.authentication_options()
|
||||
|
||||
|
||||
@router.post("/login", dependencies=[Depends(_require_enabled)])
|
||||
@limiter.limit("10/minute")
|
||||
def login(request: Request, payload: LoginRequest, db: Session = Depends(get_db)):
|
||||
try:
|
||||
row, user = passkeys.sign_in(db, payload.challenge_token, payload.credential)
|
||||
except passkeys.ChallengeExpired:
|
||||
raise HTTPException(status_code=400, detail="Passkey challenge expired — please try again.")
|
||||
except passkeys.PasskeyError as e:
|
||||
user = getattr(e, "user", None)
|
||||
log_audit_event(db, AuditEventType.LOGIN_FAILED,
|
||||
f"passkey login failed{f' for {user.username}' if user else ''}: {e} [{_channel(request)}]",
|
||||
user_id=user.id if user else None, request=request)
|
||||
raise HTTPException(status_code=401, detail="Passkey sign-in failed")
|
||||
db.commit()
|
||||
log_audit_event(db, AuditEventType.LOGIN_SUCCESS,
|
||||
f"login OK via passkey '{row.name}' as {user.username} [{_channel(request)}]",
|
||||
user_id=user.id, request=request)
|
||||
return _issue_session_response(user, db)
|
||||
@@ -4,6 +4,7 @@
|
||||
* Covers the multi-provider auth surface:
|
||||
* /auth/mfa/setup, /auth/mfa/activate, /auth/mfa/disable, /auth/mfa/verify
|
||||
* /auth/providers
|
||||
* /auth/passkeys, /auth/passkeys/{id}, /auth/passkeys/register[/options], /auth/passkeys/login[/options]
|
||||
* /auth/oidc/login, /auth/oidc/callback
|
||||
* /auth/saml/login, /auth/saml/acs, /auth/saml/metadata
|
||||
*
|
||||
|
||||
@@ -15,6 +15,7 @@ type Stage = 'credentials' | 'mfa';
|
||||
|
||||
export default function LoginPage() {
|
||||
const [providers, setProviders] = useState<AuthProvider[]>([]);
|
||||
const [passkeys, setPasskeys] = useState(false);
|
||||
const [username, setUsername] = useState('');
|
||||
const [password, setPassword] = useState('');
|
||||
const [mfaCode, setMfaCode] = useState('');
|
||||
@@ -26,7 +27,10 @@ export default function LoginPage() {
|
||||
// Fetch enabled providers so we render the right buttons.
|
||||
useEffect(() => {
|
||||
api.get('/auth/providers')
|
||||
.then(res => setProviders(res.data?.providers || []))
|
||||
.then(res => {
|
||||
setProviders(res.data?.providers || []);
|
||||
setPasskeys(Boolean(res.data?.passkeys) && typeof PublicKeyCredential?.parseRequestOptionsFromJSON === 'function');
|
||||
})
|
||||
.catch(() => setProviders([{ name: 'local', credentials: true, redirect: false, display_name: 'Local account' }]));
|
||||
}, []);
|
||||
|
||||
@@ -74,6 +78,34 @@ export default function LoginPage() {
|
||||
}
|
||||
};
|
||||
|
||||
const handlePasskeyLogin = async () => {
|
||||
setError('');
|
||||
setLoading(true);
|
||||
try {
|
||||
const start = await api.post('/auth/passkeys/login/options');
|
||||
const credential = await navigator.credentials.get({
|
||||
publicKey: PublicKeyCredential.parseRequestOptionsFromJSON(start.data.options),
|
||||
}) as PublicKeyCredential | null;
|
||||
if (!credential) throw new Error('No passkey selected');
|
||||
const response = await api.post('/auth/passkeys/login', {
|
||||
challenge_token: start.data.challenge_token,
|
||||
credential: credential.toJSON(),
|
||||
});
|
||||
if (response.data?.access_token) {
|
||||
window.location.href = '/';
|
||||
return;
|
||||
}
|
||||
setError('Passkey sign-in failed.');
|
||||
} catch (err: any) {
|
||||
// NotAllowedError = the user closed the browser prompt.
|
||||
if (err?.name === 'NotAllowedError') setError('');
|
||||
else if (err.response?.status === 429) setError('Too many login attempts. Please wait a moment.');
|
||||
else setError('Passkey sign-in failed. Use your password, or add a passkey under Settings → Profile.');
|
||||
} finally {
|
||||
setLoading(false);
|
||||
}
|
||||
};
|
||||
|
||||
const handleMfaVerify = async (e: React.FormEvent) => {
|
||||
e.preventDefault();
|
||||
setError('');
|
||||
@@ -202,6 +234,17 @@ export default function LoginPage() {
|
||||
</button>
|
||||
</div>
|
||||
|
||||
{passkeys && (
|
||||
<button
|
||||
type="button"
|
||||
onClick={handlePasskeyLogin}
|
||||
disabled={loading}
|
||||
className="flex w-full justify-center rounded-md border border-gray-300 bg-white px-3 py-1.5 text-sm font-semibold leading-6 text-gray-700 shadow-sm hover:bg-gray-50 disabled:opacity-60 font-mono"
|
||||
>
|
||||
Sign in with a passkey
|
||||
</button>
|
||||
)}
|
||||
|
||||
{credentialProviders.length > 1 && (
|
||||
<p className="text-center text-[11px] text-gray-400 font-mono">
|
||||
Tries {credentialProviders.map(p => p.display_name).join(' → ')}
|
||||
|
||||
@@ -17,6 +17,7 @@ import {
|
||||
import api from '../../lib/api';
|
||||
import { logout } from '../../lib/auth';
|
||||
import MfaCard from '../../components/auth/MfaCard';
|
||||
import PasskeyCard from '../../components/auth/PasskeyCard';
|
||||
|
||||
// One section visible at a time — the page holds ~15 cards and scrolling
|
||||
// through all of them to reach one was the whole problem. `admin` marks the
|
||||
@@ -3079,6 +3080,7 @@ export default function SettingsPage() {
|
||||
{active === 'profile' && (
|
||||
<div className="max-w-xl">
|
||||
<MfaCard />
|
||||
<PasskeyCard />
|
||||
<div className="bg-white border border-gray-200 shadow-sm rounded-sm p-6 mt-4">
|
||||
<h3 className="text-lg font-bold text-gray-900 font-mono mb-4 border-b border-gray-100 pb-2">
|
||||
Your Profile
|
||||
|
||||
@@ -0,0 +1,167 @@
|
||||
"use client";
|
||||
|
||||
import { useEffect, useState } from "react";
|
||||
import api from "../../lib/api";
|
||||
|
||||
type Passkey = {
|
||||
id: number;
|
||||
name: string;
|
||||
created_at: string;
|
||||
last_used_at: string | null;
|
||||
};
|
||||
|
||||
/**
|
||||
* Passkeys card for the Settings → Profile page.
|
||||
*
|
||||
* Add: confirm password → /auth/passkeys/register/options → browser passkey
|
||||
* prompt → /auth/passkeys/register. Hidden unless the server has
|
||||
* WEBAUTHN_ORIGIN set, and only for local accounts (see app/auth/passkey.py).
|
||||
*/
|
||||
export default function PasskeyCard() {
|
||||
const [enabled, setEnabled] = useState(false);
|
||||
const [keys, setKeys] = useState<Passkey[]>([]);
|
||||
const [adding, setAdding] = useState(false);
|
||||
const [password, setPassword] = useState("");
|
||||
const [name, setName] = useState("");
|
||||
const [err, setErr] = useState("");
|
||||
const [busy, setBusy] = useState(false);
|
||||
|
||||
const load = async () => {
|
||||
const res = await api.get<Passkey[]>("/auth/passkeys");
|
||||
setKeys(res.data);
|
||||
};
|
||||
|
||||
useEffect(() => {
|
||||
Promise.all([api.get("/auth/providers"), api.get("/auth/me")])
|
||||
.then(([prov, me]) => {
|
||||
// Local accounts only: a passkey sign-in never asks LDAP/AD,
|
||||
// so directory deactivation would not reach it.
|
||||
const on = Boolean(prov.data?.passkeys)
|
||||
&& me.data?.auth_provider === "local"
|
||||
&& typeof PublicKeyCredential?.parseCreationOptionsFromJSON === "function";
|
||||
setEnabled(on);
|
||||
if (on) load();
|
||||
})
|
||||
.catch(() => setEnabled(false));
|
||||
}, []);
|
||||
|
||||
if (!enabled) return null;
|
||||
|
||||
const reset = () => { setAdding(false); setPassword(""); setName(""); setErr(""); };
|
||||
|
||||
const add = async (e: React.FormEvent) => {
|
||||
e.preventDefault();
|
||||
setBusy(true);
|
||||
setErr("");
|
||||
try {
|
||||
const start = await api.post("/auth/passkeys/register/options", { password });
|
||||
const credential = await navigator.credentials.create({
|
||||
publicKey: PublicKeyCredential.parseCreationOptionsFromJSON(start.data.options),
|
||||
}) as PublicKeyCredential | null;
|
||||
if (!credential) throw new Error("No passkey created");
|
||||
await api.post("/auth/passkeys/register", {
|
||||
challenge_token: start.data.challenge_token,
|
||||
credential: credential.toJSON(),
|
||||
name: name || "Passkey",
|
||||
});
|
||||
await load();
|
||||
reset();
|
||||
} catch (e: any) {
|
||||
// NotAllowedError = the user closed the browser prompt.
|
||||
if (e?.name === "NotAllowedError") setErr("Passkey creation was cancelled.");
|
||||
else setErr(e?.response?.data?.detail || e?.message || "Adding the passkey failed");
|
||||
} finally {
|
||||
setBusy(false);
|
||||
}
|
||||
};
|
||||
|
||||
const remove = async (key: Passkey) => {
|
||||
if (!confirm(`Remove passkey "${key.name}"? You can no longer sign in with it.`)) return;
|
||||
try {
|
||||
await api.delete(`/auth/passkeys/${key.id}`);
|
||||
await load();
|
||||
} catch (e: any) {
|
||||
setErr(e?.response?.data?.detail || "Removing the passkey failed");
|
||||
}
|
||||
};
|
||||
|
||||
return (
|
||||
<div className="bg-white border border-gray-200 shadow-sm rounded-sm p-6 mt-4">
|
||||
<h3 className="text-lg font-bold text-gray-900 font-mono mb-4 border-b border-gray-100 pb-2">
|
||||
Passkeys
|
||||
</h3>
|
||||
<p className="text-sm text-gray-500 font-mono mb-4">
|
||||
Sign in without a password using Touch ID, Windows Hello, your phone or a security key.
|
||||
A passkey replaces password and authenticator code.
|
||||
</p>
|
||||
|
||||
{err && (
|
||||
<div className="text-red-600 text-sm font-mono bg-red-50 border border-red-200 rounded p-2 mb-3">
|
||||
{err}
|
||||
</div>
|
||||
)}
|
||||
|
||||
{keys.length > 0 && (
|
||||
<ul className="divide-y divide-gray-100 border border-gray-200 rounded mb-4">
|
||||
{keys.map(k => (
|
||||
<li key={k.id} className="flex items-center justify-between px-3 py-2">
|
||||
<div className="min-w-0">
|
||||
<div className="text-sm font-mono font-semibold text-gray-900 truncate">{k.name}</div>
|
||||
<div className="text-[11px] font-mono text-gray-400">
|
||||
added {new Date(k.created_at).toLocaleDateString()}
|
||||
{" · "}
|
||||
{k.last_used_at ? `last used ${new Date(k.last_used_at).toLocaleString()}` : "never used"}
|
||||
</div>
|
||||
</div>
|
||||
<button
|
||||
onClick={() => remove(k)}
|
||||
className="rounded-md border border-red-300 bg-white px-2 py-1 text-xs font-semibold text-red-700 hover:bg-red-50 font-mono"
|
||||
>
|
||||
Remove
|
||||
</button>
|
||||
</li>
|
||||
))}
|
||||
</ul>
|
||||
)}
|
||||
|
||||
{!adding ? (
|
||||
<button
|
||||
onClick={() => { reset(); setAdding(true); }}
|
||||
className="rounded-md bg-truevuln-blue px-3 py-1.5 text-sm font-semibold text-white shadow-sm hover:bg-blue-600 font-mono"
|
||||
>
|
||||
Add passkey
|
||||
</button>
|
||||
) : (
|
||||
<form onSubmit={add} className="space-y-3">
|
||||
<label className="block text-sm font-mono text-gray-700">
|
||||
Name (e.g. "MacBook" or "YubiKey")
|
||||
<input
|
||||
type="text"
|
||||
value={name}
|
||||
maxLength={100}
|
||||
onChange={(e) => setName(e.target.value)}
|
||||
className="mt-1 block w-full rounded-md border-0 py-1.5 text-gray-900 ring-1 ring-inset ring-gray-300 focus:ring-2 focus:ring-inset focus:ring-truevuln-blue sm:text-sm font-mono"
|
||||
/>
|
||||
</label>
|
||||
<label className="block text-sm font-mono text-gray-700">
|
||||
Confirm your current password
|
||||
<input
|
||||
type="password"
|
||||
value={password}
|
||||
onChange={(e) => setPassword(e.target.value)}
|
||||
autoComplete="current-password"
|
||||
required
|
||||
className="mt-1 block w-full rounded-md border-0 py-1.5 text-gray-900 ring-1 ring-inset ring-gray-300 focus:ring-2 focus:ring-inset focus:ring-truevuln-blue sm:text-sm font-mono"
|
||||
/>
|
||||
</label>
|
||||
<div className="flex gap-2">
|
||||
<button type="button" onClick={reset} className="rounded-md border border-gray-300 bg-white px-3 py-1.5 text-sm text-gray-700 hover:bg-gray-50 font-mono">Cancel</button>
|
||||
<button type="submit" disabled={busy} className="rounded-md bg-truevuln-blue px-3 py-1.5 text-sm font-semibold text-white hover:bg-blue-600 disabled:opacity-60 font-mono">
|
||||
{busy ? "Waiting for passkey…" : "Create passkey"}
|
||||
</button>
|
||||
</div>
|
||||
</form>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
+2
-1
@@ -63,7 +63,8 @@ api.interceptors.response.use(
|
||||
const isAuthEndpoint =
|
||||
url.includes('/auth/login') ||
|
||||
url.includes('/auth/refresh') ||
|
||||
url.includes('/auth/mfa/');
|
||||
url.includes('/auth/mfa/') ||
|
||||
url.includes('/auth/passkeys/login');
|
||||
if (error.response?.status === 401 && !isAuthEndpoint && !original._retried) {
|
||||
original._retried = true;
|
||||
try {
|
||||
|
||||
@@ -49,6 +49,8 @@ reportlab==4.4.9
|
||||
# ============================================
|
||||
# TOTP / MFA
|
||||
pyotp==2.9.0
|
||||
# Passkeys (WebAuthn): signature/attestation checks; pulls cbor2, pyOpenSSL, pyasn1-modules
|
||||
webauthn==3.0.1
|
||||
# Fernet for encrypting LDAP bind-pw + TOTP secrets at rest
|
||||
cryptography==50.0.0
|
||||
|
||||
|
||||
@@ -221,8 +221,8 @@ def test_user_email_change():
|
||||
|
||||
|
||||
_ROUTERS = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "app", "routers")
|
||||
# Session upkeep, not an action.
|
||||
_NOT_ACTIONS = {"refresh_token"}
|
||||
# Session upkeep / handing out a sign-in challenge, not an action.
|
||||
_NOT_ACTIONS = {"refresh_token", "login_options"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", sorted(f for f in os.listdir(_ROUTERS) if f.endswith(".py")))
|
||||
|
||||
@@ -0,0 +1,264 @@
|
||||
"""
|
||||
Passwordless sign-in with a passkey (WebAuthn, discoverable credential).
|
||||
|
||||
A software authenticator below signs with a real P-256 key, so every test
|
||||
goes through py_webauthn's full verification (challenge, origin, RP ID,
|
||||
user verification, signature) exactly as a browser login would.
|
||||
"""
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "0" * 64)
|
||||
|
||||
import cbor2 # noqa: E402
|
||||
import pytest # noqa: E402
|
||||
from cryptography.hazmat.primitives import hashes # noqa: E402
|
||||
from cryptography.hazmat.primitives.asymmetric import ec # noqa: E402
|
||||
from fastapi import FastAPI # noqa: E402
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
from sqlalchemy import create_engine # noqa: E402
|
||||
from sqlalchemy.orm import sessionmaker # noqa: E402
|
||||
from sqlalchemy.pool import StaticPool # noqa: E402
|
||||
|
||||
import app.models # noqa: F401,E402 (registers every mapper)
|
||||
from app.auth.jwt_handler import hash_password # noqa: E402
|
||||
from app.database import get_db # noqa: E402
|
||||
from app.models.audit_log import AuditLog # noqa: E402
|
||||
from app.models.base import Base # noqa: E402
|
||||
from app.models.passkey import Passkey # noqa: E402
|
||||
from app.models.user import AuthProvider, User, UserRole # noqa: E402
|
||||
from app.rate_limiter import limiter # noqa: E402
|
||||
from app.routers import auth, auth_passkey # noqa: E402
|
||||
|
||||
PASSWORD = "Str0ng!Passw0rd"
|
||||
ORIGIN = "https://truevuln.example.com"
|
||||
RP_ID = "truevuln.example.com"
|
||||
|
||||
|
||||
def b64u(raw: bytes) -> str:
|
||||
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
|
||||
|
||||
|
||||
class SoftAuthenticator:
|
||||
"""One passkey: P-256 key, user presence + verification always asserted."""
|
||||
|
||||
def __init__(self):
|
||||
self.key = ec.generate_private_key(ec.SECP256R1())
|
||||
self.cred_id = os.urandom(16)
|
||||
self.count = 0
|
||||
|
||||
def _cose_key(self) -> bytes:
|
||||
n = self.key.public_key().public_numbers()
|
||||
return cbor2.dumps({1: 2, 3: -7, -1: 1, -2: n.x.to_bytes(32, "big"), -3: n.y.to_bytes(32, "big")})
|
||||
|
||||
@staticmethod
|
||||
def _client_data(kind: str, challenge: str, origin: str) -> bytes:
|
||||
return json.dumps({"type": kind, "challenge": challenge, "origin": origin}).encode()
|
||||
|
||||
def create(self, options: dict, origin: str = ORIGIN) -> dict:
|
||||
auth_data = (
|
||||
hashlib.sha256(options["rp"]["id"].encode()).digest()
|
||||
+ bytes([0x45]) # UP | UV | AT
|
||||
+ (0).to_bytes(4, "big")
|
||||
+ bytes(16) # AAGUID
|
||||
+ len(self.cred_id).to_bytes(2, "big") + self.cred_id
|
||||
+ self._cose_key()
|
||||
)
|
||||
attestation = cbor2.dumps({"fmt": "none", "attStmt": {}, "authData": auth_data})
|
||||
return {
|
||||
"id": b64u(self.cred_id), "rawId": b64u(self.cred_id), "type": "public-key",
|
||||
"response": {
|
||||
"clientDataJSON": b64u(self._client_data("webauthn.create", options["challenge"], origin)),
|
||||
"attestationObject": b64u(attestation),
|
||||
},
|
||||
}
|
||||
|
||||
def get(self, options: dict, origin: str = ORIGIN, rp_id: str = RP_ID) -> dict:
|
||||
self.count += 1
|
||||
auth_data = hashlib.sha256(rp_id.encode()).digest() + bytes([0x05]) + self.count.to_bytes(4, "big")
|
||||
client_data = self._client_data("webauthn.get", options["challenge"], origin)
|
||||
signature = self.key.sign(auth_data + hashlib.sha256(client_data).digest(), ec.ECDSA(hashes.SHA256()))
|
||||
return {
|
||||
"id": b64u(self.cred_id), "rawId": b64u(self.cred_id), "type": "public-key",
|
||||
"response": {
|
||||
"clientDataJSON": b64u(client_data),
|
||||
"authenticatorData": b64u(auth_data),
|
||||
"signature": b64u(signature),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(monkeypatch):
|
||||
monkeypatch.setenv("WEBAUTHN_ORIGIN", ORIGIN)
|
||||
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
||||
Base.metadata.create_all(engine)
|
||||
db = sessionmaker(bind=engine)()
|
||||
db.add(User(username="alice", email="a@example.com", role=UserRole.EDITOR,
|
||||
is_active=True, password_hash=hash_password(PASSWORD)))
|
||||
db.commit()
|
||||
app = FastAPI()
|
||||
app.state.limiter = limiter
|
||||
limiter.reset()
|
||||
app.include_router(auth.router)
|
||||
app.include_router(auth_passkey.router)
|
||||
app.dependency_overrides[get_db] = lambda: db
|
||||
return TestClient(app), db
|
||||
|
||||
|
||||
def _bearer(client) -> dict:
|
||||
token = client.post("/auth/login", json={"username": "alice", "password": PASSWORD}).json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
def _register(client, key: SoftAuthenticator, name="Laptop") -> dict:
|
||||
headers = _bearer(client)
|
||||
start = client.post("/auth/passkeys/register/options", json={"password": PASSWORD}, headers=headers)
|
||||
assert start.status_code == 200, start.text
|
||||
body = start.json()
|
||||
done = client.post("/auth/passkeys/register", headers=headers, json={
|
||||
"challenge_token": body["challenge_token"], "credential": key.create(body["options"]), "name": name,
|
||||
})
|
||||
assert done.status_code == 200, done.text
|
||||
return headers
|
||||
|
||||
|
||||
def _login(client, key: SoftAuthenticator, **kw):
|
||||
body = client.post("/auth/passkeys/login/options").json()
|
||||
return client.post("/auth/passkeys/login", json={
|
||||
"challenge_token": body["challenge_token"], "credential": key.get(body["options"], **kw),
|
||||
})
|
||||
|
||||
|
||||
def test_register_then_sign_in_without_password(env):
|
||||
client, db = env
|
||||
key = SoftAuthenticator()
|
||||
headers = _register(client, key)
|
||||
|
||||
listed = client.get("/auth/passkeys", headers=headers).json()
|
||||
assert [p["name"] for p in listed] == ["Laptop"]
|
||||
|
||||
res = _login(client, key)
|
||||
assert res.status_code == 200, res.text
|
||||
token = res.json()["access_token"]
|
||||
assert res.json()["user"]["username"] == "alice"
|
||||
assert client.get("/auth/me", headers={"Authorization": f"Bearer {token}"}).json()["username"] == "alice"
|
||||
last = db.query(AuditLog).order_by(AuditLog.id.desc()).first()
|
||||
assert "via passkey" in last.event_description
|
||||
|
||||
|
||||
def test_login_options_require_user_verification_and_a_discoverable_key(env):
|
||||
client, _ = env
|
||||
options = client.post("/auth/passkeys/login/options").json()["options"]
|
||||
assert options["rpId"] == RP_ID
|
||||
assert options["userVerification"] == "required"
|
||||
assert not options.get("allowCredentials")
|
||||
|
||||
|
||||
def test_registration_needs_the_current_password(env):
|
||||
client, _ = env
|
||||
res = client.post("/auth/passkeys/register/options", json={"password": "wrong"}, headers=_bearer(client))
|
||||
assert res.status_code == 403
|
||||
|
||||
|
||||
def test_a_challenge_is_good_for_one_sign_in_only(env):
|
||||
client, _ = env
|
||||
key = SoftAuthenticator()
|
||||
_register(client, key)
|
||||
body = client.post("/auth/passkeys/login/options").json()
|
||||
payload = {"challenge_token": body["challenge_token"], "credential": key.get(body["options"])}
|
||||
assert client.post("/auth/passkeys/login", json=payload).status_code == 200
|
||||
assert client.post("/auth/passkeys/login", json=payload).status_code == 400
|
||||
|
||||
|
||||
def test_assertion_from_another_origin_is_rejected(env):
|
||||
client, _ = env
|
||||
key = SoftAuthenticator()
|
||||
_register(client, key)
|
||||
assert _login(client, key, origin="https://evil.example.net").status_code == 401
|
||||
|
||||
|
||||
def test_unknown_passkey_is_rejected(env):
|
||||
client, _ = env
|
||||
assert _login(client, SoftAuthenticator()).status_code == 401
|
||||
|
||||
|
||||
def test_deactivated_or_locked_user_cannot_sign_in(env):
|
||||
client, db = env
|
||||
key = SoftAuthenticator()
|
||||
_register(client, key)
|
||||
user = db.query(User).filter_by(username="alice").one()
|
||||
|
||||
user.is_active = False
|
||||
db.commit()
|
||||
assert _login(client, key).status_code == 401
|
||||
|
||||
user.is_active, user.locked = True, True
|
||||
db.commit()
|
||||
assert _login(client, key).status_code == 401
|
||||
|
||||
|
||||
def test_deleted_passkey_no_longer_signs_in(env):
|
||||
client, _ = env
|
||||
key = SoftAuthenticator()
|
||||
headers = _register(client, key)
|
||||
pid = client.get("/auth/passkeys", headers=headers).json()[0]["id"]
|
||||
assert client.delete(f"/auth/passkeys/{pid}", headers=headers).status_code == 204
|
||||
assert client.get("/auth/passkeys", headers=headers).json() == []
|
||||
assert _login(client, key).status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider", [AuthProvider.LDAP, AuthProvider.OIDC, AuthProvider.SAML])
|
||||
def test_only_local_accounts_enrol(env, provider):
|
||||
# LDAP too: a passkey sign-in never asks the directory, so a user
|
||||
# disabled or demoted in AD would keep their access and role.
|
||||
client, db = env
|
||||
headers = _bearer(client)
|
||||
db.query(User).filter_by(username="alice").one().auth_provider = provider
|
||||
db.commit()
|
||||
res = client.post("/auth/passkeys/register/options", json={"password": PASSWORD}, headers=headers)
|
||||
assert res.status_code == 403
|
||||
|
||||
|
||||
def test_account_moved_to_ldap_no_longer_signs_in_with_its_passkey(env):
|
||||
client, db = env
|
||||
key = SoftAuthenticator()
|
||||
_register(client, key)
|
||||
db.query(User).filter_by(username="alice").one().auth_provider = AuthProvider.LDAP
|
||||
db.commit()
|
||||
assert _login(client, key).status_code == 401
|
||||
|
||||
|
||||
def test_cannot_delete_someone_elses_passkey(env):
|
||||
client, db = env
|
||||
_register(client, SoftAuthenticator())
|
||||
db.add(User(username="bob", email="b@example.com", role=UserRole.READONLY,
|
||||
is_active=True, password_hash=hash_password(PASSWORD)))
|
||||
db.commit()
|
||||
bob = {"Authorization": "Bearer " + client.post(
|
||||
"/auth/login", json={"username": "bob", "password": PASSWORD}).json()["access_token"]}
|
||||
pid = db.query(Passkey).one().id
|
||||
assert client.delete(f"/auth/passkeys/{pid}", headers=bob).status_code == 404
|
||||
assert db.query(Passkey).count() == 1
|
||||
|
||||
|
||||
def test_admin_mfa_reset_also_removes_passkeys(env):
|
||||
# Lost or stolen device: the admin escape hatch has to cut passkeys off too.
|
||||
client, db = env
|
||||
key = SoftAuthenticator()
|
||||
headers = _register(client, key)
|
||||
alice = db.query(User).filter_by(username="alice").one()
|
||||
alice.role = UserRole.ADMIN
|
||||
db.commit()
|
||||
assert client.post(f"/auth/users/{alice.id}/reset-mfa", headers=headers).status_code == 200
|
||||
assert db.query(Passkey).count() == 0
|
||||
assert _login(client, key).status_code == 401
|
||||
|
||||
|
||||
def test_disabled_without_configured_origin(env, monkeypatch):
|
||||
client, _ = env
|
||||
monkeypatch.delenv("WEBAUTHN_ORIGIN")
|
||||
assert client.get("/auth/providers").json()["passkeys"] is False
|
||||
assert client.post("/auth/passkeys/login/options").status_code == 404
|
||||
@@ -31,17 +31,18 @@ from app.auth.dependencies import get_current_user # noqa: E402
|
||||
from app.database import get_db # noqa: E402
|
||||
from app.models.user import User, UserRole # noqa: E402
|
||||
from app.routers import ( # noqa: E402
|
||||
advisories, assets, audit, auth, auth_admin, compliance, groups, https_certificate, igel, intune,
|
||||
advisories, assets, audit, auth, auth_admin, auth_passkey, compliance, groups, https_certificate, igel, intune,
|
||||
nessus, notifications, policies, reports, scans, settings, trust_store, vcenter, vulnerabilities,
|
||||
)
|
||||
|
||||
ROUTERS = (auth, auth_admin, vulnerabilities, nessus, intune, vcenter, igel, advisories, assets,
|
||||
ROUTERS = (auth, auth_admin, auth_passkey, vulnerabilities, nessus, intune, vcenter, igel, advisories, assets,
|
||||
policies, scans, settings, trust_store, https_certificate, notifications, groups, reports, audit, compliance)
|
||||
|
||||
# What a readonly user may GET. Path params are substituted with "1".
|
||||
READONLY_MAY_GET = {
|
||||
"/auth/providers",
|
||||
"/auth/me",
|
||||
"/auth/passkeys", # own passkeys only
|
||||
"/api/v1/vulnerabilities",
|
||||
"/api/v1/vulnerabilities/by-package",
|
||||
"/api/v1/vulnerabilities/ai-history",
|
||||
|
||||
Reference in New Issue
Block a user