Add sliding session renewal and periodic token rotation
This commit is contained in:
@@ -22,6 +22,7 @@ PUBLIC_DIR = Path("public")
|
||||
SESSION_COOKIE = "jellomator_session"
|
||||
SESSION_TTL_SECONDS = int(os.getenv("SESSION_TTL_SECONDS", "86400"))
|
||||
SESSION_COOKIE_SECURE = os.getenv("SESSION_COOKIE_SECURE", "false").lower() in ("1", "true", "yes", "on")
|
||||
SESSION_ROTATE_SECONDS = int(os.getenv("SESSION_ROTATE_SECONDS", "3600"))
|
||||
LOGIN_MAX_ATTEMPTS = int(os.getenv("LOGIN_MAX_ATTEMPTS", "5"))
|
||||
LOGIN_WINDOW_SECONDS = int(os.getenv("LOGIN_WINDOW_SECONDS", "300"))
|
||||
LOGIN_LOCKOUT_SECONDS = int(os.getenv("LOGIN_LOCKOUT_SECONDS", "900"))
|
||||
@@ -154,14 +155,14 @@ def expires_at_iso() -> str:
|
||||
return datetime.utcfromtimestamp(now + SESSION_TTL_SECONDS).isoformat()
|
||||
|
||||
|
||||
def current_user(request: Request):
|
||||
def current_user(request: Request, response: Response | None = None):
|
||||
token = request.cookies.get(SESSION_COOKIE)
|
||||
if not token:
|
||||
return None
|
||||
with db() as c:
|
||||
with c.cursor() as cur:
|
||||
cur.execute(
|
||||
"select s.expires_at,u.username,u.role from sessions s join users u on u.id=s.user_id where s.token=%s",
|
||||
"select s.created_at,s.last_seen_at,s.expires_at,u.username,u.role from sessions s join users u on u.id=s.user_id where s.token=%s",
|
||||
(token,),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
@@ -177,10 +178,37 @@ def current_user(request: Request):
|
||||
except ValueError:
|
||||
cur.execute("delete from sessions where token=%s", (token,))
|
||||
return None
|
||||
cur.execute(
|
||||
"update sessions set last_seen_at=%s where token=%s",
|
||||
(utc_now_iso(), token),
|
||||
)
|
||||
now = datetime.utcnow()
|
||||
now_iso = now.isoformat()
|
||||
new_expires_at = expires_at_iso()
|
||||
last_seen_at = row.get("last_seen_at") or row.get("created_at")
|
||||
should_rotate = False
|
||||
if last_seen_at:
|
||||
try:
|
||||
should_rotate = (now - datetime.fromisoformat(last_seen_at)).total_seconds() >= SESSION_ROTATE_SECONDS
|
||||
except ValueError:
|
||||
should_rotate = True
|
||||
if should_rotate:
|
||||
new_token = secrets.token_urlsafe(32)
|
||||
cur.execute(
|
||||
"update sessions set token=%s,last_seen_at=%s,expires_at=%s where token=%s",
|
||||
(new_token, now_iso, new_expires_at, token),
|
||||
)
|
||||
if response is not None:
|
||||
response.set_cookie(
|
||||
SESSION_COOKIE,
|
||||
new_token,
|
||||
httponly=True,
|
||||
samesite="lax",
|
||||
secure=SESSION_COOKIE_SECURE,
|
||||
max_age=SESSION_TTL_SECONDS,
|
||||
path="/",
|
||||
)
|
||||
else:
|
||||
cur.execute(
|
||||
"update sessions set last_seen_at=%s,expires_at=%s where token=%s",
|
||||
(now_iso, new_expires_at, token),
|
||||
)
|
||||
return {"username": row["username"], "role": row["role"]}
|
||||
|
||||
|
||||
@@ -271,12 +299,13 @@ def read_icon_blob(icon: UploadFile | None) -> tuple[bytes | None, str | None]:
|
||||
|
||||
|
||||
@app.get("/api/me")
|
||||
def me(request: Request):
|
||||
def me(request: Request, response: Response):
|
||||
current = current_user(request, response)
|
||||
with db() as c:
|
||||
with c.cursor() as cur:
|
||||
cur.execute("select count(*) as count from users")
|
||||
needs_setup = cur.fetchone()["count"] == 0
|
||||
return {"needs_setup": needs_setup, "current_user": current_user(request)}
|
||||
return {"needs_setup": needs_setup, "current_user": current}
|
||||
|
||||
|
||||
@app.post("/api/setup")
|
||||
|
||||
Reference in New Issue
Block a user