mteehan
e01574c67e
- Make session_id parameter optional in validate_token — only enforced when provided, allowing WebSocket auth which cannot carry custom headers - Remove decode_token round-trip from daemon WS handler - Override WebAuthn registration username from JWT user_ctx to prevent users from registering credentials under another user's account - Frontend no longer sends username for WebAuthn registration - Redirect to login on 401 after token refresh fails for POST requests - Remove redundant auth session check from login page load - Add tests for session_id semantics and WebAuthn ownership guard
404 lines
12 KiB
Python
404 lines
12 KiB
Python
"""Authentication API blueprint.
|
|
|
|
Exposed at /api/auth/* and delegates all operations to vacuum-walld.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from flask import Blueprint, request
|
|
|
|
from daemon.client import delete, get, post
|
|
from daemon.iface import (
|
|
DELETE_AUTH_USER,
|
|
DELETE_AUTH_WEBAUTHN_CREDENTIAL,
|
|
GET_AUTH_SESSION,
|
|
GET_AUTH_USERS,
|
|
GET_AUTH_WEBAUTHN_CAPABLE,
|
|
GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS,
|
|
GET_AUTH_WEBAUTHN_CREDENTIALS,
|
|
POST_AUTH_LOGIN,
|
|
POST_AUTH_LOGOUT,
|
|
POST_AUTH_PASSWORD,
|
|
POST_AUTH_REFRESH,
|
|
POST_AUTH_USER_CREATE,
|
|
POST_AUTH_USER_UPDATE,
|
|
POST_AUTH_WEBAUTHN_AUTHENTICATE_BEGIN,
|
|
POST_AUTH_WEBAUTHN_AUTHENTICATE_FINISH,
|
|
POST_AUTH_WEBAUTHN_REGISTER_BEGIN,
|
|
POST_AUTH_WEBAUTHN_REGISTER_FINISH,
|
|
)
|
|
from webui.api.common import _error, _ok
|
|
|
|
logger = logging.getLogger(__name__)
|
|
bp = Blueprint("auth", __name__)
|
|
|
|
|
|
@bp.route("/login", methods=["POST"])
|
|
def login():
|
|
"""Authenticate user with username and password.
|
|
|
|
Endpoint:
|
|
POST /api/auth/login
|
|
Body:
|
|
{ "username": "admin", "password": "secretpass" }
|
|
Returns:
|
|
{ "tokens": { "access_token": "...", "refresh_token": "..." },
|
|
"user": { "id": 1, "username": "admin" },
|
|
"permissions": { ... } }
|
|
"""
|
|
try:
|
|
body = request.get_json(silent=True) or {}
|
|
body["client_ip"] = request.headers.get("X-Real-IP") or request.remote_addr
|
|
return _ok(post(POST_AUTH_LOGIN, body))
|
|
except Exception as exc:
|
|
logger.error("Login failed: %s", exc)
|
|
return _error(str(exc), 401)
|
|
|
|
|
|
@bp.route("/logout", methods=["POST"])
|
|
def logout():
|
|
"""Invalidate current session by blacklisting access and refresh tokens.
|
|
|
|
Endpoint:
|
|
POST /api/auth/logout
|
|
Body:
|
|
{ "refresh_token": "..." } -- client-provided refresh token
|
|
Returns:
|
|
{ "ok": true }
|
|
"""
|
|
try:
|
|
client_body = request.get_json(silent=True) or {}
|
|
body = {
|
|
**(request._user_ctx or {}),
|
|
"refresh_token": client_body.get("refresh_token"),
|
|
}
|
|
return _ok(post(POST_AUTH_LOGOUT, body))
|
|
except RuntimeError as exc:
|
|
logger.error("Logout failed: %s", exc)
|
|
return _error(str(exc), 500)
|
|
|
|
|
|
@bp.route("/refresh", methods=["POST"])
|
|
def refresh():
|
|
"""Rotate tokens using a refresh token.
|
|
|
|
Endpoint:
|
|
POST /api/auth/refresh
|
|
Body:
|
|
{ "refresh_token": "..." }
|
|
Returns:
|
|
{ "tokens": { ... }, "user": { ... }, "permissions": { ... } }
|
|
"""
|
|
try:
|
|
body = request.get_json(silent=True) or {}
|
|
return _ok(post(POST_AUTH_REFRESH, body))
|
|
except Exception as exc:
|
|
logger.error("Token refresh failed: %s", exc)
|
|
return _error(str(exc), 401)
|
|
|
|
|
|
@bp.route("/session", methods=["GET"])
|
|
def session():
|
|
"""Return current user session info.
|
|
|
|
Endpoint:
|
|
GET /api/auth/session
|
|
Returns:
|
|
{ "user": { ... }, "permissions": { ... } }
|
|
"""
|
|
try:
|
|
return _ok(get(GET_AUTH_SESSION, {**(request._user_ctx or {})}))
|
|
except Exception as exc:
|
|
logger.error("Session check failed: %s", exc)
|
|
return _error(str(exc), 401)
|
|
|
|
|
|
@bp.route("/password", methods=["POST"])
|
|
def change_password():
|
|
"""Change own password.
|
|
|
|
Endpoint:
|
|
POST /api/auth/password
|
|
Body:
|
|
{ "oldPassword": "...", "newPassword": "..." }
|
|
Returns:
|
|
{ "ok": true }
|
|
"""
|
|
try:
|
|
body = request.get_json(silent=True) or {}
|
|
user_ctx = getattr(request, "_user_ctx", None)
|
|
if user_ctx is not None:
|
|
body["username"] = user_ctx["username"]
|
|
return _ok(post(POST_AUTH_PASSWORD, body))
|
|
except Exception as exc:
|
|
logger.error("Password change failed: %s", exc)
|
|
return _error(str(exc), 400)
|
|
|
|
|
|
@bp.route("/users", methods=["GET"])
|
|
def list_users():
|
|
"""List all users.
|
|
|
|
Endpoint:
|
|
GET /api/auth/users
|
|
Returns:
|
|
{ "users": [{ "id": 1, "username": "...", "permissions": { ... }, ... }] }
|
|
"""
|
|
try:
|
|
return _ok(get(GET_AUTH_USERS))
|
|
except RuntimeError as exc:
|
|
logger.error("List users failed: %s", exc)
|
|
return _error(str(exc), 500)
|
|
|
|
|
|
@bp.route("/users", methods=["POST"])
|
|
def create_user():
|
|
"""Create a new user.
|
|
|
|
Endpoint:
|
|
POST /api/auth/users
|
|
Body:
|
|
{ "username": "...", "password": "...", "permissions": { ... } }
|
|
Returns:
|
|
{ "ok": true, "id": ..., "username": "...", "permissions": { ... } }
|
|
"""
|
|
try:
|
|
body = request.get_json(silent=True) or {}
|
|
return _ok(post(POST_AUTH_USER_CREATE, body))
|
|
except Exception as exc:
|
|
logger.error("Create user failed: %s", exc)
|
|
return _error(str(exc), 400)
|
|
|
|
|
|
@bp.route("/users/<username>", methods=["POST"])
|
|
def update_user(username: str):
|
|
"""Update user permissions.
|
|
|
|
Endpoint:
|
|
POST /api/auth/users/<username>
|
|
Body:
|
|
{ "permissions": { ... } }
|
|
Returns:
|
|
{ "ok": true, "id": ..., "username": "..." }
|
|
"""
|
|
try:
|
|
body = {**(request.get_json(silent=True) or {}), "username": username}
|
|
return _ok(post(POST_AUTH_USER_UPDATE, body))
|
|
except Exception as exc:
|
|
err = str(exc)
|
|
status = 404 if "not found" in err.lower() else 400
|
|
logger.error("Update user failed: %s", exc)
|
|
return _error(err, status)
|
|
|
|
|
|
@bp.route("/users/<username>", methods=["DELETE"])
|
|
def delete_user(username: str):
|
|
"""Delete a user.
|
|
|
|
Endpoint:
|
|
DELETE /api/auth/users/<username>
|
|
Returns:
|
|
{ "ok": true }
|
|
"""
|
|
# Prevent self-deletion
|
|
user_ctx = getattr(request, "_user_ctx", None)
|
|
if user_ctx is not None and user_ctx.get("username") == username:
|
|
return _error("Cannot delete your own account", 403)
|
|
try:
|
|
return _ok(delete(DELETE_AUTH_USER, {"username": username}))
|
|
except Exception as exc:
|
|
err = str(exc)
|
|
status = 404 if "not found" in err.lower() else 400
|
|
logger.error("Delete user failed: %s", exc)
|
|
return _error(err, status)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WebAuthn routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _resolve_webauthn_origin() -> tuple[str, str]:
|
|
"""Extract WebAuthn origin and rp_id from the current request.
|
|
|
|
Returns (origin, rp_id) derived from the actual request, falling back
|
|
to config values when the request metadata is unavailable.
|
|
"""
|
|
scheme = request.headers.get("X-Forwarded-Proto", request.scheme)
|
|
host = request.headers.get("X-Forwarded-Host", request.host.split(":")[0])
|
|
origin = f"{scheme}://{host}"
|
|
# rp_id is the registered domain (strip port numbers)
|
|
rp_id = host.split(":")[0]
|
|
return origin, rp_id
|
|
|
|
|
|
@bp.route("/webauthn/capable", methods=["GET"])
|
|
def webauthn_capable():
|
|
"""Check if WebAuthn is available on the current request domain.
|
|
|
|
Endpoint:
|
|
GET /api/auth/webauthn/capable
|
|
Returns:
|
|
{ "enabled": true/false, "rp_id": "...", "rp_name": "...", "origin": "..." }
|
|
or { "enabled": false, "reason": "..." }
|
|
"""
|
|
try:
|
|
origin, rp_id = _resolve_webauthn_origin()
|
|
body = {"webauthn_origin": origin, "webauthn_rp_id": rp_id}
|
|
return _ok(get(GET_AUTH_WEBAUTHN_CAPABLE, body))
|
|
except Exception as exc:
|
|
logger.error("WebAuthn capable check failed: %s", exc)
|
|
return _error(str(exc), 500)
|
|
|
|
|
|
@bp.route("/webauthn/register-begin", methods=["POST"])
|
|
def webauthn_register_begin():
|
|
"""Begin WebAuthn registration.
|
|
|
|
Endpoint:
|
|
POST /api/auth/webauthn/register-begin
|
|
Returns:
|
|
Registration options for navigator.credentials.create()
|
|
"""
|
|
try:
|
|
body = request.get_json(silent=True) or {}
|
|
user_ctx = getattr(request, "_user_ctx", None)
|
|
if user_ctx is not None:
|
|
body["username"] = user_ctx["username"]
|
|
origin, rp_id = _resolve_webauthn_origin()
|
|
body["webauthn_origin"] = origin
|
|
body["webauthn_rp_id"] = rp_id
|
|
return _ok(post(POST_AUTH_WEBAUTHN_REGISTER_BEGIN, body))
|
|
except Exception as exc:
|
|
logger.error("WebAuthn register begin failed: %s", exc)
|
|
return _error(str(exc), 400)
|
|
|
|
|
|
@bp.route("/webauthn/register-finish", methods=["POST"])
|
|
def webauthn_register_finish():
|
|
"""Finish WebAuthn registration.
|
|
|
|
Endpoint:
|
|
POST /api/auth/webauthn/register-finish
|
|
Body:
|
|
{ "credential_response": {...}, "registration_options": {...}, "name": "..." }
|
|
Returns:
|
|
{ "ok": true, "credential": {...} }
|
|
"""
|
|
try:
|
|
body = request.get_json(silent=True) or {}
|
|
user_ctx = getattr(request, "_user_ctx", None)
|
|
if user_ctx is not None:
|
|
body["username"] = user_ctx["username"]
|
|
origin, rp_id = _resolve_webauthn_origin()
|
|
body["webauthn_origin"] = origin
|
|
body["webauthn_rp_id"] = rp_id
|
|
return _ok(post(POST_AUTH_WEBAUTHN_REGISTER_FINISH, body))
|
|
except Exception as exc:
|
|
logger.error("WebAuthn register finish failed: %s", exc)
|
|
return _error(str(exc), 400)
|
|
|
|
|
|
@bp.route("/webauthn/authenticate-begin", methods=["POST"])
|
|
def webauthn_authenticate_begin():
|
|
"""Begin WebAuthn authentication (public endpoint).
|
|
|
|
Endpoint:
|
|
POST /api/auth/webauthn/authenticate-begin
|
|
Body:
|
|
{ "username": "..." }
|
|
Returns:
|
|
Authentication options for navigator.credentials.get()
|
|
or { "no_webauthn": true } if user has no credentials.
|
|
"""
|
|
try:
|
|
body = request.get_json(silent=True) or {}
|
|
_, rp_id = _resolve_webauthn_origin()
|
|
body["webauthn_rp_id"] = rp_id
|
|
return _ok(post(POST_AUTH_WEBAUTHN_AUTHENTICATE_BEGIN, body))
|
|
except Exception as exc:
|
|
logger.error("WebAuthn authenticate begin failed: %s", exc)
|
|
return _error(str(exc), 400)
|
|
|
|
|
|
@bp.route("/webauthn/authenticate-finish", methods=["POST"])
|
|
def webauthn_authenticate_finish():
|
|
"""Finish WebAuthn authentication (public endpoint).
|
|
|
|
Endpoint:
|
|
POST /api/auth/webauthn/authenticate-finish
|
|
Body:
|
|
{ "username": "...", "assertion_response": {...}, "auth_options": {...} }
|
|
Returns:
|
|
{ "tokens": {...}, "user": {...}, "permissions": {...} }
|
|
"""
|
|
try:
|
|
body = request.get_json(silent=True) or {}
|
|
body["client_ip"] = request.headers.get("X-Real-IP") or request.remote_addr
|
|
origin, rp_id = _resolve_webauthn_origin()
|
|
body["webauthn_origin"] = origin
|
|
body["webauthn_rp_id"] = rp_id
|
|
return _ok(post(POST_AUTH_WEBAUTHN_AUTHENTICATE_FINISH, body))
|
|
except Exception as exc:
|
|
logger.error("WebAuthn authenticate finish failed: %s", exc)
|
|
return _error(str(exc), 401)
|
|
|
|
|
|
@bp.route("/webauthn/credentials", methods=["GET"])
|
|
def webauthn_credentials_list():
|
|
"""List registered WebAuthn credentials.
|
|
|
|
Endpoint:
|
|
GET /api/auth/webauthn/credentials
|
|
Returns:
|
|
{ "credentials": [...] }
|
|
"""
|
|
try:
|
|
return _ok(get(GET_AUTH_WEBAUTHN_CREDENTIALS, {**(request._user_ctx or {})}))
|
|
except Exception as exc:
|
|
logger.error("List WebAuthn credentials failed: %s", exc)
|
|
return _error(str(exc), 500)
|
|
|
|
|
|
@bp.route("/webauthn/credential-counts", methods=["GET"])
|
|
def webauthn_credential_counts():
|
|
"""Return credential counts for all users.
|
|
|
|
Admin endpoint — returns a dict mapping usernames to credential counts.
|
|
|
|
Endpoint:
|
|
GET /api/auth/webauthn/credential-counts
|
|
Returns:
|
|
{ "counts": { "username": 2, ... } }
|
|
"""
|
|
try:
|
|
return _ok(get(GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS))
|
|
except RuntimeError as exc:
|
|
logger.error("List credential counts failed: %s", exc)
|
|
return _error(str(exc), 500)
|
|
|
|
|
|
@bp.route("/webauthn/creds/<credential_id>", methods=["DELETE"])
|
|
def webauthn_remove_credential(credential_id: str):
|
|
"""Remove a WebAuthn credential.
|
|
|
|
Endpoint:
|
|
DELETE /api/auth/webauthn/creds/<credential_id>
|
|
Returns:
|
|
{ "ok": true }
|
|
"""
|
|
try:
|
|
return _ok(
|
|
delete(
|
|
DELETE_AUTH_WEBAUTHN_CREDENTIAL,
|
|
{**(request._user_ctx or {}), "credential_id": credential_id},
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
err = str(exc)
|
|
status = 404 if "not found" in err.lower() else 400
|
|
logger.error("Remove WebAuthn credential failed: %s", exc)
|
|
return _error(err, status)
|