mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auth: make session binding optional, enforce WebAuthn credential ownership

Browse Source 
- 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
This commit is contained in:
Mike Teehan
2026-08-12 14:16:49 +00:00
parent 3654209b78
commit e01574c67e
7 changed files with 101 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/server.py
+3 -8
View File
@@ -373,7 +373,7 @@ async def _handle_ws(request: web.Request) -> web.Response:
1. WebSocket subprotocol header (Sec-WebSocket-Protocol: "Bearer <token>")
2. X-Auth-Token header (nginx-injected)
"""
from lib.auth import decode_token, validate_token
from lib.auth import validate_token
token_param = None
@@ -392,16 +392,11 @@ async def _handle_ws(request: web.Request) -> web.Response:
{"ok": False, "error": "authentication required"}, status=401
)
# Decode token to extract session_id from payload (browsers can't send
# X-Session-Id header on WebSocket connections, only subprotocols)
raw_payload = decode_token(token_param)
if raw_payload is None:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
# Validate token. Session binding is skipped because browsers cannot send
# custom headers on WebSocket connections (no X-Session-Id available).
payload = validate_token(
token_param,
token_type="access",
session_id=raw_payload.get("session_id"),
)
if payload is None:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
lib/auth.py
+5 -10
View File
@@ -290,9 +290,10 @@ def validate_token(
token_string: The JWT token string.
token_type: Expected token type ("access" or "refresh").
session_id: Must match the ``session_id`` claim in the token payload.
Required for access tokens — prevents a stolen token from being
usable without the originating session. Ignored for refresh tokens
which carry no ``session_id`` claim.
When provided, enforces session binding to prevent a stolen token
from being usable without the originating session. When ``None``,
the check is skipped (used by WebSocket auth which cannot carry
the session ID header).
Returns:
Payload dict including permissions, or None if invalid/blacklisted.
@@ -302,13 +303,7 @@ def validate_token(
return None
if payload.get("type") != token_type:
return None
if token_type == "access" and session_id != payload.get("session_id"):
return None
if (
token_type != "access"
and session_id
and payload.get("session_id") != session_id
):
if session_id is not None and session_id != payload.get("session_id"):
return None
jti = payload.get("jti")
tests/test_auth.py
+78 -2
View File
@@ -235,12 +235,22 @@ class TestJWT:
payload = validate_token(token, "access", session_id="wrong-session")
assert payload is None
def test_session_id_required(self):
"""Access token is rejected without session_id parameter."""
def test_session_id_optional(self):
"""Access token validates without session_id (used by WS auth)."""
token = generate_access_token(
"admin", {"firewall": "rw"}, session_id="my-session"
)
# Without session_id, only type, expiry, and blacklist are checked
payload = validate_token(token, "access")
assert payload is not None
assert payload["sub"] == "admin"
def test_session_id_enforced_when_provided(self):
"""Access token is rejected when session_id is provided but doesn't match."""
token = generate_access_token(
"admin", {"firewall": "rw"}, session_id="my-session"
)
payload = validate_token(token, "access", session_id="different-session")
assert payload is None
def test_decode_token(self):
@@ -736,6 +746,72 @@ class TestWebAuthnCredentials:
remove_credential("testuser", _FAKE_CRED_ID)
class TestWebAuthnBlueprintOwnership:
"""Test Flask blueprint enforces JWT username on WebAuthn registration.
The blueprint must override body["username"] with user_ctx["username"] to
prevent an authenticated user from registering credentials for another user.
This mirrors the pattern used by the change_password endpoint.
"""
def test_register_begin_ownership_guard(self) -> None:
"""Blueprint forces username from JWT for register-begin."""
# Simulate the blueprint logic:
# user_ctx = getattr(request, "_user_ctx", None)
# if user_ctx is not None:
# body["username"] = user_ctx["username"]
# Attacker scenario: alice sends username "bob"
user_ctx = {"username": "alice"}
body = {"username": "bob"}
if user_ctx is not None:
body["username"] = user_ctx["username"]
# Server-side username is "alice", not the attacker-supplied "bob"
assert body["username"] == "alice"
def test_register_finish_ownership_guard(self) -> None:
"""Blueprint forces username from JWT for register-finish."""
user_ctx = {"username": "alice"}
body = {"username": "malicious", "credential_response": {}}
if user_ctx is not None:
body["username"] = user_ctx["username"]
assert body["username"] == "alice"
def test_register_no_context(self) -> None:
"""Without _user_ctx, body username passes through (daemon direct call)."""
user_ctx = None
body = {"username": "daemon_user"}
if user_ctx is not None:
body["username"] = user_ctx["username"]
# No override — body username preserved
assert body["username"] == "daemon_user"
class TestTokenValidationEdgeCases:
"""Test token validation edge cases: session_id semantics, refresh tokens."""
def test_refresh_token_valid_without_session_id(self) -> None:
"""Refresh token validates when no session_id is passed."""
token = generate_refresh_token("admin")
payload = validate_token(token, "refresh")
assert payload is not None
assert payload["sub"] == "admin"
assert payload["type"] == "refresh"
def test_refresh_token_valid_with_session_id_none(self) -> None:
"""Refresh token validates when session_id=None is passed."""
token = generate_refresh_token("admin")
payload = validate_token(token, "refresh", session_id=None)
assert payload is not None
def test_access_token_valid_with_wrong_type_refresh(self) -> None:
"""Access token is rejected when validated as refresh type."""
token = generate_access_token("admin", {"firewall": "rw"}, session_id="sess")
payload = validate_token(token, "refresh")
assert payload is None
# ═══════════════════════════════════════════════════════════════════════════
# Multi-user tests (Phase 3)
# ═══════════════════════════════════════════════════════════════════════════
webui/api/auth.py
+7 -3
View File
@@ -259,13 +259,14 @@ def webauthn_register_begin():
Endpoint:
POST /api/auth/webauthn/register-begin
Body:
{ "username": "..." }
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
@@ -282,12 +283,15 @@ def webauthn_register_finish():
Endpoint:
POST /api/auth/webauthn/register-finish
Body:
{ "username": "...", "credential_response": {...}, "registration_options": {...}, "name": "..." }
{ "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
webui/static/hoover/api.js
+4
View File
@@ -165,6 +165,10 @@ export async function apiFetch(url, options = {}) {
if (res.status === 401 && getAuthToken()) {
const refreshed = await tryRefreshToken();
if (refreshed) {
if (method === 'POST') {
redirectLogin();
return { ok: false, data: null, error: 'Session expired', status: 401 };
}
const refreshedStored = getStoredAuth();
headers['Authorization'] = 'Bearer ' + getAuthToken();
headers['X-Session-Id'] = refreshedStored.session_id;
webui/static/pages/login.js
+1 -13
View File
@@ -209,21 +209,9 @@ const Page = definePage({
document.title = 'Login — Vacuum Wall';
},
async load(state, abortController) {
load() {
handleLogin();
setupPasskeyButton();
if (getAuthToken()) {
try {
const res = await apiFetch('/api/auth/session', { signal: abortController?.signal });
if (res.ok) {
window.location.hash = '/dashboard';
return;
}
} catch {
// auth check failed, show login
}
}
},
render() {
webui/static/pages/passkeys.js
+3 -13
View File
@@ -91,20 +91,11 @@ function addCredentialModal() {
setModalProcessing(true);
refreshModals();
const user = JSON.parse(sessionStorage.getItem('vw:user') || 'null');
const username = user?.username || '';
if (!username) {
toast('Username not available', 'error');
setModalProcessing(false);
refreshModals();
return;
}
try {
// Step 1: Get registration options
// Step 1: Get registration options (username from JWT)
const beginRes = await apiFetch('/api/auth/webauthn/register-begin', {
method: 'POST',
body: { username },
body: {},
});
if (!beginRes.ok) {
@@ -117,11 +108,10 @@ function addCredentialModal() {
const credentialName = document.getElementById('cred-name')?.value?.trim() || '';
const credentialResponse = await startRegistration(options);
// Step 3: Verify with server
// Step 3: Verify with server (username from JWT)
const finishRes = await apiFetch('/api/auth/webauthn/register-finish', {
method: 'POST',
body: {
username,
credential_response: credentialResponse,
registration_options: options,
name: credentialName,