Calculate session expiry times in operations, using a common now timestamp for everything.

paskia/db/operations.py

@@ -9,7 +9,7 @@ Write operations: Functions that validate and commit, or raise ValueError.
 import hashlib
 import logging
 import os
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from uuid import UUID
 
 import uuid7
@@ -456,7 +456,7 @@ def create_session(
     host: str,
     ip: str,
     user_agent: str,
-    expiry: datetime,
+    duration: timedelta = SESSION_LIFETIME,
     *,
     ctx: SessionContext | None = None,
 ) -> str:
@@ -465,18 +465,19 @@ def create_session(
         raise ValueError(f"User {user_uuid} not found")
     if credential_uuid not in _db.credentials:
         raise ValueError(f"Credential {credential_uuid} not found")
+    now = datetime.now(UTC)
     session = Session.create(
         user=user_uuid,
         credential=credential_uuid,
         host=host,
         ip=ip,
         user_agent=user_agent,
-        expiry=expiry,
+        expiry=now + duration,
     )
     if session.key in _db.sessions:
         raise ValueError("Session already exists")
     with _db.transaction("create_session", ctx):
-        session.store()
+        session.store(now)
     return session.key
 
 
@@ -613,7 +614,7 @@ def login(
     host: str,
     ip: str,
     user_agent: str,
-    expiry: datetime,
+    duration: timedelta = SESSION_LIFETIME,
 ) -> str:
     """Update user/credential on login and create session in a single transaction.
 
@@ -639,11 +640,11 @@ def login(
         host=host,
         ip=ip,
         user_agent=user_agent,
-        expiry=expiry,
+        expiry=now + duration,
     )
     user_str = str(user_uuid)
     with _db.transaction("login", user=user_str):
-        session.store()
+        session.store(now)
         # Update credential
         _db.credentials[credential_uuid].sign_count = sign_count
         _db.credentials[credential_uuid].last_used = now
@@ -671,7 +672,6 @@ def create_credential_session(
     """
 
     now = datetime.now(UTC)
-    expiry = now + SESSION_LIFETIME
 
     if user_uuid not in _db.users:
         raise ValueError(f"User {user_uuid} not found")
@@ -682,7 +682,7 @@ def create_credential_session(
         host=host,
         ip=ip,
         user_agent=user_agent,
-        expiry=expiry,
+        expiry=now + SESSION_LIFETIME,
     )
     user_str = str(user_uuid)
     with _db.transaction("create_credential_session", user=user_str):
@@ -694,7 +694,7 @@ def create_credential_session(
         _db.credentials[credential.uuid] = credential
 
         # Store session and record visit
-        session.store()
+        session.store(now)
 
         # Delete reset token if provided
         if reset_key:

paskia/db/structs.py

@@ -309,7 +309,7 @@ class Session(msgspec.Struct, dict=True):
             "expiry": self.expiry.isoformat(),
         }
 
-    def store(self) -> None:
+    def store(self, now: datetime) -> None:
         """Store this session in the database and record a visit.
 
         Updates user.last_seen and user.visits. Must be called inside
@@ -317,7 +317,6 @@ class Session(msgspec.Struct, dict=True):
         """
         _data = db.data()
         _data.sessions[self.key] = self
-        now = datetime.now(UTC)
         _data.users[self.user_uuid].last_seen = now
         _data.users[self.user_uuid].visits += 1
 

paskia/fastapi/wschat.py

@@ -7,7 +7,6 @@ from uuid import UUID
 from fastapi import WebSocket
 
 from paskia import db
-from paskia.authsession import expires
 from paskia.db import Credential, SessionContext
 from paskia.fastapi.session import infodict
 from paskia.fastapi.wsutil import validate_origin
@@ -105,7 +104,6 @@ async def authenticate_and_login(
         host=normalized_host,
         ip=metadata["ip"],
         user_agent=metadata["user_agent"],
-        expiry=expires(),
     )
 
     # Fetch and return the full session context

tests/conftest.py

@@ -21,7 +21,7 @@ import pytest_asyncio
 
 import paskia.db.operations as ops_db
 from paskia import globals as paskia_globals
-from paskia.authsession import expires, reset_expires
+from paskia.authsession import reset_expires
 from paskia.db import (
     Credential,
     Org,
@@ -183,7 +183,6 @@ async def session_token(
         host="localhost",
         ip="127.0.0.1",
         user_agent="pytest",
-        expiry=expires(),
     )
 
 
@@ -198,7 +197,6 @@ async def regular_session_token(
         host="localhost",
         ip="127.0.0.1",
         user_agent="pytest",
-        expiry=expires(),
     )
 
 

tests/test_admin.py

@@ -22,7 +22,6 @@ import pytest_asyncio
 import uuid7
 
 from paskia import db
-from paskia.authsession import expires
 from paskia.db import (
     Credential,
     Org,
@@ -104,7 +103,6 @@ async def second_org_session_token(
         host="localhost",
         ip="127.0.0.1",
         user_agent="pytest",
-        expiry=expires(),
     )
 
 
@@ -161,7 +159,6 @@ async def org_admin_session_token(
         host="localhost",
         ip="127.0.0.1",
         user_agent="pytest",
-        expiry=expires(),
     )
 
 
@@ -1299,7 +1296,6 @@ class TestAdminSessions:
             host="other.host:4401",
             ip="192.168.1.1",
             user_agent="other-agent",
-            expiry=expires(),
         )
 
         response = await client.delete(

tests/test_api.py

@@ -11,7 +11,7 @@ These tests cover:
 """
 
 import secrets
-from datetime import UTC, datetime, timedelta
+from datetime import timedelta
 
 import httpx
 import pytest
@@ -521,15 +521,14 @@ class TestValidateSessionRefresh:
     ):
         """Validate should return 401 if session disappears during refresh."""
 
-        # Create a session with an old expiry time to trigger refresh
-        old_expiry = datetime.now(UTC) + EXPIRES - timedelta(minutes=10)
+        # Create a session with a short remaining duration to trigger refresh
         token = create_session(
             user_uuid=test_user.uuid,
             credential_uuid=test_credential.uuid,
             host="localhost",
             ip="127.0.0.1",
             user_agent="pytest",
-            expiry=old_expiry,
+            duration=EXPIRES - timedelta(minutes=10),
         )
 
         # Delete the session right before validate tries to refresh