mirror of
https://github.com/fusion44/blitz_api.git
synced 2026-08-16 12:20:14 +02:00
dccaa17a8e
sign_jwt added JWT_EXPIRY_TIME (seconds) to a milliseconds epoch and stored it in a custom 'expires' claim, while register_cookie_updater slept JWT_EXPIRY_TIME as seconds. With the code default (300) tokens effectively expired almost immediately; with the sampled 3600000 the cookie-refresh loop slept ~41 days, so the local .cookie held an expired token nearly always. The custom claim also meant PyJWT never validated expiry itself. - issue standard 'iat'/'exp' claims (seconds) and let PyJWT validate, requiring 'exp' on decode - derive the cookie refresh interval from the same unit, guarded against tiny/negative values - default BAPI_JWT_EXPIRY_TIME to 3600s and fix .env_sample (was 3600000 'milliseconds') Existing tokens and the local .cookie are invalidated by this change; clients re-login and the cookie regenerates at startup. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
85 lines
2.3 KiB
Python
85 lines
2.3 KiB
Python
import asyncio
|
|
import os
|
|
import time
|
|
|
|
import jwt
|
|
from loguru import logger
|
|
|
|
from app.api.config import config
|
|
|
|
JWT_SECRET = config("BAPI_JWT_SECRET")
|
|
JWT_ALGORITHM = config("BAPI_JWT_ALGORITHM")
|
|
# Token lifetime in seconds.
|
|
JWT_EXPIRY_TIME = config("BAPI_JWT_EXPIRY_TIME", default=3600, cast=int)
|
|
|
|
|
|
def sign_jwt() -> str:
|
|
now = int(time.time())
|
|
payload = {
|
|
"user_id": "admin",
|
|
"iat": now,
|
|
# standard 'exp' claim (seconds) so PyJWT validates expiry itself
|
|
"exp": now + JWT_EXPIRY_TIME,
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
|
|
def decodeJWT(token: str) -> dict:
|
|
try:
|
|
# 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 {}
|
|
|
|
|
|
def handle_local_cookie():
|
|
remove_local_cookie()
|
|
|
|
blitz_path = os.path.join(os.path.expanduser("~"), ".blitz_api")
|
|
full_cookie_file_path = os.path.join(blitz_path, ".cookie")
|
|
enabled = config("BAPI_ENABLE_LOCAL_COOKIE_AUTH", default=False, cast=bool)
|
|
|
|
if not enabled:
|
|
return
|
|
|
|
if not os.path.exists(blitz_path):
|
|
try:
|
|
os.makedirs(blitz_path)
|
|
except OSError as e:
|
|
logger.error(
|
|
f"""Unable to create the .blit_api folder: {e}
|
|
Please make sure that the target folder is readable.
|
|
"""
|
|
)
|
|
f = open(full_cookie_file_path, "w")
|
|
f.write(sign_jwt())
|
|
f.close()
|
|
|
|
|
|
def remove_local_cookie():
|
|
full_cookie_file_path = os.path.join(
|
|
os.path.expanduser("~"), ".blitz_api", ".cookie"
|
|
)
|
|
|
|
if os.path.exists(path=full_cookie_file_path):
|
|
os.remove(full_cookie_file_path)
|
|
|
|
|
|
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(refresh_interval)
|
|
handle_local_cookie()
|
|
|
|
loop = asyncio.get_event_loop()
|
|
loop.create_task(_cookie_updater())
|