LeoVasanko/passkey-auth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: LeoVasanko:e0717f005a92db97303dc01116ff8fd5a37ba15a
Branches Tags
LeoVasanko:main
..
compare: LeoVasanko:d2a6bfd2a5486ae9d31b5fb820c452caa9092c43
Branches Tags
LeoVasanko:main

No commits in common. "e0717f005a92db97303dc01116ff8fd5a37ba15a" and "d2a6bfd2a5486ae9d31b5fb820c452caa9092c43" have entirely different histories.
e0717f005a ... d2a6bfd2a5

9 changed files with 63 additions and 387 deletions
Show Stats Expand all files Collapse all files

13
frontend/admin/index.html
View File

@ -1,13 +0,0 @@
<!DOCTYPE html>
<html lang="">
<head>
<meta charset="UTF-8" />
<link rel="icon" href="/src/assets/icon.webp" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Passkey Admin</title>
</head>
<body>
<div id="admin-app"></div>
<script type="module" src="/src/admin/main.js"></script>
</body>
</html>

70
frontend/src/admin/AdminApp.vue
View File

@ -1,70 +0,0 @@
<script setup>
import { ref, onMounted } from 'vue'
const info = ref(null)
const loading = ref(true)
const error = ref(null)
async function load() {
loading.value = true
error.value = null
try {
const res = await fetch('/auth/user-info', { method: 'POST' })
const data = await res.json()
if (data.detail) throw new Error(data.detail)
info.value = data
} catch (e) {
error.value = e.message
} finally {
loading.value = false
}
}
onMounted(load)
</script>
<template>
<div class="container">
<h1>Passkey Admin</h1>
<p class="subtitle">Manage organizations, roles, and permissions</p>
<div v-if="loading">Loading…</div>
<div v-else-if="error" class="error">{{ error }}</div>
<div v-else>
<div v-if="!info?.authenticated">
<p>You must be authenticated.</p>
</div>
<div v-else-if="!(info?.is_global_admin || info?.is_org_admin)">
<p>Insufficient permissions.</p>
</div>
<div v-else>
<div class="card">
<h2>User</h2>
<div>{{ info.user.user_name }} ({{ info.user.user_uuid }})</div>
<div>Role: {{ info.role?.display_name }}</div>
</div>
<div class="card">
<h2>Organization</h2>
<div>{{ info.org?.display_name }} ({{ info.org?.uuid }})</div>
<div>Role permissions: {{ info.role?.permissions?.join(', ') }}</div>
<div>Org grantable: {{ info.org?.permissions?.join(', ') }}</div>
</div>
<div class="card">
<h2>Permissions</h2>
<div>Effective: {{ info.permissions?.join(', ') }}</div>
<div>Global admin: {{ info.is_global_admin ? 'yes' : 'no' }}</div>
<div>Org admin: {{ info.is_org_admin ? 'yes' : 'no' }}</div>
</div>
</div>
</div>
</div>
</template>
<style scoped>
.container { max-width: 960px; margin: 2rem auto; padding: 0 1rem; }
.subtitle { color: #888 }
.card { margin: 1rem 0; padding: 1rem; border: 1px solid #eee; border-radius: 8px; }
.error { color: #a00 }
</style>

6
frontend/src/admin/main.js
View File

@ -1,6 +0,0 @@
import '../assets/style.css'
import { createApp } from 'vue'
import AdminApp from './AdminApp.vue'
createApp(AdminApp).mount('#admin-app')

13
frontend/vite.config.js
View File

@ -1,7 +1,6 @@
import { fileURLToPath, URL } from 'node:url'
import { defineConfig } from 'vite'
import { resolve } from 'node:path'
import vue from '@vitejs/plugin-vue'
// https://vite.dev/config/
@ -14,7 +13,6 @@ export default defineConfig(({ command, mode }) => ({
'@': fileURLToPath(new URL('./src', import.meta.url))
},
},
// Use absolute paths at dev, deploy under /auth/
base: command === 'build' ? '/auth/' : '/',
server: {
port: 4403,
@ -29,15 +27,6 @@ export default defineConfig(({ command, mode }) => ({
build: {
outDir: '../passkey/frontend-build',
emptyOutDir: true,
assetsDir: 'assets',
rollupOptions: {
input: {
index: resolve(__dirname, 'index.html'),
admin: resolve(__dirname, 'admin/index.html')
},
output: {
// Ensure HTML files land as /auth/index.html and /auth/admin.html -> we will serve /auth/admin mapping in backend
}
}
assetsDir: 'assets'
}
}))

11
passkey/bootstrap.py
View File

@ -66,12 +66,7 @@ async def bootstrap_system(
)
await globals.db.instance.create_permission(perm1)
# Allow this org to grant admin permissions
await globals.db.instance.add_permission_to_organization(str(org.uuid), perm0.id)
await globals.db.instance.add_permission_to_organization(str(org.uuid), perm1.id)
# Create an Administration role granting both org and global admin
role = Role(uuid7.create(), org.uuid, "Administration", permissions=[perm0.id, perm1.id])
role = Role(uuid7.create(), org.uuid, "Administration")
await globals.db.instance.create_role(role)
user = User(
@ -115,9 +110,9 @@ async def check_admin_credentials() -> bool:
# Get users from the first organization with admin permission
org_users = await globals.db.instance.get_organization_users(
str(permission_orgs[0].uuid)
permission_orgs[0].id
)
admin_users = [user for user, role in org_users if role == "Administration"]
admin_users = [user for user, role in org_users if role == "Admin"]
if not admin_users:
return False

31
passkey/db/__init__.py
View File

@ -6,7 +6,7 @@ users, credentials, and sessions in a WebAuthn authentication system.
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from dataclasses import dataclass
from datetime import datetime
from uuid import UUID
@ -22,18 +22,15 @@ class Role:
uuid: UUID
org_uuid: UUID
display_name: str
# List of permission IDs this role grants to its members
permissions: list[str] = field(default_factory=list) # permission IDs
permissions: list[Permission]
@dataclass
class Org:
uuid: UUID
display_name: str
# All permission IDs that the Org is allowed to grant to its roles
permissions: list[str] = field(default_factory=list) # permission IDs
# Roles belonging to this org
roles: list[Role] = field(default_factory=list)
permissions: list[Permission] # All that the Org can grant
roles: list[Role]
@dataclass
@ -163,7 +160,7 @@ class DatabaseInterface(ABC):
@abstractmethod
async def get_organization(self, org_id: str) -> Org:
"""Get organization by ID, including its permission IDs and roles (with their permission IDs)."""
"""Get organization by ID."""
@abstractmethod
async def update_organization(self, org: Org) -> None:
@ -242,23 +239,6 @@ class DatabaseInterface(ABC):
async def get_permission_organizations(self, permission_id: str) -> list[Org]:
"""Get all organizations that have a specific permission."""
# Role-permission operations
@abstractmethod
async def add_permission_to_role(self, role_uuid: UUID, permission_id: str) -> None:
"""Add a permission to a role."""
@abstractmethod
async def remove_permission_from_role(self, role_uuid: UUID, permission_id: str) -> None:
"""Remove a permission from a role."""
@abstractmethod
async def get_role_permissions(self, role_uuid: UUID) -> list[Permission]:
"""List all permissions granted to a role."""
@abstractmethod
async def get_permission_roles(self, permission_id: str) -> list[Role]:
"""List all roles that grant a permission."""
# Combined operations
@abstractmethod
async def login(self, user_uuid: UUID, credential: Credential) -> None:
@ -281,7 +261,6 @@ __all__ = [
"Session",
"SessionContext",
"Org",
"Role",
"Permission",
"DatabaseInterface",
]

254
passkey/db/sql.py
View File

@ -54,7 +54,6 @@ class OrgModel(Base):
display_name: Mapped[str] = mapped_column(String, nullable=False)
def as_dataclass(self):
# Base Org without permissions/roles (filled by data accessors)
return Org(UUID(bytes=self.uuid), self.display_name)
@staticmethod
@ -72,7 +71,6 @@ class RoleModel(Base):
display_name: Mapped[str] = mapped_column(String, nullable=False)
def as_dataclass(self):
# Base Role without permissions (filled by data accessors)
return Role(
uuid=UUID(bytes=self.uuid),
org_uuid=UUID(bytes=self.org_uuid),
@ -260,17 +258,7 @@ class DB(DatabaseInterface):
async def create_role(self, role: Role) -> None:
async with self.session() as session:
# Create role record
session.add(RoleModel.from_dataclass(role))
# Persist role permissions
if role.permissions:
for perm_id in role.permissions:
session.add(
RolePermission(
role_uuid=role.uuid.bytes,
permission_id=perm_id,
)
)
async def create_credential(self, credential: Credential) -> None:
async with self.session() as session:
@ -441,12 +429,6 @@ class DB(DatabaseInterface):
display_name=org.display_name,
)
session.add(org_model)
# Persist org permissions the org is allowed to grant
if org.permissions:
for perm_id in org.permissions:
session.add(
OrgPermission(org_uuid=org.uuid.bytes, permission_id=perm_id)
)
async def get_organization(self, org_id: str) -> Org:
async with self.session() as session:
@ -459,34 +441,7 @@ class DB(DatabaseInterface):
if not org_model:
raise ValueError("Organization not found")
# Build Org with permissions and roles
org_dc = org_model.as_dataclass()
# Load org permission IDs
perm_stmt = select(OrgPermission.permission_id).where(
OrgPermission.org_uuid == org_uuid.bytes
)
perm_result = await session.execute(perm_stmt)
org_dc.permissions = [row[0] for row in perm_result.fetchall()]
# Load roles for org
roles_stmt = select(RoleModel).where(RoleModel.org_uuid == org_uuid.bytes)
roles_result = await session.execute(roles_stmt)
roles_models = roles_result.scalars().all()
roles: list[Role] = []
if roles_models:
# For each role, load permission IDs
for r_model in roles_models:
r_dc = r_model.as_dataclass()
r_perm_stmt = select(RolePermission.permission_id).where(
RolePermission.role_uuid == r_model.uuid
)
r_perm_result = await session.execute(r_perm_stmt)
r_dc.permissions = [row[0] for row in r_perm_result.fetchall()]
roles.append(r_dc)
org_dc.roles = roles
return org_dc
return org_model.as_dataclass()
async def update_organization(self, org: Org) -> None:
async with self.session() as session:
@ -496,19 +451,6 @@ class DB(DatabaseInterface):
.values(display_name=org.display_name)
)
await session.execute(stmt)
# Synchronize org permissions join table to match org.permissions
# Delete existing rows for this org
await session.execute(
delete(OrgPermission).where(OrgPermission.org_uuid == org.uuid.bytes)
)
# Insert new rows
if org.permissions:
for perm_id in org.permissions:
await session.merge(
OrgPermission(
org_uuid=org.uuid.bytes, permission_id=perm_id
)
)
async def delete_organization(self, org_uuid: UUID) -> None:
async with self.session() as session:
@ -517,10 +459,9 @@ class DB(DatabaseInterface):
await session.execute(stmt)
async def add_user_to_organization(
self, user_uuid: UUID, org_id: str, role: str
self, user_uuid: UUID, org_uuid: UUID, role: str
) -> None:
async with self.session() as session:
org_uuid = UUID(org_id)
# Get user and organization models
user_stmt = select(UserModel).where(UserModel.uuid == user_uuid.bytes)
user_result = await session.execute(user_stmt)
@ -536,32 +477,40 @@ class DB(DatabaseInterface):
if not org_model:
raise ValueError("Organization not found")
# Find the role within this organization by display_name
role_stmt = select(RoleModel).where(
RoleModel.org_uuid == org_uuid.bytes,
RoleModel.display_name == role,
)
role_result = await session.execute(role_stmt)
role_model = role_result.scalar_one_or_none()
if not role_model:
raise ValueError("Role not found in organization")
# Update the user's role assignment
# Update the user's organization and role
stmt = (
update(UserModel)
.where(UserModel.uuid == user_uuid.bytes)
.values(role_uuid=role_model.uuid)
.values(org_uuid=org_uuid.bytes, role=role)
)
await session.execute(stmt)
async def transfer_user_to_organization(
self, user_uuid: UUID, new_org_id: str, new_role: str | None = None
) -> None:
# Users are members of an org that never changes after creation.
# Disallow transfers across organizations to enforce invariant.
raise ValueError("Users cannot be transferred to a different organization")
async with self.session() as session:
# Convert string ID to UUID bytes for lookup
new_org_uuid = UUID(new_org_id)
async def get_user_organization(self, user_uuid: UUID) -> tuple[Org, str]:
# Verify the new organization exists
org_stmt = select(OrgModel).where(OrgModel.uuid == new_org_uuid.bytes)
org_result = await session.execute(org_stmt)
org_model = org_result.scalar_one_or_none()
if not org_model:
raise ValueError("Target organization not found")
# Update the user's organization and role
stmt = (
update(UserModel)
.where(UserModel.uuid == user_uuid.bytes)
.values(org_uuid=new_org_uuid.bytes, role=new_role)
)
result = await session.execute(stmt)
if result.rowcount == 0:
raise ValueError("User not found")
async def get_user_organization(self, user_uuid: UUID) -> Org:
async with self.session() as session:
stmt = select(UserModel).where(UserModel.uuid == user_uuid.bytes)
result = await session.execute(stmt)
@ -570,31 +519,20 @@ class DB(DatabaseInterface):
if not user_model:
raise ValueError("User not found")
# Find user's role to get org
role_stmt = select(RoleModel).where(RoleModel.uuid == user_model.role_uuid)
role_result = await session.execute(role_stmt)
role_model = role_result.scalar_one()
# Fetch the organization details
org_stmt = select(OrgModel).where(OrgModel.uuid == role_model.org_uuid)
org_stmt = select(OrgModel).where(OrgModel.uuid == user_model.role_uuid)
org_result = await session.execute(org_stmt)
org_model = org_result.scalar_one()
# Convert UUID bytes back to string for the interface
return org_model.as_dataclass(), role_model.display_name
return org_model.as_dataclass()
async def get_organization_users(self, org_id: str) -> list[tuple[User, str]]:
async def get_organization_users(self, org_id: str) -> list[User]:
async with self.session() as session:
org_uuid = UUID(org_id)
# Join users with roles to filter by org and return role names
stmt = (
select(UserModel, RoleModel.display_name)
.join(RoleModel, UserModel.role_uuid == RoleModel.uuid)
.where(RoleModel.org_uuid == org_uuid.bytes)
)
stmt = select(UserModel).where(UserModel.role_uuid == role.uuid.bytes)
result = await session.execute(stmt)
rows = result.fetchall()
return [(u.as_dataclass(), role_name) for (u, role_name) in rows]
user_models = result.scalars().all()
return [u.as_dataclass() for u in user_models]
async def get_user_role_in_organization(
self, user_uuid: UUID, org_id: str
@ -603,14 +541,9 @@ class DB(DatabaseInterface):
async with self.session() as session:
# Convert string ID to UUID bytes for lookup
org_uuid = UUID(org_id)
stmt = (
select(RoleModel.display_name)
.select_from(UserModel)
.join(RoleModel, UserModel.role_uuid == RoleModel.uuid)
.where(
stmt = select(UserModel.role).where(
UserModel.uuid == user_uuid.bytes,
RoleModel.org_uuid == org_uuid.bytes,
)
UserModel.org_uuid == org_uuid.bytes,
)
result = await session.execute(stmt)
return result.scalar_one_or_none()
@ -620,35 +553,14 @@ class DB(DatabaseInterface):
) -> None:
"""Update a user's role in their organization."""
async with self.session() as session:
# Find user's current org via their role
user_stmt = select(UserModel).where(UserModel.uuid == user_uuid.bytes)
user_result = await session.execute(user_stmt)
user_model = user_result.scalar_one_or_none()
if not user_model:
raise ValueError("User not found")
current_role_stmt = select(RoleModel).where(
RoleModel.uuid == user_model.role_uuid
)
current_role_result = await session.execute(current_role_stmt)
current_role = current_role_result.scalar_one()
# Find the new role within the same organization
role_stmt = select(RoleModel).where(
RoleModel.org_uuid == current_role.org_uuid,
RoleModel.display_name == new_role,
)
role_result = await session.execute(role_stmt)
role_model = role_result.scalar_one_or_none()
if not role_model:
raise ValueError("Role not found in user's organization")
stmt = (
update(UserModel)
.where(UserModel.uuid == user_uuid.bytes)
.values(role_uuid=role_model.uuid)
.values(role=new_role)
)
await session.execute(stmt)
result = await session.execute(stmt)
if result.rowcount == 0:
raise ValueError("User not found")
# Permission operations
async def create_permission(self, permission: Permission) -> None:
@ -686,53 +598,6 @@ class DB(DatabaseInterface):
stmt = delete(PermissionModel).where(PermissionModel.id == permission_id)
await session.execute(stmt)
async def add_permission_to_role(self, role_uuid: UUID, permission_id: str) -> None:
async with self.session() as session:
# Ensure role exists
role_stmt = select(RoleModel).where(RoleModel.uuid == role_uuid.bytes)
role_result = await session.execute(role_stmt)
role_model = role_result.scalar_one_or_none()
if not role_model:
raise ValueError("Role not found")
# Ensure permission exists
perm_stmt = select(PermissionModel).where(PermissionModel.id == permission_id)
perm_result = await session.execute(perm_stmt)
if not perm_result.scalar_one_or_none():
raise ValueError("Permission not found")
session.add(
RolePermission(role_uuid=role_uuid.bytes, permission_id=permission_id)
)
async def remove_permission_from_role(self, role_uuid: UUID, permission_id: str) -> None:
async with self.session() as session:
await session.execute(
delete(RolePermission)
.where(RolePermission.role_uuid == role_uuid.bytes)
.where(RolePermission.permission_id == permission_id)
)
async def get_role_permissions(self, role_uuid: UUID) -> list[Permission]:
async with self.session() as session:
stmt = (
select(PermissionModel)
.join(RolePermission, PermissionModel.id == RolePermission.permission_id)
.where(RolePermission.role_uuid == role_uuid.bytes)
)
result = await session.execute(stmt)
return [p.as_dataclass() for p in result.scalars().all()]
async def get_permission_roles(self, permission_id: str) -> list[Role]:
async with self.session() as session:
stmt = (
select(RoleModel)
.join(RolePermission, RoleModel.uuid == RolePermission.role_uuid)
.where(RolePermission.permission_id == permission_id)
)
result = await session.execute(stmt)
return [r.as_dataclass() for r in result.scalars().all()]
async def add_permission_to_organization(
self, org_id: str, permission_id: str
) -> None:
@ -814,7 +679,9 @@ class DB(DatabaseInterface):
)
org_result = await session.execute(org_stmt)
org_model = org_result.scalar_one()
organizations.append(org_model.as_dataclass())
# Convert UUID bytes back to string for the interface
organizations.append(org.as_dataclass())
return organizations
@ -830,21 +697,21 @@ class DB(DatabaseInterface):
Uses efficient JOINs to retrieve all related data in a single database query.
"""
async with self.session() as session:
# Build a query that joins sessions, users, roles, organizations, and role_permissions
# Build a query that joins sessions, users, organizations, org_permissions, and permissions
stmt = (
select(
SessionModel,
UserModel,
RoleModel,
OrgModel,
PermissionModel,
)
.select_from(SessionModel)
.join(UserModel, SessionModel.user_uuid == UserModel.uuid)
.join(RoleModel, UserModel.role_uuid == RoleModel.uuid)
.join(OrgModel, RoleModel.org_uuid == OrgModel.uuid)
.outerjoin(RolePermission, RoleModel.uuid == RolePermission.role_uuid)
.outerjoin(PermissionModel, RolePermission.permission_id == PermissionModel.id)
.join(OrgModel, UserModel.org_uuid == OrgModel.uuid)
.outerjoin(OrgPermission, OrgModel.uuid == OrgPermission.org_uuid)
.outerjoin(
PermissionModel, OrgPermission.permission_id == PermissionModel.id
)
.where(SessionModel.key == session_key)
)
@ -856,7 +723,7 @@ class DB(DatabaseInterface):
# Extract the first row to get session and user data
first_row = rows[0]
session_model, user_model, role_model, org_model, _ = first_row
session_model, user_model, org_model, _ = first_row
# Create the session object
session_obj = Session(
@ -872,21 +739,14 @@ class DB(DatabaseInterface):
# Create the user object
user_obj = user_model.as_dataclass()
# Create organization object (fill permissions later if needed)
# Create organization object (always exists now)
organization = Org(UUID(bytes=org_model.uuid), org_model.display_name)
# Create role object
role = Role(
uuid=UUID(bytes=role_model.uuid),
org_uuid=UUID(bytes=role_model.org_uuid),
display_name=role_model.display_name,
)
# Collect all unique permissions for the role
# Collect all unique permissions
permissions = []
seen_permission_ids = set()
for row in rows:
_, _, _, _, permission_model = row
_, _, _, permission_model = row
if permission_model and permission_model.id not in seen_permission_ids:
permissions.append(
Permission(
@ -896,20 +756,10 @@ class DB(DatabaseInterface):
)
seen_permission_ids.add(permission_model.id)
# Attach permission IDs to role
role.permissions = list(seen_permission_ids)
# Load org permission IDs as well
org_perm_stmt = select(OrgPermission.permission_id).where(
OrgPermission.org_uuid == org_model.uuid
)
org_perm_result = await session.execute(org_perm_stmt)
organization.permissions = [row[0] for row in org_perm_result.fetchall()]
return SessionContext(
session=session_obj,
user=user_obj,
org=organization,
role=role,
role=user_model.role,
permissions=permissions if permissions else None,
)

35
passkey/fastapi/api.py
View File

@ -41,9 +41,6 @@ def register_api_routes(app: FastAPI):
"""Get full user information for the authenticated user."""
reset = passphrase.is_well_formed(auth)
s = await (get_reset if reset else get_session)(auth)
# Session context (org, role, permissions)
ctx = await db.instance.get_session_context(session_key(auth))
# Fallback if context not available (e.g., reset session)
u = await db.instance.get_user_by_uuid(s.user_uuid)
# Get all credentials for the user
credential_ids = await db.instance.get_credentials_by_user_uuid(s.user_uuid)
@ -81,33 +78,6 @@ def register_api_routes(app: FastAPI):
# Sort credentials by creation date (earliest first, most recently created last)
credentials.sort(key=lambda cred: cred["created_at"])
# Permissions and roles
role_info = None
org_info = None
effective_permissions = []
is_global_admin = False
is_org_admin = False
if ctx:
role_info = {
"uuid": str(ctx.role.uuid),
"display_name": ctx.role.display_name,
"permissions": ctx.role.permissions, # IDs
}
org_info = {
"uuid": str(ctx.org.uuid),
"display_name": ctx.org.display_name,
"permissions": ctx.org.permissions, # IDs the org can grant
}
# Effective permissions are role permissions; API also returns full objects for convenience
effective_permissions = [p.id for p in (ctx.permissions or [])]
is_global_admin = "auth/admin" in role_info["permissions"]
# org admin permission is auth/org:<org_uuid>
is_org_admin = (
f"auth/org:{org_info['uuid']}" in role_info["permissions"]
if org_info
else False
)
return {
"authenticated": not reset,
"session_type": s.info["type"],
@ -118,11 +88,6 @@ def register_api_routes(app: FastAPI):
"last_seen": u.last_seen.isoformat() if u.last_seen else None,
"visits": u.visits,
},
"org": org_info,
"role": role_info,
"permissions": effective_permissions,
"is_global_admin": is_global_admin,
"is_org_admin": is_org_admin,
"credentials": credentials,
"aaguid_info": aaguid_info,
}

13
passkey/fastapi/mainapp.py
View File

@ -70,19 +70,6 @@ async def redirect_to_index():
return FileResponse(STATIC_DIR / "index.html")
@app.get("/auth/admin")
async def serve_admin():
"""Serve the admin app entry point."""
# Vite MPA builds admin as admin.html in the same outDir
admin_html = STATIC_DIR / "admin.html"
# If configured to emit admin/index.html, support that too
if not admin_html.exists():
alt = STATIC_DIR / "admin" / "index.html"
if alt.exists():
return FileResponse(alt)
return FileResponse(admin_html)
# Register API routes
register_api_routes(app)
register_reset_routes(app)