enforce mandatory X-Session-Id header for access token validation

Session binding was bypassable: if the X-Session-Id header was absent,
validate_token skipped the check entirely, allowing a stolen JWT to be
used without the originating session.

Server-side: reject 401 early in Flask middleware and daemon WebSocket
handler when X-Session-Id is missing, before calling validate_token.
Updated validate_token to always enforce session_id matching for access
tokens (refresh tokens are unaffected as they carry no session_id claim).

Frontend: removed dead if (stored.session_id) guards in api.js since
the header is now always required. Added X-Session-Id to logout request
headers and always store session_id on login/refresh.
This commit is contained in:
2026-07-30 22:55:16 +00:00
parent 43b44ad340
commit b69ca330f4
6 changed files with 39 additions and 28 deletions
+19 -11
View File
@@ -235,13 +235,13 @@ class TestJWT:
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."""
def test_session_id_required(self):
"""Access token is rejected without session_id parameter."""
token = generate_access_token(
"admin", {"firewall": "rw"}, session_id="my-session"
)
payload = validate_token(token, "access")
assert payload is not None
assert payload is None
def test_decode_token(self):
token = generate_access_token("admin", {"firewall": "rw"})
@@ -251,23 +251,27 @@ class TestJWT:
assert payload["type"] == "access"
def test_validate_access_token(self):
token = generate_access_token("admin", {"firewall": "rw"})
payload = validate_token(token, "access")
token = generate_access_token(
"admin", {"firewall": "rw"}, session_id="test-session"
)
payload = validate_token(token, "access", session_id="test-session")
assert payload is not None
assert payload["sub"] == "admin"
assert payload["permissions"]["firewall"] == "rw"
def test_validate_wrong_type(self):
token = generate_refresh_token("admin")
payload = validate_token(token, "access")
payload = validate_token(token, "access", session_id="test-session")
assert payload is None
def test_validate_invalid_token(self):
payload = validate_token("invalid.token.here", "access")
payload = validate_token(
"invalid.token.here", "access", session_id="test-session"
)
assert payload is None
def test_blacklist_token(self):
token = generate_access_token("admin", {})
token = generate_access_token("admin", {}, session_id="test-session")
payload = decode_token(token)
assert payload is not None
jti = payload["jti"]
@@ -275,7 +279,7 @@ class TestJWT:
blacklist_token(jti)
assert is_blacklisted(jti) is True
result = validate_token(token, "access")
result = validate_token(token, "access", session_id="test-session")
assert result is None
def test_blacklist_cleanup(self):
@@ -373,7 +377,9 @@ class TestLoginFlow:
assert verify_user_password("testadmin", "secretpass") is not None
tokens = generate_tokens("testadmin", user["permissions"])
payload = validate_token(tokens["access_token"], "access")
payload = validate_token(
tokens["access_token"], "access", session_id=tokens["session_id"]
)
assert payload is not None
assert payload["sub"] == "testadmin"
assert payload["permissions"]["firewall"] == "rw"
@@ -387,7 +393,9 @@ class TestLoginFlow:
blacklist_token(payload["jti"])
result = validate_token(tokens["access_token"], "access")
result = validate_token(
tokens["access_token"], "access", session_id=tokens["session_id"]
)
assert result is None
def test_token_refresh_flow(self):