LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Session keys hardened (namespaced hashes of tokens). Various cleanup.

Browse Source
This commit is contained in:
Leo Vasanko
2026-02-15 20:14:48 +00:00
parent 18722f0e01
commit e59852b44c
21 changed files with 495 additions and 565 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+52 -10
View File
@@ -9,10 +9,14 @@ Since we can't emulate WebAuthn passkeys, we create sessions directly
in the database to test authenticated endpoints.
"""
from __future__ import annotations
import asyncio
import os
import secrets
import tempfile
from collections.abc import AsyncGenerator
from datetime import UTC, datetime, timedelta
from uuid import UUID
import httpx
@@ -22,6 +26,7 @@ import pytest_asyncio
import paskia.db.operations as ops_db
from paskia import globals as paskia_globals
from paskia.authsession import reset_expires
from paskia.config import SESSION_LIFETIME
from paskia.db import (
Config,
Credential,
@@ -33,14 +38,15 @@ from paskia.db import (
create_credential,
create_reset_token,
create_role,
create_session,
create_user,
)
from paskia.db.jsonl import JsonlStore
from paskia.db.operations import DB
from paskia.db.structs import Session
from paskia.fastapi.mainapp import app
from paskia.fastapi.session import AUTH_COOKIE_NAME
from paskia.sansio import Passkey
from paskia.util.crypto import hash_secret
@pytest.fixture(scope="session")
@@ -60,7 +66,6 @@ async def test_db() -> AsyncGenerator[DB, None]:
- A default organization with Administration role
- An admin user with the Administration role
"""
with tempfile.NamedTemporaryFile(suffix=".jsonl", delete=True) as f:
db = DB(config=Config(rp_id="test.example.com"))
store = JsonlStore(db, f.name)
@@ -179,13 +184,11 @@ async def session_token(
test_db: DB, test_user: User, test_credential: Credential
) -> str:
"""Create a session for the admin user and return the token."""
return create_session(
_db_key, secret = create_test_session(
user_uuid=test_user.uuid,
credential_uuid=test_credential.uuid,
host="localhost",
ip="127.0.0.1",
user_agent="pytest",
)
return secret
@pytest_asyncio.fixture(scope="function")
@@ -193,13 +196,11 @@ async def regular_session_token(
test_db: DB, regular_user: User, regular_credential: Credential
) -> str:
"""Create a session for a regular user and return the token."""
return create_session(
_db_key, secret = create_test_session(
user_uuid=regular_user.uuid,
credential_uuid=regular_credential.uuid,
host="localhost",
ip="127.0.0.1",
user_agent="pytest",
)
return secret
@pytest_asyncio.fixture(scope="function")
@@ -241,3 +242,44 @@ def auth_cookie(token: str) -> httpx.Cookies:
cookies = httpx.Cookies()
cookies.set(AUTH_COOKIE_NAME, token, domain="localhost")
return cookies
def create_test_session(
user_uuid: UUID,
credential_uuid: UUID,
host: str = "localhost",
ip: str = "127.0.0.1",
user_agent: str = "pytest",
duration: timedelta | None = None,
) -> tuple[bytes, str]:
"""Create a test session. Returns (key, token) tuple.
- key: bytes used for session lookup (base64url encode for URLs)
- token: stored in cookie/sent to client
"""
if duration is None:
duration = SESSION_LIFETIME
if user_uuid not in ops_db._db.users:
raise ValueError(f"User {user_uuid} not found")
if credential_uuid not in ops_db._db.credentials:
raise ValueError(f"Credential {credential_uuid} not found")
now = datetime.now(UTC)
# Generate token and derive key
token = secrets.token_urlsafe(12)
key = hash_secret("cookie", token)
session = Session.create(
user=user_uuid,
credential=credential_uuid,
key=key,
host=host,
ip=ip,
user_agent=user_agent,
expiry=now + duration,
)
if session.key in ops_db._db.sessions:
raise ValueError("Session already exists")
with ops_db._db.transaction("create_test_session"):
session.store(now)
return session.key, token
tests/test_admin.py
+13 -15
View File
@@ -16,6 +16,7 @@ import secrets
from datetime import UTC, datetime
from uuid import UUID
import base64url
import httpx
import pytest
import pytest_asyncio
@@ -33,11 +34,11 @@ from paskia.db import (
create_org,
create_permission,
create_role,
create_session,
create_user,
)
from paskia.db.operations import DB
from tests.conftest import auth_headers
from paskia.util.crypto import hash_secret
from tests.conftest import auth_headers, create_test_session
# -------------------- Additional Fixtures --------------------
@@ -97,13 +98,11 @@ async def second_org_session_token(
test_db: DB, second_org_user: User, second_org_credential: Credential
) -> str:
"""Create a session for the second org admin user."""
return create_session(
_db_key, secret = create_test_session(
user_uuid=second_org_user.uuid,
credential_uuid=second_org_credential.uuid,
host="localhost",
ip="127.0.0.1",
user_agent="pytest",
)
return secret
@pytest_asyncio.fixture(scope="function")
@@ -153,13 +152,11 @@ async def org_admin_session_token(
test_db: DB, org_admin_user: User, org_admin_credential: Credential
) -> str:
"""Create a session for the org admin user."""
return create_session(
_db_key, secret = create_test_session(
user_uuid=org_admin_user.uuid,
credential_uuid=org_admin_credential.uuid,
host="localhost",
ip="127.0.0.1",
user_agent="pytest",
)
return secret
@pytest_asyncio.fixture(scope="function")
@@ -1290,7 +1287,7 @@ class TestAdminSessions:
):
"""Admin should be able to delete a user's session."""
# Create an additional session to delete
extra_token = create_session(
extra_db_key, _extra_secret = create_test_session(
user_uuid=test_user.uuid,
credential_uuid=test_credential.uuid,
host="other.host:4401",
@@ -1299,7 +1296,7 @@ class TestAdminSessions:
)
response = await client.delete(
f"/auth/api/admin/users/{test_user.uuid}/sessions/{extra_token}",
f"/auth/api/admin/users/{test_user.uuid}/sessions/{base64url.enc(extra_db_key)}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1316,8 +1313,9 @@ class TestAdminSessions:
test_user,
):
"""Admin can delete their own current session."""
session_db_key = base64url.enc(hash_secret("cookie", session_token))
response = await client.delete(
f"/auth/api/admin/users/{test_user.uuid}/sessions/{session_token}",
f"/auth/api/admin/users/{test_user.uuid}/sessions/{session_db_key}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1362,9 +1360,9 @@ class TestAdminSessions:
f"/auth/api/admin/users/{test_user.uuid}/sessions/invalid!!id",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
assert response.status_code == 400
data = response.json()
assert "Session not found" in data["detail"]
assert "Invalid session ID format" in data["detail"]
@pytest.mark.asyncio
async def test_delete_session_not_found(
tests/test_api.py
+5 -5
View File
@@ -17,9 +17,9 @@ import httpx
import pytest
from paskia.authsession import EXPIRES
from paskia.db import create_session, delete_session
from paskia.db import delete_session
from paskia.util.passphrase import generate
from tests.conftest import auth_headers
from tests.conftest import auth_headers, create_test_session
class TestSettingsEndpoint:
@@ -522,7 +522,7 @@ class TestValidateSessionRefresh:
"""Validate should return 401 if session disappears during refresh."""
# Create a session with a short remaining duration to trigger refresh
token = create_session(
db_key, secret = create_test_session(
user_uuid=test_user.uuid,
credential_uuid=test_credential.uuid,
host="localhost",
@@ -532,11 +532,11 @@ class TestValidateSessionRefresh:
)
# Delete the session right before validate tries to refresh
delete_session(token)
delete_session(db_key)
response = await client.post(
"/auth/api/validate",
headers={**auth_headers(token), "Host": "localhost:4401"},
headers={**auth_headers(secret), "Host": "localhost:4401"},
)
# Session was found initially but disappeared during refresh
assert response.status_code == 401