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

Finish the database key-in-object refactoring.

Browse Source
This commit is contained in:
LeoVasanko
2026-01-27 02:11:09 +00:00
parent 0f29544bdb
commit 9bdca1f43a
16 changed files with 415 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files
e2e/tests/10-passkey.spec.ts
+23 -23
View File
@@ -138,9 +138,9 @@ test.describe('Passkey Authentication E2E', () => {
const validation = await validateSession(page, baseUrl, sessionToken)
expect(validation.valid).toBe(true)
expect(validation.user_uuid).toBe(userUuid)
expect(validation.ctx.user.uuid).toBe(userUuid)
console.log(`✓ Session validated for user: ${validation.user_uuid}`)
console.log(`✓ Session validated for user: ${validation.ctx.user.uuid}`)
})
test('should retrieve user info', async ({ page }) => {
@@ -148,8 +148,8 @@ test.describe('Passkey Authentication E2E', () => {
const userInfo = await getUserInfo(page, baseUrl, sessionToken)
expect(userInfo.user.user_uuid).toBe(userUuid)
expect(userInfo.user.user_name).toBe('Admin User')
expect(userInfo.ctx.user.uuid).toBe(userUuid)
expect(userInfo.ctx.user.display_name).toBe('Admin User')
expect(userInfo.credentials).toBeDefined()
expect(userInfo.credentials.length).toBeGreaterThanOrEqual(1)
@@ -169,7 +169,7 @@ test.describe('Passkey Authentication E2E', () => {
await page.screenshot({ path: 'test-results/profile-view.png' })
console.log('✓ Screenshot saved: test-results/profile-view.png')
console.log(`✓ User info retrieved: ${userInfo.user.user_name}`)
console.log(`✓ User info retrieved: ${userInfo.ctx.user.display_name}`)
console.log(`✓ Credentials count: ${userInfo.credentials.length}`)
})
@@ -219,7 +219,7 @@ test.describe('Passkey Authentication E2E', () => {
const validation = await validateSession(page, baseUrl, sessionToken)
expect(validation.valid).toBe(true)
expect(validation.user_uuid).toBe(userUuid)
expect(validation.ctx.user.uuid).toBe(userUuid)
console.log(`✓ New session validated`)
})
@@ -291,8 +291,8 @@ test.describe('Device Addition Dialog', () => {
// Wait for the profile view to load
await page.waitForSelector('[data-view="profile"]', { timeout: 5000 })
// Click the "Add Another Device" button
const addDeviceButton = page.getByRole('button', { name: 'Add Another Device' })
// Click the "Another Device" button
const addDeviceButton = page.getByRole('button', { name: 'Another Device' })
await expect(addDeviceButton).toBeVisible()
await addDeviceButton.click()
@@ -301,7 +301,7 @@ test.describe('Device Addition Dialog', () => {
await expect(dialog).toBeVisible({ timeout: 5000 })
// Verify dialog contains expected elements
await expect(dialog.locator('h2')).toContainText('Device Registration Link')
await expect(dialog.locator('h2')).toContainText('Add Another Device')
// Wait for QR code to be generated (canvas should have content)
const qrCanvas = dialog.locator('.qr-code')
@@ -318,16 +318,16 @@ test.describe('Device Addition Dialog', () => {
expect(linkHref).toContain('http://localhost:4404/auth/')
console.log(`✓ Device link displayed: ${linkText} (href: ${linkHref})`)
// Verify expiration warning is shown
await expect(dialog.locator('.reg-help')).toContainText('Expires')
// Verify help text is shown
await expect(dialog.locator('.reg-help')).toContainText('Scan this QR code')
// Take screenshot of the dialog
await dialog.screenshot({ path: 'test-results/device-addition-dialog.png' })
console.log(`✓ Screenshot saved: test-results/device-addition-dialog.png`)
// Verify Copy Link button exists
const copyButton = dialog.getByRole('button', { name: 'Copy Link' })
await expect(copyButton).toBeVisible()
// Verify the QR link element is clickable (copy functionality is built into clicking it)
const qrLink = dialog.locator('a.qr-link')
await expect(qrLink).toBeVisible()
// Close the dialog (use the text button, not the icon button)
const closeButton = dialog.locator('button.btn-secondary', { hasText: 'Close' })
@@ -357,12 +357,12 @@ test.describe('Device Addition Dialog', () => {
await page.waitForSelector('[data-view="profile"]', { timeout: 5000 })
// Open the dialog
await page.getByRole('button', { name: 'Add Another Device' }).click()
await page.getByRole('button', { name: 'Another Device' }).click()
const dialog = page.locator('.device-dialog')
await expect(dialog).toBeVisible({ timeout: 5000 })
// Extract the reset token from the displayed URL
const linkText = dialog.locator('.qr-link p')
const linkText = dialog.locator('.qr-link .link-text')
const linkContent = await linkText.textContent()
// URL format: localhost/auth/word1.word2.word3.word4.word5
@@ -405,7 +405,7 @@ test.describe('Device Addition Dialog', () => {
})
})
test.describe('ProfileView - Add New Passkey', () => {
test.describe('ProfileView - Register New', () => {
const baseUrl = process.env.BASE_URL || 'http://localhost:4404'
test('should show credentials list in profile', async ({ page }) => {
@@ -427,7 +427,7 @@ test.describe('ProfileView - Add New Passkey', () => {
console.log(`✓ Profile shows ${credentialItems} credential(s) in list`)
})
test('should add a new passkey using Add New Passkey button', async ({ page }) => {
test('should add a new passkey using Register New button', async ({ page }) => {
const sessionToken = getSavedSessionToken()
test.skip(!sessionToken, 'Requires saved session token')
@@ -444,8 +444,8 @@ test.describe('ProfileView - Add New Passkey', () => {
const initialCredentialCount = await page.locator('.credential-item').count()
console.log(`Initial credential count: ${initialCredentialCount}`)
// Click "Add New Passkey" button
const addPasskeyBtn = page.locator('button:has-text("Add New Passkey")')
// Click "Register New" button
const addPasskeyBtn = page.locator('button:has-text("Register New")')
await expect(addPasskeyBtn).toBeVisible()
await addPasskeyBtn.click()
@@ -490,7 +490,7 @@ test.describe('ProfileView - Add New Passkey', () => {
// Try to add a passkey - with excludeCredentials the authenticator should
// prevent re-registration of the same credential
const addPasskeyBtn = page.locator('button:has-text("Add New Passkey")')
const addPasskeyBtn = page.locator('button:has-text("Register New")')
await expect(addPasskeyBtn).toBeVisible()
await addPasskeyBtn.click()
@@ -541,8 +541,8 @@ test.describe('ProfileView - Multi-Authenticator', () => {
await page.waitForSelector('.credential-list', { timeout: 10000 })
const initialCredentialCount = await page.locator('.credential-item').count()
// Click "Add New Passkey" button
const addPasskeyBtn = page.locator('button:has-text("Add New Passkey")')
// Click "Register New" button
const addPasskeyBtn = page.locator('button:has-text("Register New")')
await expect(addPasskeyBtn).toBeVisible()
await addPasskeyBtn.click()
e2e/tests/20-api-auth.spec.ts
+1 -1
View File
@@ -268,7 +268,7 @@ test.describe('API Mode - 401 Login Flow', () => {
// Wait for API call to complete and verify result
const result = await apiCallPromise
expect(result.status).toBe(200)
expect(result.data.user).toBeDefined()
expect(result.data.ctx).toBeDefined()
console.log('✓ API call succeeded after authentication')
// Save the session for other tests
e2e/tests/fixtures/passkey-helpers.ts
Vendored
+36 -2
View File
@@ -23,6 +23,40 @@ export interface AuthenticationResult {
session_token: string
}
export interface SessionContext {
user: { uuid: string; display_name: string }
org: { uuid: string; display_name: string }
role: { uuid: string; display_name: string }
permissions: string[]
}
export interface UserInfo {
ctx: SessionContext
created_at: string
last_seen: string
visits: number
credentials: Array<{
credential_uuid: string
aaguid: string
created_at: string
last_used: string | null
last_verified: string | null
sign_count: number
is_current_session: boolean
}>
aaguid_info: Record<string, { name: string; icon_light?: string; icon_dark?: string }>
sessions: Array<{
id: string
credential_uuid: string
host: string
ip: string
user_agent: string
last_renewed: string
is_current: boolean
is_current_host: boolean
}>
}
/**
* Get the bootstrap reset token from the test state file.
*/
@@ -376,7 +410,7 @@ export async function validateSession(
page: Page,
baseUrl: string,
sessionToken: string
): Promise<{ valid: boolean; user_uuid: string; renewed: boolean }> {
): Promise<{ valid: boolean; ctx: SessionContext; renewed: boolean }> {
const cookieName = getSessionCookieName()
const response = await page.request.post(`${baseUrl}/auth/api/validate`, {
headers: {
@@ -393,7 +427,7 @@ export async function getUserInfo(
page: Page,
baseUrl: string,
sessionToken: string
): Promise<any> {
): Promise<UserInfo> {
const cookieName = getSessionCookieName()
const response = await page.request.post(`${baseUrl}/auth/api/user-info`, {
headers: {
e2e/tests/global-setup.ts
+6 -4
View File
@@ -42,21 +42,23 @@ export default async function globalSetup() {
const serverArgs = COLLECT_COVERAGE
? [
'run', 'coverage', 'run', '--parallel-mode',
'-m', 'paskia.fastapi', 'serve', 'localhost:4404',
'-m', 'paskia.fastapi', 'localhost:4404',
'--rp-id', 'localhost'
]
: [
'run', 'paskia', 'serve', 'localhost:4404',
'run', 'paskia', 'localhost:4404',
'--rp-id', 'localhost'
]
// Use a temporary jsonl file for test database
const testDbFile = join(testDataDir, 'test-db.jsonl')
// Start the server using Node's spawn
// Use in-memory SQLite for faster tests
const serverProcess = spawn('uv', serverArgs, {
cwd: projectRoot,
env: {
...process.env,
PASKIA_DB: 'sqlite+aiosqlite:///:memory:',
PASKIA_DB: testDbFile,
COVERAGE_FILE: join(projectRoot, '.coverage'),
},
stdio: ['ignore', 'pipe', 'pipe'],
e2e/tests/global-teardown.ts
+5 -12
View File
@@ -59,18 +59,11 @@ export default async function globalTeardown() {
rmSync(stateFile, { force: true })
}
// Optionally clean up test database (keep it for debugging by default)
if (process.env.CLEANUP_TEST_DB === 'true') {
const dbPath = join(testDataDir, 'test.sqlite')
if (existsSync(dbPath)) {
console.log(' Removing test database...')
rmSync(dbPath)
}
// Remove wal/shm files too
for (const ext of ['-wal', '-shm']) {
const file = dbPath + ext
if (existsSync(file)) rmSync(file)
}
// Clean up test database
const testDbFile = join(testDataDir, 'test-db.jsonl')
if (existsSync(testDbFile)) {
console.log(' Removing test database...')
rmSync(testDbFile)
}
// Generate Python coverage report if coverage was collected
paskia/db/operations.py
+35 -77
View File
@@ -37,13 +37,9 @@ from paskia.db.structs import (
Session,
SessionContext,
User,
_CredentialData,
_DatabaseData,
_OrgData,
_PermissionData,
_ResetTokenData,
_RoleData,
_SessionData,
)
from paskia.util.passphrase import is_well_formed as _is_passphrase
@@ -170,10 +166,9 @@ async def init(*args, **kwargs):
def build_permission(uuid: UUID) -> Permission:
p = _db._data.permissions[uuid]
return Permission(
uuid=uuid, scope=p.scope, display_name=p.display_name, domain=p.domain
)
perm = _db._data.permissions[uuid]
perm.uuid = uuid
return perm
def build_user(uuid: UUID) -> User:
@@ -184,12 +179,13 @@ def build_user(uuid: UUID) -> User:
def build_role(uuid: UUID) -> Role:
r = _db._data.roles[uuid]
return Role(
uuid=uuid,
org_uuid=r.org,
role = Role(
org=r.org,
display_name=r.display_name,
permissions=[str(pid) for pid in r.permissions.keys()],
)
role.uuid = uuid
return role
def build_org(uuid: UUID, include_roles: bool = False) -> Org:
@@ -197,7 +193,8 @@ def build_org(uuid: UUID, include_roles: bool = False) -> Org:
perm_uuids = [
str(pid) for pid, p in _db._data.permissions.items() if uuid in p.orgs
]
org = Org(uuid=uuid, display_name=o.display_name, permissions=perm_uuids)
org = Org(display_name=o.display_name, permissions=perm_uuids)
org.uuid = uuid
if include_roles:
org.roles = [
build_role(rid) for rid, r in _db._data.roles.items() if r.org == uuid
@@ -206,41 +203,21 @@ def build_org(uuid: UUID, include_roles: bool = False) -> Org:
def build_credential(uuid: UUID) -> Credential:
c = _db._data.credentials[uuid]
return Credential(
uuid=uuid,
credential_id=c.credential_id,
user_uuid=c.user,
aaguid=c.aaguid,
public_key=c.public_key,
sign_count=c.sign_count,
created_at=c.created_at,
last_used=c.last_used,
last_verified=c.last_verified,
)
cred = _db._data.credentials[uuid]
cred.uuid = uuid
return cred
def build_session(key: str) -> Session:
s = _db._data.sessions[key]
return Session(
key=key,
user_uuid=s.user,
credential_uuid=s.credential,
host=s.host,
ip=s.ip,
user_agent=s.user_agent,
expiry=s.expiry,
)
s.key = key
return s
def build_reset_token(key: bytes) -> ResetToken:
t = _db._data.reset_tokens[key]
return ResetToken(
key=key,
user_uuid=t.user,
expiry=t.expiry,
token_type=t.token_type,
)
t.key = key
return t
# -------------------------------------------------------------------------
@@ -520,12 +497,7 @@ def create_permission(perm: Permission, *, ctx: SessionContext | None = None) ->
if perm.uuid in _db._data.permissions:
raise ValueError(f"Permission {perm.uuid} already exists")
with _db.transaction("Created permission", ctx):
_db._data.permissions[perm.uuid] = _PermissionData(
scope=perm.scope,
display_name=perm.display_name,
domain=perm.domain,
orgs={},
)
_db._data.permissions[perm.uuid] = perm
def update_permission(perm: Permission, *, ctx: SessionContext | None = None) -> None:
@@ -724,11 +696,11 @@ def create_role(role: Role, *, ctx: SessionContext | None = None) -> None:
"""Create a new role."""
if role.uuid in _db._data.roles:
raise ValueError(f"Role {role.uuid} already exists")
if role.org_uuid not in _db._data.orgs:
raise ValueError(f"Organization {role.org_uuid} not found")
if role.org not in _db._data.orgs:
raise ValueError(f"Organization {role.org} not found")
with _db.transaction("Created role", ctx):
_db._data.roles[role.uuid] = _RoleData(
org=role.org_uuid,
org=role.org,
display_name=role.display_name,
permissions={UUID(pid): True for pid in role.permissions},
)
@@ -901,19 +873,10 @@ def create_credential(cred: Credential, *, ctx: SessionContext | None = None) ->
"""Create a new credential."""
if cred.uuid in _db._data.credentials:
raise ValueError(f"Credential {cred.uuid} already exists")
if cred.user_uuid not in _db._data.users:
raise ValueError(f"User {cred.user_uuid} not found")
if cred.user not in _db._data.users:
raise ValueError(f"User {cred.user} not found")
with _db.transaction("Added credential", ctx):
_db._data.credentials[cred.uuid] = _CredentialData(
credential_id=cred.credential_id,
user=cred.user_uuid,
aaguid=cred.aaguid,
public_key=cred.public_key,
sign_count=cred.sign_count,
created_at=cred.created_at,
last_used=cred.last_used,
last_verified=cred.last_verified,
)
_db._data.credentials[cred.uuid] = cred
def update_credential_sign_count(
@@ -981,7 +944,7 @@ def create_session(
if credential_uuid not in _db._data.credentials:
raise ValueError(f"Credential {credential_uuid} not found")
with _db.transaction("Created session", ctx):
_db._data.sessions[key] = _SessionData(
_db._data.sessions[key] = Session(
user=user_uuid,
credential=credential_uuid,
host=host,
@@ -1073,7 +1036,7 @@ def create_reset_token(
# For self-service, derive user from the user_uuid param
user_str = str(user_uuid) if not ctx else None
with _db.transaction("Created reset token", ctx, user=user_str):
_db._data.reset_tokens[key] = _ResetTokenData(
_db._data.reset_tokens[key] = ResetToken(
user=user_uuid, expiry=expiry, token_type=token_type
)
@@ -1155,7 +1118,7 @@ def login(
_db._data.credentials[credential.uuid].sign_count = credential.sign_count
_db._data.credentials[credential.uuid].last_used = now
# Create session
_db._data.sessions[session_key] = _SessionData(
_db._data.sessions[session_key] = Session(
user=user_uuid,
credential=credential.uuid,
host=host,
@@ -1201,19 +1164,10 @@ def create_credential_session(
_db._data.users[user_uuid].display_name = display_name
# Create credential
_db._data.credentials[credential.uuid] = _CredentialData(
credential_id=credential.credential_id,
user=user_uuid,
aaguid=credential.aaguid,
public_key=credential.public_key,
sign_count=credential.sign_count,
created_at=credential.created_at,
last_used=credential.last_used,
last_verified=credential.last_verified,
)
_db._data.credentials[credential.uuid] = credential
# Create session
_db._data.sessions[session_key] = _SessionData(
_db._data.sessions[session_key] = Session(
user=user_uuid,
credential=credential.uuid,
host=host,
@@ -1291,18 +1245,22 @@ def bootstrap(
with _db.transaction("bootstrap"):
# Create auth:admin permission
_db._data.permissions[perm_admin_uuid] = _PermissionData(
perm_admin = Permission(
scope="auth:admin",
display_name="Master Admin",
orgs={org_uuid: True}, # Grant to org
)
perm_admin.uuid = perm_admin_uuid
_db._data.permissions[perm_admin_uuid] = perm_admin
# Create auth:org:admin permission
_db._data.permissions[perm_org_admin_uuid] = _PermissionData(
perm_org_admin = Permission(
scope="auth:org:admin",
display_name="Org Admin",
orgs={org_uuid: True}, # Grant to org
)
perm_org_admin.uuid = perm_org_admin_uuid
_db._data.permissions[perm_org_admin_uuid] = perm_org_admin
# Create organization
_db._data.orgs[org_uuid] = _OrgData(
@@ -1329,7 +1287,7 @@ def bootstrap(
_db._data.users[user_uuid] = admin_user
# Create reset token
_db._data.reset_tokens[reset_key] = _ResetTokenData(
_db._data.reset_tokens[reset_key] = ResetToken(
user=user_uuid,
expiry=reset_expiry,
token_type="admin bootstrap",
paskia/db/structs.py
+126 -54
View File
@@ -5,26 +5,84 @@ import msgspec
import uuid7
class Permission(msgspec.Struct, omit_defaults=True):
uuid: UUID # UUID primary key
class Permission(msgspec.Struct, dict=True, omit_defaults=True):
scope: str # Permission scope identifier (e.g. "auth:admin", "myapp:write")
display_name: str
domain: str | None = None # If set, scopes permission to this domain
orgs: dict[UUID, bool] = {} # org_uuid -> True (which orgs can grant this)
def __post_init__(self):
self.uuid: UUID | None = None # Convenience field, not serialized
@classmethod
def create(
cls,
scope: str,
display_name: str,
domain: str | None = None,
) -> "Permission":
"""Create a new Permission with auto-generated uuid7."""
perm = cls(
scope=scope,
display_name=display_name,
domain=domain,
)
perm.uuid = uuid7.create()
return perm
class Role(msgspec.Struct):
uuid: UUID
org_uuid: UUID
class Role(msgspec.Struct, dict=True):
org: UUID
display_name: str
permissions: list[str] = [] # permission UUIDs this role grants
def __post_init__(self):
self.uuid: UUID | None = None # Convenience field, not serialized
class Org(msgspec.Struct):
uuid: UUID
@classmethod
def create(
cls,
org: UUID,
display_name: str,
permissions: list[str] | None = None,
) -> "Role":
"""Create a new Role with auto-generated uuid7."""
role = cls(
org=org,
display_name=display_name,
permissions=permissions or [],
)
role.uuid = uuid7.create()
return role
# Legacy alias for org field
@property
def org_uuid(self) -> UUID:
return self.org
class Org(msgspec.Struct, dict=True):
display_name: str
permissions: list[str] = [] # permission UUIDs this org can grant
roles: list[Role] = [] # roles belonging to this org
def __post_init__(self):
self.uuid: UUID | None = None # Convenience field, not serialized
@classmethod
def create(
cls,
display_name: str,
permissions: list[str] | None = None,
) -> "Org":
"""Create a new Org with auto-generated uuid7."""
org = cls(
display_name=display_name,
permissions=permissions or [],
)
org.uuid = uuid7.create()
return org
class User(msgspec.Struct, dict=True):
display_name: str
@@ -55,10 +113,9 @@ class User(msgspec.Struct, dict=True):
return user
class Credential(msgspec.Struct):
uuid: UUID
class Credential(msgspec.Struct, dict=True):
credential_id: bytes # Long binary ID from the authenticator
user_uuid: UUID
user: UUID
aaguid: UUID
public_key: bytes
sign_count: int
@@ -66,16 +123,57 @@ class Credential(msgspec.Struct):
last_used: datetime | None = None
last_verified: datetime | None = None
def __post_init__(self):
self.uuid: UUID | None = None # Convenience field, not serialized
class Session(msgspec.Struct):
key: str
user_uuid: UUID
credential_uuid: UUID
@classmethod
def create(
cls,
credential_id: bytes,
user: UUID,
aaguid: UUID,
public_key: bytes,
sign_count: int,
created_at: datetime | None = None,
) -> "Credential":
"""Create a new Credential with auto-generated uuid7."""
from datetime import timezone
now = created_at or datetime.now(timezone.utc)
cred = cls(
credential_id=credential_id,
user=user,
aaguid=aaguid,
public_key=public_key,
sign_count=sign_count,
created_at=now,
last_used=now,
last_verified=now,
)
cred.uuid = uuid7.create(now)
return cred
class Session(msgspec.Struct, dict=True):
user: UUID
credential: UUID
host: str | None
ip: str | None
user_agent: str | None
expiry: datetime
def __post_init__(self):
self.key: str | None = None # Convenience field, not serialized
# Legacy aliases
@property
def user_uuid(self) -> UUID:
return self.user
@property
def credential_uuid(self) -> UUID:
return self.credential
def metadata(self) -> dict:
"""Return session metadata for backwards compatibility."""
return {
@@ -85,12 +183,19 @@ class Session(msgspec.Struct):
}
class ResetToken(msgspec.Struct):
key: bytes
user_uuid: UUID
class ResetToken(msgspec.Struct, dict=True):
user: UUID
expiry: datetime
token_type: str
def __post_init__(self):
self.key: bytes | None = None # Convenience field, not serialized
# Legacy alias
@property
def user_uuid(self) -> UUID:
return self.user
class SessionContext(msgspec.Struct):
session: Session
@@ -106,13 +211,6 @@ class SessionContext(msgspec.Struct):
# -------------------------------------------------------------------------
class _PermissionData(msgspec.Struct, omit_defaults=True):
scope: str # Permission scope identifier
display_name: str
domain: str | None = None
orgs: dict[UUID, bool] = {} # org_uuid -> True (which orgs can grant this)
class _OrgData(msgspec.Struct):
display_name: str
created_at: datetime | None = None
@@ -124,38 +222,12 @@ class _RoleData(msgspec.Struct):
permissions: dict[UUID, bool] = {} # permission_uuid -> True
class _CredentialData(msgspec.Struct):
credential_id: bytes
user: UUID
aaguid: UUID
public_key: bytes
sign_count: int
created_at: datetime
last_used: datetime | None
last_verified: datetime | None
class _SessionData(msgspec.Struct):
user: UUID
credential: UUID
host: str | None
ip: str | None
user_agent: str | None
expiry: datetime
class _ResetTokenData(msgspec.Struct):
user: UUID
expiry: datetime
token_type: str
class _DatabaseData(msgspec.Struct, omit_defaults=True):
permissions: dict[UUID, _PermissionData]
permissions: dict[UUID, Permission]
orgs: dict[UUID, _OrgData]
roles: dict[UUID, _RoleData]
users: dict[UUID, User]
credentials: dict[UUID, _CredentialData]
sessions: dict[str, _SessionData]
reset_tokens: dict[bytes, _ResetTokenData]
credentials: dict[UUID, Credential]
sessions: dict[str, Session]
reset_tokens: dict[bytes, ResetToken]
v: int = 0
paskia/fastapi/admin.py
+12 -21
View File
@@ -135,13 +135,12 @@ async def admin_create_org(
)
from ..db import Org as OrgDC # local import to avoid cycles
org_uuid = uuid4()
display_name = payload.get("display_name") or "New Organization"
permissions = payload.get("permissions") or []
org = OrgDC(uuid=org_uuid, display_name=display_name, permissions=permissions)
org = OrgDC.create(display_name=display_name, permissions=permissions)
db.create_organization(org, ctx=ctx)
return {"uuid": str(org_uuid)}
return {"uuid": str(org.uuid)}
@app.patch("/orgs/{org_uuid}")
@@ -264,7 +263,6 @@ async def admin_create_role(
)
from ..db import Role as RoleDC
role_uuid = uuid4()
display_name = payload.get("display_name") or "New Role"
perms = payload.get("permissions") or []
org = db.get_organization(str(org_uuid))
@@ -281,14 +279,13 @@ async def admin_create_role(
raise ValueError(f"Permission not grantable by org: {pid}")
permission_uuids.append(perm_uuid_str)
role = RoleDC(
uuid=role_uuid,
org_uuid=org_uuid,
role = RoleDC.create(
org=org_uuid,
display_name=display_name,
permissions=permission_uuids,
)
db.create_role(role, ctx=ctx)
return {"uuid": str(role_uuid)}
return {"uuid": str(role.uuid)}
@app.patch("/orgs/{org_uuid}/roles/{role_uuid}")
@@ -939,8 +936,6 @@ async def admin_create_permission(
match=permutil.has_all,
max_age="5m",
)
import uuid7
from ..db import Permission as PermDC
scope = payload.get("scope") or payload.get(
@@ -953,9 +948,7 @@ async def admin_create_permission(
querysafe.assert_safe(scope, field="scope")
_validate_permission_domain(domain)
db.create_permission(
PermDC(
uuid=uuid7.create(), scope=scope, display_name=display_name, domain=domain
),
PermDC.create(scope=scope, display_name=display_name, domain=domain),
ctx=ctx,
)
return {"status": "ok"}
@@ -999,15 +992,13 @@ async def admin_update_permission(
from ..db import Permission as PermDC
db.update_permission(
PermDC(
uuid=perm.uuid,
scope=new_scope,
display_name=new_display_name,
domain=domain_value,
),
ctx=ctx,
updated_perm = PermDC(
scope=new_scope,
display_name=new_display_name,
domain=domain_value,
)
updated_perm.uuid = perm.uuid
db.update_permission(updated_perm, ctx=ctx)
return {"status": "ok"}
paskia/fastapi/remote.py
+4 -4
View File
@@ -347,7 +347,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
token_str = passphrase.generate()
expiry = expires()
db.create_reset_token(
user_uuid=stored_cred.user_uuid,
user_uuid=stored_cred.user,
passphrase=token_str,
expiry=expiry,
token_type="device addition",
@@ -356,7 +356,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
# Also create a session so the device is logged in
normalized_host = hostutil.normalize_host(request.host)
session_token = db.login(
user_uuid=stored_cred.user_uuid,
user_uuid=stored_cred.user,
credential=stored_cred,
host=normalized_host,
ip=request.ip,
@@ -370,7 +370,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
normalized_host = hostutil.normalize_host(request.host)
session_token = db.login(
user_uuid=stored_cred.user_uuid,
user_uuid=stored_cred.user,
credential=stored_cred,
host=normalized_host,
ip=request.ip,
@@ -382,7 +382,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
completed = await remoteauth.instance.complete_request(
token=request.key,
session_token=session_token,
user_uuid=stored_cred.user_uuid,
user_uuid=stored_cred.user,
credential_uuid=stored_cred.uuid,
reset_token=reset_token,
)
paskia/fastapi/ws.py
+3 -3
View File
@@ -136,7 +136,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
)
# If reauth mode, verify the credential belongs to the session's user
if session_user_uuid and stored_cred.user_uuid != session_user_uuid:
if session_user_uuid and stored_cred.user != session_user_uuid:
raise ValueError("This passkey belongs to a different account")
# Verify the credential matches the stored data
@@ -154,7 +154,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
raise ValueError(f"Host must be the same as or a subdomain of {rp_id}")
token = db.login(
user_uuid=stored_cred.user_uuid,
user_uuid=stored_cred.user,
credential=stored_cred,
host=normalized_host,
ip=metadata.get("ip") or "",
@@ -164,7 +164,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
await ws.send_json(
{
"user_uuid": str(stored_cred.user_uuid),
"user_uuid": str(stored_cred.user),
"session_token": token,
}
)
paskia/migrate/__init__.py
+25 -19
View File
@@ -55,13 +55,13 @@ async def migrate_from_sql(
from paskia.db.operations import DB as JSONDB
from paskia.db.structs import (
Credential,
Permission,
ResetToken,
Session,
User,
_CredentialData,
_OrgData,
_PermissionData,
_ResetTokenData,
_RoleData,
_SessionData,
)
# Initialize source SQL database
@@ -90,11 +90,13 @@ async def migrate_from_sql(
# Migrate permissions with UUID keys and scope field
# Always create exactly one common auth:org:admin permission for all org admin needs
org_admin_perm_uuid: UUID = uuid7.create()
json_db._data.permissions[org_admin_perm_uuid] = _PermissionData(
org_admin_perm = Permission(
scope="auth:org:admin",
display_name="Org Admin",
orgs={},
)
org_admin_perm.uuid = org_admin_perm_uuid
json_db._data.permissions[org_admin_perm_uuid] = org_admin_perm
# Mapping from old permission ID to new permission UUID
perm_id_to_uuid: dict[str, UUID] = {}
@@ -113,11 +115,13 @@ async def migrate_from_sql(
# Regular permission - create with UUID key
perm_uuid: UUID = uuid7.create()
json_db._data.permissions[perm_uuid] = _PermissionData(
new_perm = Permission(
scope=perm.id, # Old ID becomes the scope
display_name=perm.display_name,
orgs={},
)
new_perm.uuid = perm_uuid
json_db._data.permissions[perm_uuid] = new_perm
perm_id_to_uuid[perm.id] = perm_uuid
print(
f" Migrated {len(permissions)} permissions (with {len(org_admin_uuids)} org-specific admins consolidated to auth:org:admin)"
@@ -181,18 +185,20 @@ async def migrate_from_sql(
result = await session.execute(select(CredentialModel))
cred_models = result.scalars().all()
for cm in cred_models:
cred = cm.as_dataclass()
cred_key: UUID = cred.uuid
json_db._data.credentials[cred_key] = _CredentialData(
credential_id=cred.credential_id,
user=cred.user_uuid,
aaguid=cred.aaguid,
public_key=cred.public_key,
sign_count=cred.sign_count,
created_at=cred.created_at,
last_used=cred.last_used,
last_verified=cred.last_verified,
legacy_cred = cm.as_dataclass()
cred_key: UUID = legacy_cred.uuid
new_cred = Credential(
credential_id=legacy_cred.credential_id,
user=legacy_cred.user_uuid,
aaguid=legacy_cred.aaguid,
public_key=legacy_cred.public_key,
sign_count=legacy_cred.sign_count,
created_at=legacy_cred.created_at,
last_used=legacy_cred.last_used,
last_verified=legacy_cred.last_verified,
)
new_cred.uuid = cred_key
json_db._data.credentials[cred_key] = new_cred
print(f" Migrated {len(cred_models)} credentials")
# Migrate sessions
@@ -209,7 +215,7 @@ async def migrate_from_sql(
else:
# Already in new format or unknown - try to use as-is
session_key = base64url.enc(old_key[:12])
json_db._data.sessions[session_key] = _SessionData(
json_db._data.sessions[session_key] = Session(
user=sess.user_uuid,
credential=sess.credential_uuid,
host=sess.host,
@@ -233,7 +239,7 @@ async def migrate_from_sql(
else:
# Already in new format or unknown - truncate to 9 bytes
token_key = old_key[:9]
json_db._data.reset_tokens[token_key] = _ResetTokenData(
json_db._data.reset_tokens[token_key] = ResetToken(
user=token.user_uuid,
expiry=token.expiry,
token_type=token.token_type,
paskia/migrate/sql.py
+62 -25
View File
@@ -26,10 +26,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from paskia.db import (
Credential,
Org,
ResetToken,
Role,
)
@@ -46,6 +43,58 @@ class _LegacyUser:
visits: int = 0
# Legacy Credential class for SQL schema (uses 'user_uuid' not 'user')
@dataclass
class _LegacyCredential:
"""Credential as stored in the old SQL schema with user_uuid field."""
uuid: UUID
credential_id: bytes
user_uuid: UUID
aaguid: UUID
public_key: bytes
sign_count: int
created_at: datetime
last_used: datetime | None = None
last_verified: datetime | None = None
# Legacy Role class for SQL schema (uses 'org_uuid' not 'org')
@dataclass
class _LegacyRole:
"""Role as stored in the old SQL schema with org_uuid field."""
uuid: UUID
org_uuid: UUID
display_name: str
permissions: list[str] | None = None
# Legacy Session class for SQL schema (uses 'key' as field, 'user_uuid', 'credential_uuid')
@dataclass
class _LegacySession:
"""Session as stored in the old SQL schema."""
key: bytes
user_uuid: UUID
credential_uuid: UUID
host: str
ip: str
user_agent: str
renewed: datetime
# Legacy ResetToken class for SQL schema (uses 'key' as field, 'user_uuid')
@dataclass
class _LegacyResetToken:
"""ResetToken as stored in the old SQL schema."""
key: bytes
user_uuid: UUID
token_type: str
expiry: datetime
# Local Permission class for SQL schema (uses 'id' not 'uuid' + 'scope')
@dataclass
class SqlPermission:
@@ -58,20 +107,6 @@ class SqlPermission:
DB_PATH_DEFAULT = "sqlite+aiosqlite:///paskia.sqlite"
# Local Session class for SQL schema (uses 'renewed' not 'expiry')
@dataclass
class _SqlSession:
"""Session as stored in the old SQL schema with renewed timestamp."""
key: bytes
user_uuid: UUID
credential_uuid: UUID
host: str
ip: str
user_agent: str
renewed: datetime
def _normalize_dt(value: datetime | None) -> datetime | None:
if value is None:
return None
@@ -92,7 +127,9 @@ class OrgModel(Base):
def as_dataclass(self):
# Base Org without permissions/roles (filled by data accessors)
return Org(UUID(bytes=self.uuid), self.display_name)
org = Org(display_name=self.display_name)
org.uuid = UUID(bytes=self.uuid)
return org
@staticmethod
def from_dataclass(org: Org):
@@ -110,14 +147,14 @@ class RoleModel(Base):
def as_dataclass(self):
# Base Role without permissions (filled by data accessors)
return Role(
return _LegacyRole(
uuid=UUID(bytes=self.uuid),
org_uuid=UUID(bytes=self.org_uuid),
display_name=self.display_name,
)
@staticmethod
def from_dataclass(role: Role):
def from_dataclass(role: _LegacyRole):
return RoleModel(
uuid=role.uuid.bytes,
org_uuid=role.org_uuid.bytes,
@@ -187,7 +224,7 @@ class CredentialModel(Base):
)
def as_dataclass(self):
return Credential(
return _LegacyCredential(
uuid=UUID(bytes=self.uuid),
credential_id=self.credential_id,
user_uuid=UUID(bytes=self.user_uuid),
@@ -222,7 +259,7 @@ class SessionModel(Base):
)
def as_dataclass(self):
return _SqlSession(
return _LegacySession(
key=self.key,
user_uuid=UUID(bytes=self.user_uuid),
credential_uuid=UUID(bytes=self.credential_uuid),
@@ -233,7 +270,7 @@ class SessionModel(Base):
)
@staticmethod
def from_dataclass(session: _SqlSession):
def from_dataclass(session: _LegacySession):
return SessionModel(
key=session.key,
user_uuid=session.user_uuid.bytes,
@@ -255,8 +292,8 @@ class ResetTokenModel(Base):
token_type: Mapped[str] = mapped_column(String, nullable=False)
expiry: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
def as_dataclass(self) -> ResetToken:
return ResetToken(
def as_dataclass(self) -> _LegacyResetToken:
return _LegacyResetToken(
key=self.key,
user_uuid=UUID(bytes=self.user_uuid),
token_type=self.token_type,
paskia/sansio.py
+2 -4
View File
@@ -176,14 +176,12 @@ class Passkey:
expected_origin=origin,
expected_rp_id=self.rp_id,
)
return Credential(
uuid=uuid7.create(),
return Credential.create(
credential_id=credential.raw_id,
user_uuid=user_uuid,
user=user_uuid,
aaguid=UUID(registration.aaguid),
public_key=registration.credential_public_key,
sign_count=registration.sign_count,
created_at=datetime.now(timezone.utc),
)
### Authentication Methods ###
tests/conftest.py
+15 -41
View File
@@ -84,8 +84,7 @@ async def passkey_instance() -> Passkey:
@pytest_asyncio.fixture(scope="function")
async def test_org(test_db: DB, admin_permission: Permission) -> Org:
"""Create a test organization with admin permission."""
org = Org(
uuid=uuid7.create(),
org = Org.create(
display_name="Test Organization",
permissions=[str(admin_permission.uuid)], # Org can grant this permission
)
@@ -96,11 +95,7 @@ async def test_org(test_db: DB, admin_permission: Permission) -> Org:
@pytest_asyncio.fixture(scope="function")
async def admin_permission(test_db: DB) -> Permission:
"""Create the auth:admin permission."""
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="auth:admin", display_name="Master Admin"
)
perm = Permission.create(scope="auth:admin", display_name="Master Admin")
create_permission(perm)
return perm
@@ -108,11 +103,7 @@ async def admin_permission(test_db: DB) -> Permission:
@pytest_asyncio.fixture(scope="function")
async def org_admin_permission(test_db: DB, test_org: Org) -> Permission:
"""Create the auth:org:admin permission."""
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="auth:org:admin", display_name="Organization Admin"
)
perm = Permission.create(scope="auth:org:admin", display_name="Organization Admin")
create_permission(perm)
# Make it grantable by the org
add_permission_to_organization(str(test_org.uuid), "auth:org:admin")
@@ -127,9 +118,8 @@ async def test_role(
org_admin_permission: Permission,
) -> Role:
"""Create a test role with admin permission."""
role = Role(
uuid=uuid7.create(),
org_uuid=test_org.uuid,
role = Role.create(
org=test_org.uuid,
display_name="Test Admin Role",
permissions=[str(admin_permission.uuid), str(org_admin_permission.uuid)],
)
@@ -140,11 +130,9 @@ async def test_role(
@pytest_asyncio.fixture(scope="function")
async def user_role(test_db: DB, test_org: Org) -> Role:
"""Create a test role without admin permission (regular user)."""
role = Role(
uuid=uuid7.create(),
org_uuid=test_org.uuid,
role = Role.create(
org=test_org.uuid,
display_name="User Role",
permissions=[],
)
create_role(role)
return role
@@ -153,12 +141,9 @@ async def user_role(test_db: DB, test_org: Org) -> Role:
@pytest_asyncio.fixture(scope="function")
async def test_user(test_db: DB, test_role: Role) -> User:
"""Create a test user with admin role."""
user = User(
uuid=uuid7.create(),
user = User.create(
display_name="Test Admin",
role_uuid=test_role.uuid,
created_at=datetime.now(timezone.utc),
visits=0,
role=test_role.uuid,
)
create_user(user)
return user
@@ -167,12 +152,9 @@ async def test_user(test_db: DB, test_role: Role) -> User:
@pytest_asyncio.fixture(scope="function")
async def regular_user(test_db: DB, user_role: Role) -> User:
"""Create a regular test user without admin permissions."""
user = User(
uuid=uuid7.create(),
user = User.create(
display_name="Regular User",
role_uuid=user_role.uuid,
created_at=datetime.now(timezone.utc),
visits=0,
role=user_role.uuid,
)
create_user(user)
return user
@@ -181,16 +163,12 @@ async def regular_user(test_db: DB, user_role: Role) -> User:
@pytest_asyncio.fixture(scope="function")
async def test_credential(test_db: DB, test_user: User) -> Credential:
"""Create a test credential for the admin user."""
credential = Credential(
uuid=uuid7.create(),
credential = Credential.create(
credential_id=os.urandom(32),
user_uuid=test_user.uuid,
user=test_user.uuid,
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=os.urandom(64),
sign_count=0,
created_at=datetime.now(timezone.utc),
last_used=None,
last_verified=None,
)
create_credential(credential)
return credential
@@ -199,16 +177,12 @@ async def test_credential(test_db: DB, test_user: User) -> Credential:
@pytest_asyncio.fixture(scope="function")
async def regular_credential(test_db: DB, regular_user: User) -> Credential:
"""Create a test credential for the regular user."""
credential = Credential(
uuid=uuid7.create(),
credential = Credential.create(
credential_id=os.urandom(32),
user_uuid=regular_user.uuid,
user=regular_user.uuid,
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=os.urandom(64),
sign_count=0,
created_at=datetime.now(timezone.utc),
last_used=None,
last_verified=None,
)
create_credential(credential)
return credential
tests/test_admin.py
+30 -92
View File
@@ -43,10 +43,8 @@ from tests.conftest import auth_headers
@pytest_asyncio.fixture(scope="function")
async def second_org(test_db: DB) -> Org:
"""Create a second organization for deletion tests."""
org = Org(
uuid=uuid7.create(),
org = Org.create(
display_name="Second Organization",
permissions=[],
)
create_organization(org)
return org
@@ -57,9 +55,8 @@ async def second_org_role(
test_db: DB, second_org: Org, admin_permission: Permission
) -> Role:
"""Create a role in the second org with admin permission."""
role = Role(
uuid=uuid7.create(),
org_uuid=second_org.uuid,
role = Role.create(
org=second_org.uuid,
display_name="Second Org Admin Role",
permissions=[str(admin_permission.uuid)],
)
@@ -70,12 +67,9 @@ async def second_org_role(
@pytest_asyncio.fixture(scope="function")
async def second_org_user(test_db: DB, second_org_role: Role) -> User:
"""Create a user in the second org."""
user = User(
uuid=uuid7.create(),
user = User.create(
display_name="Second Org User",
role_uuid=second_org_role.uuid,
created_at=datetime.now(timezone.utc),
visits=0,
role=second_org_role.uuid,
)
create_user(user)
return user
@@ -86,16 +80,12 @@ async def second_org_credential(test_db: DB, second_org_user: User) -> Credentia
"""Create a credential for the second org user."""
import os
credential = Credential(
uuid=uuid7.create(),
credential = Credential.create(
credential_id=os.urandom(32),
user_uuid=second_org_user.uuid,
user=second_org_user.uuid,
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=os.urandom(64),
sign_count=0,
created_at=datetime.now(timezone.utc),
last_used=datetime.now(timezone.utc),
last_verified=datetime.now(timezone.utc),
)
create_credential(credential)
return credential
@@ -124,9 +114,8 @@ async def org_admin_role(
test_db: DB, test_org: Org, org_admin_permission: Permission
) -> Role:
"""Create a role with org admin permission only (no global admin)."""
role = Role(
uuid=uuid7.create(),
org_uuid=test_org.uuid,
role = Role.create(
org=test_org.uuid,
display_name="Org Admin Role",
permissions=[str(org_admin_permission.uuid)],
)
@@ -137,14 +126,12 @@ async def org_admin_role(
@pytest_asyncio.fixture(scope="function")
async def org_admin_user(test_db: DB, org_admin_role: Role) -> User:
"""Create a user with org admin permission only."""
user = User(
uuid=uuid7.create(),
user = User.create(
display_name="Org Admin User",
role_uuid=org_admin_role.uuid,
created_at=datetime.now(timezone.utc),
visits=5,
last_seen=datetime.now(timezone.utc),
role=org_admin_role.uuid,
)
user.visits = 5
user.last_seen = datetime.now(timezone.utc)
create_user(user)
return user
@@ -154,16 +141,12 @@ async def org_admin_credential(test_db: DB, org_admin_user: User) -> Credential:
"""Create a credential for the org admin user."""
import os
credential = Credential(
uuid=uuid7.create(),
credential = Credential.create(
credential_id=os.urandom(32),
user_uuid=org_admin_user.uuid,
user=org_admin_user.uuid,
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=os.urandom(64),
sign_count=0,
created_at=datetime.now(timezone.utc),
last_used=datetime.now(timezone.utc),
last_verified=None,
)
create_credential(credential)
return credential
@@ -190,11 +173,7 @@ async def org_admin_session_token(
@pytest_asyncio.fixture(scope="function")
async def grantable_permission(test_db: DB, test_org: Org) -> Permission:
"""Create a permission and add it to org's grantable permissions."""
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:grantable:perm", display_name="Grantable Perm"
)
perm = Permission.create(scope="test:grantable:perm", display_name="Grantable Perm")
create_permission(perm)
# Add to org's grantable permissions
add_permission_to_organization(str(test_org.uuid), perm.scope)
@@ -418,21 +397,16 @@ class TestAdminOrganizations:
client: httpx.AsyncClient,
session_token: str,
test_db: DB,
):
):
"""Admin should be able to delete another organization."""
import uuid7
# Create org to delete
org_to_delete = Org(
uuid=uuid7.create(),
org_to_delete = Org.create(
display_name="Org To Delete",
permissions=[],
)
create_organization(org_to_delete)
# Create some org-specific permissions to test cleanup
org_perm = Permission(
uuid=uuid7.create(),
org_perm = Permission.create(
scope=f"test:org:{org_to_delete.uuid}:feature",
display_name="Org Feature",
)
@@ -608,10 +582,7 @@ class TestAdminRoles:
):
"""Creating role with non-grantable permission should fail."""
# Create permission but don't add to org
import uuid7
perm = Permission(
uuid=uuid7.create(),
perm = Permission.create(
scope="test:not:grantable",
display_name="Not Grantable",
)
@@ -683,10 +654,7 @@ class TestAdminRoles:
test_db: DB,
):
"""Adding non-grantable permission to role should fail."""
import uuid7
perm = Permission(
uuid=uuid7.create(),
perm = Permission.create(
scope="test:not:grantable:update",
display_name="Not Grantable",
)
@@ -1110,12 +1078,9 @@ class TestAdminUsersInOrg:
):
"""Creating link for user without credentials should return registration link."""
# Create user without credentials
user_no_cred = User(
uuid=uuid7.create(),
user_no_cred = User.create(
display_name="User Without Creds",
role_uuid=user_role.uuid,
created_at=datetime.now(timezone.utc),
visits=0,
role=user_role.uuid,
)
create_user(user_no_cred)
@@ -1391,11 +1356,7 @@ class TestAdminPermissions:
):
"""Admin should be able to update a permission."""
# Create permission first
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:updateable", display_name="Updateable"
)
perm = Permission.create(scope="test:updateable", display_name="Updateable")
create_permission(perm)
response = await client.patch(
@@ -1412,11 +1373,7 @@ class TestAdminPermissions:
):
"""Updating permission with empty name should fail."""
# Create permission first
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:perm", display_name="Test Perm"
)
perm = Permission.create(scope="test:perm", display_name="Test Perm")
create_permission(perm)
response = await client.patch(
@@ -1433,11 +1390,7 @@ class TestAdminPermissions:
):
"""Admin should be able to rename a permission."""
# Create permission first
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:renameable2", display_name="Renameable"
)
perm = Permission.create(scope="test:renameable2", display_name="Renameable")
create_permission(perm)
response = await client.post(
@@ -1480,11 +1433,7 @@ class TestAdminPermissions:
self, client: httpx.AsyncClient, session_token: str, test_db: DB
):
"""Renaming permission can also update display name."""
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:rename:withname", display_name="Old Name"
)
perm = Permission.create(scope="test:rename:withname", display_name="Old Name")
create_permission(perm)
response = await client.post(
@@ -1504,11 +1453,7 @@ class TestAdminPermissions:
):
"""Admin should be able to delete a permission."""
# Create permission first
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:deleteable", display_name="Deleteable"
)
perm = Permission.create(scope="test:deleteable", display_name="Deleteable")
create_permission(perm)
response = await client.delete(
@@ -1537,14 +1482,10 @@ class TestAdminPermissions:
self, client: httpx.AsyncClient, session_token: str, test_db: DB
):
"""Can delete an auth:admin permission if another accessible one exists."""
import uuid7
from paskia.db import Permission
# Create a second auth:admin permission (no domain restriction)
perm2 = Permission(
uuid=uuid7.create(), scope="auth:admin", display_name="Secondary Admin"
)
perm2 = Permission.create(scope="auth:admin", display_name="Secondary Admin")
create_permission(perm2)
# Now we can delete the original one
@@ -1561,13 +1502,10 @@ class TestAdminPermissions:
self, client: httpx.AsyncClient, session_token: str, test_db: DB
):
"""Cannot delete auth:admin if remaining one has mismatched domain."""
import uuid7
from paskia.db import Permission
# Create a second auth:admin permission with a different domain
perm2 = Permission(
uuid=uuid7.create(),
perm2 = Permission.create(
scope="auth:admin",
display_name="Other Domain Admin",
domain="other.example.com",
tests/test_api.py
+30 -30
View File
@@ -76,7 +76,9 @@ class TestValidateEndpoint:
assert response.status_code == 200
data = response.json()
assert data["valid"] is True
assert "user_uuid" in data
assert "ctx" in data
assert "user" in data["ctx"]
assert "uuid" in data["ctx"]["user"]
@pytest.mark.asyncio
async def test_validate_with_permission_check(
@@ -243,9 +245,9 @@ class TestUserInfoEndpoint:
)
assert response.status_code == 200
data = response.json()
assert "user" in data
assert data["user"]["user_uuid"] == str(test_user.uuid)
assert data["user"]["user_name"] == test_user.display_name
assert "ctx" in data
assert data["ctx"]["user"]["uuid"] == str(test_user.uuid)
assert data["ctx"]["user"]["display_name"] == test_user.display_name
@pytest.mark.asyncio
async def test_user_info_includes_credentials(
@@ -286,7 +288,8 @@ class TestUserInfoEndpoint:
)
assert response.status_code == 200
data = response.json()
assert "permissions" in data
assert "ctx" in data
assert "permissions" in data["ctx"]
class TestSetSessionEndpoint:
@@ -392,47 +395,44 @@ class TestForwardAuthHtmlResponse:
assert data["auth"]["mode"] == "login"
class TestUserInfoWithResetToken:
"""Tests for user-info endpoint with reset tokens"""
class TestTokenInfoEndpoint:
"""Tests for token-info endpoint with reset tokens"""
@pytest.mark.asyncio
async def test_user_info_with_invalid_reset_token(self, client: httpx.AsyncClient):
"""User info with invalid reset token format should return 401."""
# Invalid format - not a well-formed passphrase (wrong separator)
response = await client.post(
"/auth/api/user-info?reset=invalid-token-format",
async def test_token_info_with_invalid_token(self, client: httpx.AsyncClient):
"""Token info with invalid token format should return 400."""
response = await client.get(
"/auth/api/token-info",
headers={"Authorization": "Bearer invalid-token-format"},
)
# Invalid format raises ValueError which gets converted to 401 HTTPException
assert response.status_code == 401
data = response.json()
assert "Invalid reset token" in data["detail"]
assert response.status_code == 400
@pytest.mark.asyncio
async def test_user_info_with_nonexistent_reset_token(
self, client: httpx.AsyncClient
):
"""User info with well-formed but non-existent reset token should return 401."""
# We need a well-formed passphrase that doesn't exist in DB
async def test_token_info_with_nonexistent_token(self, client: httpx.AsyncClient):
"""Token info with well-formed but non-existent token should return 401."""
from paskia.util.passphrase import generate
fake_token = generate() # Generates a well-formed token
response = await client.post(
f"/auth/api/user-info?reset={fake_token}",
fake_token = generate()
response = await client.get(
"/auth/api/token-info",
headers={"Authorization": f"Bearer {fake_token}"},
)
# Should return 401 for non-existent token
assert response.status_code == 401
@pytest.mark.asyncio
async def test_user_info_with_valid_reset_token(
async def test_token_info_with_valid_token(
self, client: httpx.AsyncClient, reset_token: str, test_user
):
"""User info with valid reset token should return minimal user info."""
response = await client.post(
f"/auth/api/user-info?reset={reset_token}",
"""Token info with valid reset token should return token type and display name."""
response = await client.get(
"/auth/api/token-info",
headers={"Authorization": f"Bearer {reset_token}"},
)
assert response.status_code == 200
data = response.json()
assert "user" in data
assert "token_type" in data
assert "display_name" in data
assert data["display_name"] == test_user.display_name
class TestSetSessionErrors: