T13.1: idle timeout + absolute ceiling (D18)

Sessions now slide on activity (AUTH_IDLE_MINUTES, default 30) capped by
a hard ceiling from original sign-in (AUTH_SESSION_HOURS, meaning changed,
default 12 -> proposed 8). login_at carried across reissues so the ceiling
survives refreshes; pre-D18 tokens with no login_at fall back to iat.
Refresh is throttled (~IDLE_MINUTES/3) so the cookie isn't rewritten on
every request. Wired into auth_gate (server/app.py) - no DB hit, reads
only the already-validated claims.

Verified: 7 unit-level checks (fresh-token expiry, past-ceiling refusal,
throttling, mid-session extension, legacy-token fallback both live and
expired, idle cutoff itself) all pass, plus the full 27-check smoke
suite still passes end to end through the new middleware path.
This commit is contained in:
2026-09-23 11:40:59 -07:00
parent 358469531c
commit 6cde6e3f60
4 changed files with 120 additions and 7 deletions

View File

@@ -58,6 +58,15 @@ POSTGRES_PASSWORD=<strong-random-password>
# openssl rand -base64 48 # openssl rand -base64 48
AUTH_SECRET_KEY=<strong-random-secret> AUTH_SECRET_KEY=<strong-random-secret>
# OPTIONAL — D18 (2026-09-23): a session slides on activity (AUTH_IDLE_MINUTES,
# default 30) capped by a hard ceiling from original sign-in regardless of
# activity (AUTH_SESSION_HOURS, default 8). Both are proposed defaults, not
# confirmed against this tenant's Okta SSO session policy — if Okta's own
# session outlives either one, re-auth here is likely a fast silent redirect
# rather than a real login screen. Full explanation in server/.env.example.
# AUTH_IDLE_MINUTES=30
# AUTH_SESSION_HOURS=8
# REQUIRED (in spirit — see the note below) — Okta OIDC is the only sign-in # REQUIRED (in spirit — see the note below) — Okta OIDC is the only sign-in
# path (D15/D16). There is no local password anywhere in this app to fall back # path (D15/D16). There is no local password anywhere in this app to fall back
# to, so without these nobody can sign in at all. Get them from the Okta app # to, so without these nobody can sign in at all. Get them from the Okta app

View File

@@ -27,8 +27,18 @@ DATABASE_URL=postgresql+psycopg://wpsuite:CHANGE_ME@localhost:5432/wpsuite
# python -c "import secrets; print(secrets.token_urlsafe(48))" # python -c "import secrets; print(secrets.token_urlsafe(48))"
AUTH_SECRET_KEY=CHANGE_ME_run_the_command_above AUTH_SECRET_KEY=CHANGE_ME_run_the_command_above
# How long a login lasts before re-authentication (hours). Default 12. # D18 (2026-09-23): a session slides on activity, capped by a hard ceiling
# AUTH_SESSION_HOURS=12 # underneath - not one flat lifetime. Both defaults are proposed, not
# confirmed against this tenant's actual Okta SSO session policy - if Okta's
# own session outlives either number, re-auth here is likely a fast redirect
# rather than a real login screen, so these cost less than they look like.
#
# No request for this many minutes ends the session outright.
# AUTH_IDLE_MINUTES=30
#
# The absolute ceiling from original sign-in, regardless of activity - no
# session outlives this no matter how continuously active it is. Default 8.
# AUTH_SESSION_HOURS=8
# ── Okta OIDC (required — this is the only sign-in path) ─────────────────────── # ── Okta OIDC (required — this is the only sign-in path) ───────────────────────
# The Okta *authorization server* issuer, e.g. https://yourorg.okta.com/oauth2/default # The Okta *authorization server* issuer, e.g. https://yourorg.okta.com/oauth2/default

View File

@@ -103,12 +103,23 @@ def _csrf_ok(request: Request) -> bool:
async def auth_gate(request: Request, call_next): async def auth_gate(request: Request, call_next):
path = request.url.path path = request.url.path
method = request.method method = request.method
claims = None
if method != "OPTIONS" and auth._needs_auth(path): if method != "OPTIONS" and auth._needs_auth(path):
if not auth.is_request_authenticated(request): claims = auth.is_request_authenticated(request)
if not claims:
return JSONResponse(status_code=401, content={"detail": "Not authenticated"}) return JSONResponse(status_code=401, content={"detail": "Not authenticated"})
if method in _UNSAFE_METHODS and path.startswith("/api/") and not _csrf_ok(request): if method in _UNSAFE_METHODS and path.startswith("/api/") and not _csrf_ok(request):
return JSONResponse(status_code=403, content={"detail": "Cross-origin request rejected"}) return JSONResponse(status_code=403, content={"detail": "Cross-origin request rejected"})
return await call_next(request) response = await call_next(request)
# D18: slide the session forward on activity, capped by its absolute
# ceiling. Reads only the already-validated claims - no DB hit here, and
# the separate is_active/token_version check in get_current_user still
# runs on its own for every request regardless of whether this refreshes.
if claims is not None:
refreshed = auth.maybe_refresh_token(claims)
if refreshed:
auth.set_session_cookie(response, request, refreshed)
return response
def gen_id(prefix: str) -> str: def gen_id(prefix: str) -> str:

View File

@@ -59,8 +59,26 @@ log = logging.getLogger("wpsuite.auth")
COOKIE_NAME = "wp_session" COOKIE_NAME = "wp_session"
JWT_ALG = "HS256" JWT_ALG = "HS256"
# How long a session lasts before the person must sign in again. # D18 (2026-09-23): a session now slides on activity, capped by a hard ceiling
SESSION_HOURS = int(os.getenv("AUTH_SESSION_HOURS", "12")) # underneath - not a single flat lifetime. A pure idle timer with no ceiling
# would let a continuously-active session never force a fresh Okta recheck,
# which is a worse fit for this item's own purpose (catching someone still
# active after being deprovisioned) than a flat expiry would have been. Both
# numbers are proposed defaults, not confirmed against the tenant's actual
# Okta SSO session policy - see docs/waves/decisions-2026-09-17.md.
#
# No request for this long invalidates the session outright.
IDLE_MINUTES = int(os.getenv("AUTH_IDLE_MINUTES", "30"))
# The absolute ceiling from the ORIGINAL sign-in, regardless of activity. Same
# env var name as the old flat-lifetime design; the meaning changed, the name
# didn't, because it still answers "how long can this session possibly live."
SESSION_HOURS = int(os.getenv("AUTH_SESSION_HOURS", "8"))
# How much later a refreshed exp must be before it's worth rewriting the
# cookie. Without this, an active session gets a new Set-Cookie on literally
# every request - correct, but wasteful, and it makes the cookie a busier
# target than it needs to be. A third of the idle window is a reasonable
# balance: refreshed a few times within any idle window, never every request.
_REFRESH_SLACK = timedelta(minutes=max(1, IDLE_MINUTES // 3))
# ── permissions roles ───────────────────────────────────────────────────────── # ── permissions roles ─────────────────────────────────────────────────────────
ROLE_ADMIN = "admin" ROLE_ADMIN = "admin"
@@ -160,6 +178,30 @@ SECRET_KEY = _load_secret()
# ── tokens ────────────────────────────────────────────────────────────────── # ── tokens ──────────────────────────────────────────────────────────────────
def _parse_claim_dt(v) -> Optional[datetime]:
"""`login_at` is a custom claim, so unlike `exp`/`iat` (which PyJWT
special-cases for a datetime -> POSIX-timestamp conversion on encode) it
is stored and read back as a plain numeric timestamp. Returns None for
anything unparseable rather than raising - a malformed or legacy token
should fail closed into "no refresh", not 500."""
if v is None:
return None
try:
return datetime.fromtimestamp(float(v), tz=timezone.utc)
except (TypeError, ValueError, OSError):
return None
def _next_exp(login_at: datetime, now: datetime) -> datetime:
"""Whichever comes first: another IDLE_MINUTES of quiet from now, or the
absolute SESSION_HOURS ceiling measured from the session's original
sign-in. Shared by create_token and maybe_refresh_token so the two can't
drift apart."""
ceiling = login_at + timedelta(hours=SESSION_HOURS)
idle_edge = now + timedelta(minutes=IDLE_MINUTES)
return min(ceiling, idle_edge)
def create_token(user: "models.User") -> str: def create_token(user: "models.User") -> str:
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
payload = { payload = {
@@ -168,7 +210,48 @@ def create_token(user: "models.User") -> str:
"role": user.role, "role": user.role,
"ver": user.token_version or 0, "ver": user.token_version or 0,
"iat": now, "iat": now,
"exp": now + timedelta(hours=SESSION_HOURS), "login_at": now.timestamp(),
"exp": _next_exp(now, now),
}
return jwt.encode(payload, SECRET_KEY, algorithm=JWT_ALG)
def maybe_refresh_token(claims: dict) -> Optional[str]:
"""Given a validated token's claims, return a reissued token if the
session is worth extending, or None if nothing should change. Called from
`auth_gate` on every authenticated request (D18) - deliberately reads only
the already-validated claims, never the database, so it costs nothing
beyond the JWT encode itself. The separate is_active/token_version check
in get_current_user is unaffected either way.
Three ways this returns None: past the absolute ceiling (session is done,
full re-login required - never extended, not even by a second); the
claims can't be parsed (fails closed into no-refresh rather than guessing);
or a refresh happened recently enough that a new cookie isn't worth
writing yet (_REFRESH_SLACK)."""
now = datetime.now(timezone.utc)
login_at = _parse_claim_dt(claims.get("login_at"))
if login_at is None:
# Pre-D18 token (no login_at claim) - fall back to iat so it still
# gets a real ceiling instead of riding on the old flat exp forever.
login_at = _parse_claim_dt(claims.get("iat"))
if login_at is None:
return None
ceiling = login_at + timedelta(hours=SESSION_HOURS)
if now >= ceiling:
return None
new_exp = _next_exp(login_at, now)
current_exp = _parse_claim_dt(claims.get("exp"))
if current_exp is not None and (new_exp - current_exp) < _REFRESH_SLACK:
return None
payload = {
"sub": claims.get("sub"),
"username": claims.get("username"),
"role": claims.get("role"),
"ver": claims.get("ver", 0),
"iat": now,
"login_at": login_at.timestamp(),
"exp": new_exp,
} }
return jwt.encode(payload, SECRET_KEY, algorithm=JWT_ALG) return jwt.encode(payload, SECRET_KEY, algorithm=JWT_ALG)