diff --git a/.env_sample b/.env_sample index 7eb5794..00ddb1d 100644 --- a/.env_sample +++ b/.env_sample @@ -1,7 +1,7 @@ BAPI_JWT_SECRET=please_please_update_me_please BAPI_JWT_ALGORITHM=HS256 -# expiry time in milliseconds (3600000 = 1 hour) -BAPI_JWT_EXPIRY_TIME=3600000 +# token lifetime in seconds (3600 = 1 hour) +BAPI_JWT_EXPIRY_TIME=3600 # Set the ASGI root_path for applications submounted below a given URL path. BAPI_ROOT_PATH = "/" diff --git a/app/auth/auth_handler.py b/app/auth/auth_handler.py index 0c68fd4..d942a76 100644 --- a/app/auth/auth_handler.py +++ b/app/auth/auth_handler.py @@ -1,7 +1,6 @@ import asyncio import os import time -from typing import Dict import jwt from loguru import logger @@ -10,22 +9,30 @@ from app.api.config import config JWT_SECRET = config("BAPI_JWT_SECRET") JWT_ALGORITHM = config("BAPI_JWT_ALGORITHM") -JWT_EXPIRY_TIME = config("BAPI_JWT_EXPIRY_TIME", default=300, cast=int) +# Token lifetime in seconds. +JWT_EXPIRY_TIME = config("BAPI_JWT_EXPIRY_TIME", default=3600, cast=int) -def sign_jwt() -> Dict[str, str]: +def sign_jwt() -> str: + now = int(time.time()) payload = { "user_id": "admin", - "expires": int(round(time.time() * 1000) + JWT_EXPIRY_TIME), + "iat": now, + # standard 'exp' claim (seconds) so PyJWT validates expiry itself + "exp": now + JWT_EXPIRY_TIME, } - token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) - return token + return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) def decodeJWT(token: str) -> dict: try: - decoded_token = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) - return decoded_token if decoded_token["expires"] >= time.time() * 1000 else None + # PyJWT validates the 'exp' claim and raises on expiry + return jwt.decode( + token, + JWT_SECRET, + algorithms=[JWT_ALGORITHM], + options={"require": ["exp"]}, + ) except Exception as e: logger.warning(f"Unable to decode jwt_token {e}") return {} @@ -67,8 +74,11 @@ def remove_local_cookie(): def register_cookie_updater(): # We need to update the cookie file once the cookie is expired async def _cookie_updater(): + # refresh shortly before expiry; JWT_EXPIRY_TIME is in seconds. + # guard against tiny/negative values that would busy-loop. + refresh_interval = max(JWT_EXPIRY_TIME - 10, 1) while True: - await asyncio.sleep(JWT_EXPIRY_TIME - 10) + await asyncio.sleep(refresh_interval) handle_local_cookie() loop = asyncio.get_event_loop() diff --git a/tests/test_jwt.py b/tests/test_jwt.py new file mode 100644 index 0000000..40c4aa3 --- /dev/null +++ b/tests/test_jwt.py @@ -0,0 +1,58 @@ +""" +Tests for JWT signing/validation. + +Two bugs are covered: +- expiry was computed in seconds but added to a milliseconds epoch, so with + the default (300) tokens effectively never got the intended lifetime, and + the cookie-refresh loop slept for the wrong unit. +- a custom "expires" claim was used instead of the standard "exp", so PyJWT + performed no expiry validation of its own. +""" + +import time + +import jwt + +from app.auth.auth_bearer import JWTBearer +from app.auth.auth_handler import ( + JWT_ALGORITHM, + JWT_EXPIRY_TIME, + JWT_SECRET, + decodeJWT, + sign_jwt, +) + + +def test_signed_token_expiry_is_in_seconds(): + token = sign_jwt() + payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) + + assert "exp" in payload, "token must carry a standard 'exp' claim" + # exp must be roughly now + JWT_EXPIRY_TIME *seconds* (not ms) + expected = time.time() + JWT_EXPIRY_TIME + assert abs(payload["exp"] - expected) < 5, ( + f"exp {payload['exp']} not ~{expected} (unit mismatch?)" + ) + + +def test_expired_standard_claim_is_rejected(): + now = int(time.time()) + # old code would accept this (future ms 'expires'); the fixed code must + # reject it because the standard 'exp' claim is in the past + token = jwt.encode( + { + "user_id": "admin", + "expires": int((now + 9999) * 1000), + "exp": now - 10, + }, + JWT_SECRET, + algorithm=JWT_ALGORITHM, + ) + + assert decodeJWT(token) == {} + assert JWTBearer().verify_jwt(token) is False + + +def test_fresh_token_verifies(): + token = sign_jwt() + assert JWTBearer().verify_jwt(token) is True