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

OAuth2 OpenID Connect provider support, API and DB refactoring (#3)

Browse Source 
Allows Paskia to authenticate the user to a client site.
- User friendly client registration flow on the admin app
- Redirect-based authentication flow (per spec)
- Backchannel logout both ways to keep sessions synchronized
- Groups integrated with Paskia's permission system
- Adds email, preferred username and telephone fields on user profile
- All new user basic info layout to show the new information, better looks
- API and DB structures redesigned
- Various unrelated fixes to theming and layout
This commit was merged in pull request #3.
This commit is contained in:
LeoVasanko
2026-02-18 02:40:27 +00:00
parent ccf2deee95
commit 4604a65646
71 changed files with 3706 additions and 805 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+53 -10
View File
@@ -9,12 +9,17 @@ 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 base64url
import httpx
import pytest
import pytest_asyncio
@@ -22,6 +27,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 +39,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 +67,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 +185,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 +197,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 +243,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[str, str]:
"""Create a test session. Returns (key, token) tuple.
- key: str used for session lookup (base64url encoded)
- 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 = base64url.enc(hash_secret("cookie", token))
session = Session.create(
user=user_uuid,
credential=credential_uuid,
key=key,
host=host,
ip=ip,
user_agent=user_agent,
validated=now,
)
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
+48 -40
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")
@@ -182,7 +179,7 @@ class TestExceptionHandlers:
async def test_auth_exception_handler(self, client: httpx.AsyncClient):
"""AuthException should return proper JSON with auth info."""
# Accessing admin without auth triggers AuthException
response = await client.get("/auth/api/admin/orgs")
response = await client.get("/auth/api/admin/info")
assert response.status_code == 401
data = response.json()
assert "detail" in data
@@ -219,7 +216,7 @@ class TestAdminOrganizations:
@pytest.mark.asyncio
async def test_list_orgs_requires_auth(self, client: httpx.AsyncClient):
"""List orgs without auth should return 401."""
response = await client.get("/auth/api/admin/orgs")
response = await client.get("/auth/api/admin/info")
assert response.status_code == 401
@pytest.mark.asyncio
@@ -228,7 +225,7 @@ class TestAdminOrganizations:
):
"""List orgs without admin permission should return 403."""
response = await client.get(
"/auth/api/admin/orgs",
"/auth/api/admin/info",
headers={
**auth_headers(regular_session_token),
"Host": "localhost:4401",
@@ -242,19 +239,24 @@ class TestAdminOrganizations:
):
"""Admin user should be able to list organizations."""
response = await client.get(
"/auth/api/admin/orgs",
"/auth/api/admin/info",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) >= 1
assert isinstance(data, dict)
assert "orgs" in data
orgs_data = data["orgs"]
assert isinstance(orgs_data, dict)
assert len(orgs_data) >= 1
# Check org structure
org = data[0]
org_data = list(orgs_data.values())[0]
assert "org" in org_data
org = org_data["org"]
assert "uuid" in org
assert "display_name" in org
assert "roles" in org
assert "users" in org
assert "roles" in org_data
assert "users" in org_data
@pytest.mark.asyncio
async def test_list_orgs_with_org_admin(
@@ -266,13 +268,13 @@ class TestAdminOrganizations:
):
"""Org admin should only see their own organization."""
response = await client.get(
"/auth/api/admin/orgs",
"/auth/api/admin/info",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
data = response.json()
# Should only see their own org, not the second org
org_uuids = [o["uuid"] for o in data]
org_uuids = [org_data["org"]["uuid"] for org_data in data["orgs"].values()]
assert str(test_org.uuid) in org_uuids
@pytest.mark.asyncio
@@ -794,7 +796,8 @@ class TestAdminUsersInOrg:
)
assert response.status_code == 200
data = response.json()
assert "display_name" in data
assert "user" in data
assert "display_name" in data["user"]
assert "credentials" in data
assert "sessions" in data
assert "aaguid_info" in data
@@ -845,7 +848,8 @@ class TestAdminUsersInOrg:
)
assert response.status_code == 200
data = response.json()
assert "display_name" in data
assert "user" in data
assert "display_name" in data["user"]
@pytest.mark.asyncio
async def test_update_user_display_name_in_org(
@@ -853,7 +857,7 @@ class TestAdminUsersInOrg:
):
"""Admin should be able to update user display name."""
response = await client.patch(
f"/auth/api/admin/users/{test_user.uuid}/display-name",
f"/auth/api/admin/users/{test_user.uuid}/info",
json={"display_name": "Updated Admin Name"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -866,7 +870,7 @@ class TestAdminUsersInOrg:
"""Updating non-existent user should return 404."""
fake_uuid = uuid7.create()
response = await client.patch(
f"/auth/api/admin/users/{fake_uuid}/display-name",
f"/auth/api/admin/users/{fake_uuid}/info",
json={"display_name": "New Name"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -884,7 +888,7 @@ class TestAdminUsersInOrg:
):
"""Org admin cannot update user from another org."""
response = await client.patch(
f"/auth/api/admin/users/{second_org_user.uuid}/display-name",
f"/auth/api/admin/users/{second_org_user.uuid}/info",
json={"display_name": "New Name"},
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
@@ -896,13 +900,13 @@ class TestAdminUsersInOrg:
):
"""Updating user with empty display name should fail."""
response = await client.patch(
f"/auth/api/admin/users/{test_user.uuid}/display-name",
f"/auth/api/admin/users/{test_user.uuid}/info",
json={"display_name": " "},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
data = response.json()
assert "display_name required" in data["detail"]
assert "display_name cannot be empty" in data["detail"]
@pytest.mark.asyncio
async def test_update_user_display_name_too_long(
@@ -910,13 +914,13 @@ class TestAdminUsersInOrg:
):
"""Updating user with too long display name should fail."""
response = await client.patch(
f"/auth/api/admin/users/{test_user.uuid}/display-name",
json={"display_name": "x" * 100},
f"/auth/api/admin/users/{test_user.uuid}/info",
json={"display_name": "x" * 65},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
data = response.json()
assert "too long" in data["detail"]
assert "display_name too long" in data["detail"]
@pytest.mark.asyncio
async def test_update_user_role_in_org(
@@ -1290,7 +1294,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 +1303,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/{extra_db_key}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1316,8 +1320,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
@@ -1394,14 +1399,17 @@ class TestAdminPermissions:
):
"""Admin should be able to list all permissions."""
response = await client.get(
"/auth/api/admin/permissions",
"/auth/api/admin/info",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert isinstance(data, dict)
assert "permissions" in data
permissions_data = data["permissions"]
assert isinstance(permissions_data, dict)
# Should include at least auth:admin
perm_scopes = [p["scope"] for p in data]
perm_scopes = [p["scope"] for p in permissions_data.values()]
assert "auth:admin" in perm_scopes
@pytest.mark.asyncio
@@ -1414,13 +1422,13 @@ class TestAdminPermissions:
):
"""Org admin should only see permissions their org can grant."""
response = await client.get(
"/auth/api/admin/permissions",
"/auth/api/admin/info",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
data = response.json()
# Should only see permissions the org can grant
perm_scopes = [p["scope"] for p in data]
perm_scopes = [p["scope"] for p in data["permissions"].values()]
assert grantable_permission.scope in perm_scopes
# test_org CAN grant auth:admin (it's in org.permissions), so org admin sees it
assert "auth:admin" in perm_scopes
@@ -1704,7 +1712,7 @@ class TestOrgAdminAuthExceptions:
):
"""Regular user trying to update display name should get 403."""
response = await client.patch(
f"/auth/api/admin/users/{test_user.uuid}/display-name",
f"/auth/api/admin/users/{test_user.uuid}/info",
json={"display_name": "New Name"},
headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
)
tests/test_api.py
+29 -22
View File
@@ -11,15 +11,16 @@ These tests cover:
"""
import secrets
from datetime import timedelta
from datetime import UTC, datetime, timedelta
import httpx
import pytest
from paskia import authcode
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:
@@ -229,12 +230,12 @@ class TestLogoutEndpoint:
class TestUserInfoEndpoint:
"""Tests for POST /auth/api/user-info"""
"""Tests for GET /auth/api/user-info"""
@pytest.mark.asyncio
async def test_user_info_without_auth_returns_401(self, client: httpx.AsyncClient):
"""User info without session should return 401."""
response = await client.post("/auth/api/user-info")
response = await client.get("/auth/api/user-info")
assert response.status_code == 401
@pytest.mark.asyncio
@@ -242,22 +243,22 @@ class TestUserInfoEndpoint:
self, client: httpx.AsyncClient, session_token: str, test_user
):
"""User info with valid session should return user data."""
response = await client.post(
response = await client.get(
"/auth/api/user-info",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
data = response.json()
assert "ctx" in data
assert data["ctx"]["user"]["uuid"] == str(test_user.uuid)
assert data["ctx"]["user"]["display_name"] == test_user.display_name
assert "user" in data
assert data["user"]["uuid"] == str(test_user.uuid)
assert data["user"]["display_name"] == test_user.display_name
@pytest.mark.asyncio
async def test_user_info_includes_credentials(
self, client: httpx.AsyncClient, session_token: str
):
"""User info should include user's credentials."""
response = await client.post(
response = await client.get(
"/auth/api/user-info",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -271,7 +272,7 @@ class TestUserInfoEndpoint:
self, client: httpx.AsyncClient, session_token: str
):
"""User info should include user's active sessions."""
response = await client.post(
response = await client.get(
"/auth/api/user-info",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -285,36 +286,42 @@ class TestUserInfoEndpoint:
self, client: httpx.AsyncClient, session_token: str
):
"""User info should include user's permissions."""
response = await client.post(
response = await client.get(
"/auth/api/user-info",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
data = response.json()
assert "ctx" in data
assert "permissions" in data["ctx"]
assert "permissions" in data
class TestSetSessionEndpoint:
"""Tests for POST /auth/api/set-session"""
@pytest.mark.asyncio
async def test_set_session_without_bearer_returns_401(
async def test_set_session_without_bearer_returns_400(
self, client: httpx.AsyncClient
):
"""Set session without bearer token should return 401."""
"""Set session without bearer token should return 400."""
response = await client.post("/auth/api/set-session")
assert response.status_code == 401
assert response.status_code == 400
@pytest.mark.asyncio
async def test_set_session_with_valid_bearer_token(
self, client: httpx.AsyncClient, session_token: str
):
"""Set session with valid bearer token should set cookie."""
"""Set session with valid auth code as bearer should set cookie."""
code = authcode.store_cookie(
authcode.CookieCode(
session_key=session_token,
created=datetime.now(UTC),
)
)
response = await client.post(
"/auth/api/set-session",
headers={
"Authorization": f"Bearer {session_token}",
"Authorization": f"Bearer {code}",
"Host": "localhost:4401",
},
)
@@ -522,7 +529,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 +539,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