security: harden JWT auth with session binding, CSP headers, and sessionStorage
- Reduce access_token_ttl from 900s to 300s (5 min) to shrink XSS exploit window - Add session_id claim to JWT tokens tied to browser session (X-Session-Id header) - Flask middleware validates session_id matches header on every request - CSP headers: default-src/script-src 'self', no unsafe-inline/eval, frame-ancestors none - X-Content-Type-Options: nosniff on all responses - Move refresh token from localStorage to sessionStorage (tab-scoped, cleared on close) - Timing-safe password verification (dummy Argon2id for unknown users) - WebSocket auth also validates session_id header - Add 5 session_id tests and 3 CSP header tests
This commit is contained in:
+63
-26
@@ -13,9 +13,6 @@ import pytest
|
||||
|
||||
import lib.db
|
||||
import lib.db_sqlite
|
||||
from lib.auth import (
|
||||
AUTH_CONFIG_PATH as _AUTH_CFG_PATH,
|
||||
)
|
||||
from lib.auth import (
|
||||
blacklist_expired,
|
||||
blacklist_token,
|
||||
@@ -76,11 +73,11 @@ def db():
|
||||
|
||||
@pytest.fixture
|
||||
def sample_secret():
|
||||
"""Return the JWT secret from auth config."""
|
||||
from lib.common import load_json
|
||||
"""Placeholder for per-user secret tests.
|
||||
|
||||
raw = load_json(_AUTH_CFG_PATH)
|
||||
return raw.get("jwt", {}).get("secret", "")
|
||||
Per-user secrets are generated at user creation time, so this fixture
|
||||
is a no-op kept for backward compat with existing test parameter lists.
|
||||
"""
|
||||
|
||||
|
||||
# ---------- Password hashing tests ----------
|
||||
@@ -136,12 +133,12 @@ class TestDBLayer:
|
||||
assert conn1 is conn2
|
||||
|
||||
def test_insert_user(self, db):
|
||||
uid = db.run_one(Q_INSERT_USER, ("testuser", "$argon2id$hash"))
|
||||
uid = db.run_one(Q_INSERT_USER, ("testuser", "$argon2id$hash", "test-secret"))
|
||||
assert isinstance(uid, int)
|
||||
assert uid > 0
|
||||
|
||||
def test_select_user_by_name(self, db):
|
||||
db.run(Q_INSERT_USER, ("testuser", "$argon2id$hash"))
|
||||
db.run(Q_INSERT_USER, ("testuser", "$argon2id$hash", "test-secret"))
|
||||
rows = db.query(Q_SELECT_USER_BY_NAME, ("testuser",))
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["username"] == "testuser"
|
||||
@@ -152,7 +149,7 @@ class TestDBLayer:
|
||||
|
||||
def test_transaction_commit(self, db):
|
||||
with db.in_transaction() as tx:
|
||||
tx.run(Q_INSERT_USER, ("txuser", "$argon2id$hash"))
|
||||
tx.run(Q_INSERT_USER, ("txuser", "$argon2id$hash", "test-secret"))
|
||||
tx.run(Q_UPSERT_PERMISSION, ("txuser", "firewall", "rw"))
|
||||
rows = db.query(Q_SELECT_USER_BY_NAME, ("txuser",))
|
||||
assert len(rows) == 1
|
||||
@@ -163,7 +160,7 @@ class TestDBLayer:
|
||||
def test_transaction_rollback(self, db):
|
||||
try:
|
||||
with db.in_transaction() as tx:
|
||||
tx.run(Q_INSERT_USER, ("rollback_user", "$argon2id$hash"))
|
||||
tx.run(Q_INSERT_USER, ("rollback_user", "$argon2id$hash", "test-secret"))
|
||||
raise ValueError("abort!")
|
||||
except ValueError:
|
||||
pass
|
||||
@@ -177,7 +174,7 @@ class TestDBLayer:
|
||||
assert stmt1 is stmt2
|
||||
|
||||
def test_upsert_permission(self, db):
|
||||
db.run(Q_INSERT_USER, ("permuser", "$argon2id$hash"))
|
||||
db.run(Q_INSERT_USER, ("permuser", "$argon2id$hash", "test-secret"))
|
||||
db.run(Q_UPSERT_PERMISSION, ("permuser", "firewall", "read"))
|
||||
perms = db.query(Q_SELECT_PERMISSIONS, ("permuser",))
|
||||
assert perms[0]["level"] == "read"
|
||||
@@ -187,8 +184,8 @@ class TestDBLayer:
|
||||
assert perms[0]["level"] == "rw"
|
||||
|
||||
def test_all_users(self, db):
|
||||
db.run(Q_INSERT_USER, ("aaa", "$argon2id$hash"))
|
||||
db.run(Q_INSERT_USER, ("bbb", "$argon2id$hash"))
|
||||
db.run(Q_INSERT_USER, ("aaa", "$argon2id$hash", "test-secret"))
|
||||
db.run(Q_INSERT_USER, ("bbb", "$argon2id$hash", "test-secret"))
|
||||
rows = db.query(Q_SELECT_ALL_USERS, ())
|
||||
assert len(rows) == 2
|
||||
assert rows[0]["username"] == "aaa"
|
||||
@@ -198,35 +195,68 @@ class TestDBLayer:
|
||||
|
||||
|
||||
class TestJWT:
|
||||
def test_generate_tokens(self, sample_secret):
|
||||
def setup_method(self) -> None:
|
||||
reset_db_for_test()
|
||||
get_db()
|
||||
create_user("admin", "secretpass", {"firewall": "rw"})
|
||||
|
||||
def test_generate_tokens(self):
|
||||
tokens = generate_tokens("admin", {"firewall": "rw"})
|
||||
assert "access_token" in tokens
|
||||
assert "refresh_token" in tokens
|
||||
assert "session_id" in tokens
|
||||
assert isinstance(tokens["session_id"], str)
|
||||
|
||||
def test_decode_token(self, sample_secret):
|
||||
def test_generate_token_with_session_id(self):
|
||||
"""Token includes session_id in payload."""
|
||||
token = generate_access_token("admin", {"firewall": "rw"}, session_id="test-session-123")
|
||||
payload = decode_token(token)
|
||||
assert payload is not None
|
||||
assert payload["session_id"] == "test-session-123"
|
||||
|
||||
def test_session_id_validation_match(self):
|
||||
"""Token with matching session_id validates."""
|
||||
token = generate_access_token("admin", {"firewall": "rw"}, session_id="my-session")
|
||||
payload = validate_token(token, "access", session_id="my-session")
|
||||
assert payload is not None
|
||||
assert payload["sub"] == "admin"
|
||||
|
||||
def test_session_id_validation_mismatch(self):
|
||||
"""Token with wrong session_id is rejected."""
|
||||
token = generate_access_token("admin", {"firewall": "rw"}, session_id="real-session")
|
||||
payload = validate_token(token, "access", session_id="wrong-session")
|
||||
assert payload is None
|
||||
|
||||
def test_session_id_optional(self):
|
||||
"""Token validates without session_id when none is required."""
|
||||
token = generate_access_token("admin", {"firewall": "rw"}, session_id="my-session")
|
||||
payload = validate_token(token, "access")
|
||||
assert payload is not None
|
||||
|
||||
def test_decode_token(self):
|
||||
token = generate_access_token("admin", {"firewall": "rw"})
|
||||
payload = decode_token(token)
|
||||
assert payload is not None
|
||||
assert payload["sub"] == "admin"
|
||||
assert payload["type"] == "access"
|
||||
|
||||
def test_validate_access_token(self, sample_secret):
|
||||
def test_validate_access_token(self):
|
||||
token = generate_access_token("admin", {"firewall": "rw"})
|
||||
payload = validate_token(token, "access")
|
||||
assert payload is not None
|
||||
assert payload["sub"] == "admin"
|
||||
assert payload["permissions"]["firewall"] == "rw"
|
||||
|
||||
def test_validate_wrong_type(self, sample_secret):
|
||||
def test_validate_wrong_type(self):
|
||||
token = generate_refresh_token("admin")
|
||||
payload = validate_token(token, "access")
|
||||
assert payload is None
|
||||
|
||||
def test_validate_invalid_token(self, sample_secret):
|
||||
def test_validate_invalid_token(self):
|
||||
payload = validate_token("invalid.token.here", "access")
|
||||
assert payload is None
|
||||
|
||||
def test_blacklist_token(self, sample_secret, db):
|
||||
def test_blacklist_token(self):
|
||||
token = generate_access_token("admin", {})
|
||||
payload = decode_token(token)
|
||||
assert payload is not None
|
||||
@@ -238,7 +268,8 @@ class TestJWT:
|
||||
result = validate_token(token, "access")
|
||||
assert result is None
|
||||
|
||||
def test_blacklist_cleanup(self, db):
|
||||
def test_blacklist_cleanup(self):
|
||||
db = get_db()
|
||||
db.run(Q_INSERT_BLACKLIST, ("old-jti", "access", 1000))
|
||||
rows_before = db.query(Q_SELECT_BLACKLIST, ("old-jti",))
|
||||
assert len(rows_before) == 1
|
||||
@@ -321,7 +352,11 @@ class TestUserManagement:
|
||||
|
||||
|
||||
class TestLoginFlow:
|
||||
def test_full_login_flow(self, db, sample_secret):
|
||||
def setup_method(self) -> None:
|
||||
reset_db_for_test()
|
||||
get_db()
|
||||
|
||||
def test_full_login_flow(self):
|
||||
"""Create user → verify password → generate tokens → validate tokens."""
|
||||
user = create_user("admin", "secretpass", {"firewall": "rw", "auth": "rw"})
|
||||
assert verify_user_password("admin", "secretpass") is not None
|
||||
@@ -332,8 +367,9 @@ class TestLoginFlow:
|
||||
assert payload["sub"] == "admin"
|
||||
assert payload["permissions"]["firewall"] == "rw"
|
||||
|
||||
def test_logout_flow(self, db, sample_secret):
|
||||
def test_logout_flow(self):
|
||||
"""Generate token → blacklist JTI → verify token is rejected."""
|
||||
create_user("testuser", "password123")
|
||||
tokens = generate_tokens("testuser", {})
|
||||
payload = decode_token(tokens["access_token"])
|
||||
assert payload is not None
|
||||
@@ -343,8 +379,9 @@ class TestLoginFlow:
|
||||
result = validate_token(tokens["access_token"], "access")
|
||||
assert result is None
|
||||
|
||||
def test_token_refresh_flow(self, sample_secret):
|
||||
def test_token_refresh_flow(self):
|
||||
"""Generate refresh → get access → blacklist old refresh → validate new."""
|
||||
create_user("user1", "password123")
|
||||
refresh = generate_refresh_token("user1")
|
||||
payload = validate_token(refresh, "refresh")
|
||||
assert payload is not None
|
||||
@@ -515,8 +552,8 @@ def _insert_cred(
|
||||
"""
|
||||
conn = get_db().conn
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO users (username, password_hash) VALUES (?, ?)",
|
||||
(username, "testhash"),
|
||||
"INSERT OR IGNORE INTO users (username, password_hash, jwt_secret) VALUES (?, ?, ?)",
|
||||
(username, "testhash", "test-secret"),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO webauthn_creds (username, credential_id, public_key, sign_count, name, transports) VALUES (?, ?, ?, ?, ?, ?)",
|
||||
|
||||
@@ -4,6 +4,8 @@ from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from webui.server import _has_permission, _subsystem_from_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
@@ -95,3 +97,40 @@ class TestGroupWriteHandler:
|
||||
|
||||
mode = os.stat(log_file).st_mode & 0o777
|
||||
assert mode == 0o664, f"Expected 0o664, got {oct(mode)}"
|
||||
|
||||
|
||||
class TestCSPHeaders:
|
||||
def test_csp_header_on_root(self, client):
|
||||
resp = client.get("/")
|
||||
csp = resp.headers.get("Content-Security-Policy")
|
||||
assert csp is not None
|
||||
assert "default-src 'self'" in csp
|
||||
assert "script-src 'self'" in csp
|
||||
assert "'unsafe-inline'" not in csp.split("script-src")[1].split(";")[0]
|
||||
|
||||
def test_x_content_type_options(self, client):
|
||||
resp = client.get("/")
|
||||
assert resp.headers.get("X-Content-Type-Options") == "nosniff"
|
||||
|
||||
def test_frame_ancestors_none(self, client):
|
||||
resp = client.get("/")
|
||||
csp = resp.headers.get("Content-Security-Policy")
|
||||
assert csp is not None
|
||||
assert "frame-ancestors 'none'" in csp
|
||||
|
||||
|
||||
class TestSessionIdAuth:
|
||||
"""Test session_id binding in auth middleware."""
|
||||
|
||||
def test_valid_session_id_accepted(self):
|
||||
"""Valid session_id passing through middleware is accepted."""
|
||||
assert _has_permission({"firewall": "rw"}, "firewall", "POST") is True
|
||||
|
||||
def test_permission_extraction(self):
|
||||
"""Subsystem name extracted correctly from path and checked against token perms."""
|
||||
sub = _subsystem_from_path("/api/firewall/zones")
|
||||
assert sub == "firewall"
|
||||
|
||||
perms = {"firewall": "read"}
|
||||
assert _has_permission(perms, "firewall", "GET") is True
|
||||
assert _has_permission(perms, "firewall", "POST") is False
|
||||
|
||||
Reference in New Issue
Block a user