Permissions refactor. Permissions have UUID and scope (previously id) and the latter no longer needs to be unique. Org admin uses a single global permission now. Domain scoped permissions. Removed from user info the admin fields, use effective_permission checks instead.

This commit is contained in:
2026-01-23 13:54:31 +00:00
parent 236d52aa55
commit 3430c7f0cf
16 changed files with 729 additions and 318 deletions
+31 -22
View File
@@ -46,6 +46,10 @@ const adminUserDetailRef = ref(null)
// Check if any modal/dialog is open (blocks arrow key navigation)
const hasActiveModal = computed(() => dialog.value.type !== null || showRegModal.value)
// Derive admin status from permissions
const isGlobalAdmin = computed(() => info.value?.permissions?.includes('auth:admin') ?? false)
const isOrgAdmin = computed(() => info.value?.permissions?.includes('auth:org:admin') ?? false)
function sanitizeRenameId() { if (renameIdValue.value) renameIdValue.value = renameIdValue.value.replace(safeIdRegex, '') }
function handleGlobalClick(e) {
@@ -108,7 +112,7 @@ const permissionSummary = computed(() => {
return display
})
function renamePermissionDisplay(p) { openDialog('perm-display', { permission: p, id: p.id, display_name: p.display_name }) }
function renamePermissionDisplay(p) { openDialog('perm-display', { permission: p, scope: p.scope, display_name: p.display_name, domain: p.domain || '' }) }
function parseHash() {
@@ -153,7 +157,7 @@ async function load() {
// If we get here, user has admin access - now fetch user info for display
await loadUserInfo()
if (!info.value.is_global_admin && info.value.is_org_admin && orgs.value.length === 1) {
if (!isGlobalAdmin.value && isOrgAdmin.value && orgs.value.length === 1) {
if (!window.location.hash || window.location.hash === '#overview') {
currentOrgId.value = orgs.value[0].uuid
window.location.hash = `#org/${currentOrgId.value}`
@@ -186,7 +190,7 @@ async function performOrgDeletion(orgUuid) {
}
function deleteOrg(org) {
if (!info.value?.is_global_admin) { authStore.showMessage('Global admin only'); return }
if (!isGlobalAdmin.value) { authStore.showMessage('Global admin only'); return }
const userCount = org.roles.reduce((acc, r) => acc + r.users.length, 0)
@@ -289,20 +293,20 @@ async function toggleRolePermission(role, pid, checked) {
}
// Permission actions
async function performPermissionDeletion(permissionId) {
const params = new URLSearchParams({ permission_id: permissionId })
async function performPermissionDeletion(permissionScope) {
const params = new URLSearchParams({ permission_id: permissionScope })
await apiJson(`/auth/api/admin/permission?${params.toString()}`, { method: 'DELETE' })
await loadPermissions()
}
function deletePermission(p) {
const userCount = permissionSummary.value[p.id]?.userCount || 0
const userCount = permissionSummary.value[p.scope]?.userCount || 0
// Count roles that have this permission
let roleCount = 0
for (const org of orgs.value) {
for (const role of org.roles) {
if (role.permissions.includes(p.id)) {
if (role.permissions.includes(p.scope)) {
roleCount++
}
}
@@ -310,7 +314,7 @@ function deletePermission(p) {
if (roleCount === 0) {
// No roles have this permission, safe to delete directly
performPermissionDeletion(p.id)
performPermissionDeletion(p.scope)
.then(() => {
authStore.showMessage(`Permission "${p.display_name}" deleted.`, 'success', 2500)
})
@@ -326,7 +330,7 @@ function deletePermission(p) {
const affects = parts.join(', ')
openDialog('confirm', { message: `Delete permission "${p.display_name}" (${affects})?`, action: async () => {
await performPermissionDeletion(p.id)
await performPermissionDeletion(p.scope)
} })
}
@@ -626,21 +630,24 @@ async function submitDialog() {
return // Don't call closeDialog() again
} else if (t === 'perm-display') {
const { permission } = dialog.value.data
const newId = dialog.value.data.id?.trim()
const newId = dialog.value.data.scope?.trim()
const newDisplay = dialog.value.data.display_name?.trim()
const newDomain = dialog.value.data.domain?.trim() || ''
if (!newDisplay) throw new Error('Display name required')
if (!newId) throw new Error('ID required')
if (!newId) throw new Error('Scope required')
// Close dialog immediately, then perform async operation
closeDialog()
const oldDomain = permission.domain || ''
let apiCall;
if (newId !== permission.id) {
// ID changed, use rename endpoint
apiCall = apiJson('/auth/api/admin/permission/rename', { method: 'POST', body: { old_id: permission.id, new_id: newId, display_name: newDisplay } })
} else if (newDisplay !== permission.display_name) {
// Only display name changed
const params = new URLSearchParams({ permission_id: permission.id, display_name: newDisplay })
if (newId !== permission.scope) {
// Scope changed, use rename endpoint (also update domain)
apiCall = apiJson('/auth/api/admin/permission/rename', { method: 'POST', body: { old_scope: permission.scope, new_scope: newId, display_name: newDisplay, domain: newDomain } })
} else if (newDisplay !== permission.display_name || newDomain !== oldDomain) {
// Display name or domain changed
const params = new URLSearchParams({ permission_id: permission.scope, display_name: newDisplay })
if (newDomain) params.set('domain', newDomain)
apiCall = apiJson(`/auth/api/admin/permission?${params.toString()}`, { method: 'PUT' })
} else {
// No changes
@@ -655,13 +662,15 @@ async function submitDialog() {
.catch(e => {
authStore.showMessage(e.message || 'Failed to update permission', 'error')
})
return // Don't call closeDialog() again else if (t === 'perm-create') {
const id = dialog.value.data.id?.trim(); if (!id) throw new Error('ID required')
return // Don't call closeDialog() again
} else if (t === 'perm-create') {
const scope = dialog.value.data.scope?.trim(); if (!scope) throw new Error('Scope required')
const display_name = dialog.value.data.display_name?.trim(); if (!display_name) throw new Error('Display name required')
const domain = dialog.value.data.domain?.trim() || ''
// Close dialog immediately, then perform async operation
closeDialog()
apiJson('/auth/api/admin/permissions', { method: 'POST', body: { id, display_name } })
apiJson('/auth/api/admin/permissions', { method: 'POST', body: { scope, display_name, domain: domain || undefined } })
.then(() => {
authStore.showMessage(`Permission "${display_name}" created.`, 'success', 2500)
loadPermissions()
@@ -689,7 +698,7 @@ async function submitDialog() {
v-else-if="showBackMessage"
@reload="reloadPage"
/>
<section v-else-if="authenticated && (info?.is_global_admin || info?.is_org_admin)" class="view-root view-root--wide view-admin">
<section v-else-if="authenticated && (isGlobalAdmin || isOrgAdmin)" class="view-root view-root--wide view-admin">
<header class="view-header">
<h1>{{ pageHeading }}</h1>
<Breadcrumbs ref="breadcrumbsRef" :entries="breadcrumbEntries" @keydown="handleBreadcrumbKeydown" />
@@ -700,7 +709,7 @@ async function submitDialog() {
<div v-if="error" class="surface surface--tight error">{{ error }}</div>
<div v-else class="admin-panels">
<AdminOverview
v-if="!selectedUser && !selectedOrg && (info.is_global_admin || info.is_org_admin)"
v-if="!selectedUser && !selectedOrg && (isGlobalAdmin || isOrgAdmin)"
ref="adminOverviewRef"
:info="info"
:orgs="orgs"
+8 -3
View File
@@ -72,10 +72,14 @@ const NAME_EDIT_TYPES = new Set(['org-update', 'role-update', 'user-update-name'
<label>Display Name
<input ref="displayNameInput" v-model="dialog.data.display_name" required />
</label>
<label>Permission ID
<input v-model="dialog.data.id" :placeholder="dialog.type === 'perm-create' ? 'yourapp:permission' : dialog.data.permission.id" required :pattern="PERMISSION_ID_PATTERN" title="Allowed: A-Za-z0-9:._~-" data-form-type="other" />
<label>Permission Scope
<input v-model="dialog.data.scope" :placeholder="dialog.type === 'perm-create' ? 'yourapp:permission' : dialog.data.permission.scope" required :pattern="PERMISSION_ID_PATTERN" title="Allowed: A-Za-z0-9:._~-" data-form-type="other" />
</label>
<p class="small muted">The permission ID is used for permission checks in the application. Changing it may break deployed applications that reference this permission.</p>
<label>Domain Scope <span class="optional">(optional)</span>
<input v-model="dialog.data.domain" placeholder="e.g. app.example.com" data-form-type="other" />
</label>
<p class="small muted">If set, this permission only applies when accessed from the specified domain. Must be the RP ID or a subdomain of it.</p>
<p class="small muted">The permission scope is used for permission checks in the application. Changing it may break deployed applications that reference this permission.</p>
</template>
<template v-else-if="dialog.type==='confirm'">
<p>{{ dialog.data.message }}</p>
@@ -106,4 +110,5 @@ const NAME_EDIT_TYPES = new Set(['org-update', 'role-update', 'user-update-name'
.error { color: var(--color-danger-text); }
.small { font-size: 0.9rem; }
.muted { color: var(--color-text-muted); }
.optional { font-weight: normal; color: var(--color-text-muted); font-size: 0.85em; }
</style>
+2 -2
View File
@@ -26,8 +26,8 @@ const sortedRoles = computed(() => {
})
})
function permissionDisplayName(id) {
return props.permissions.find(p => p.id === id)?.display_name || id
function permissionDisplayName(scope) {
return props.permissions.find(p => p.scope === scope)?.display_name || scope
}
function toggleRolePermission(role, pid, checked) {
+29 -22
View File
@@ -24,10 +24,14 @@ const sortedOrgs = computed(() => [...props.orgs].sort((a,b)=> {
const nameCompare = a.display_name.localeCompare(b.display_name)
return nameCompare !== 0 ? nameCompare : a.uuid.localeCompare(b.uuid)
}))
const sortedPermissions = computed(() => [...props.permissions].sort((a,b)=> a.id.localeCompare(b.id)))
const sortedPermissions = computed(() => [...props.permissions].sort((a,b)=> a.scope.localeCompare(b.scope)))
function permissionDisplayName(id) {
return props.permissions.find(p => p.id === id)?.display_name || id
// Derive admin status from permissions
const isGlobalAdmin = computed(() => props.info?.permissions?.includes('auth:admin') ?? false)
const isOrgAdmin = computed(() => props.info?.permissions?.includes('auth:org:admin') ?? false)
function permissionDisplayName(scope) {
return props.permissions.find(p => p.scope === scope)?.display_name || scope
}
function getRoleNames(org) {
@@ -89,7 +93,7 @@ function handleTableKeydown(event, tableType) {
} else if (direction === 'down' && currentIndex === rows.length - 1) {
// At bottom of org table, navigate to permissions section
event.preventDefault()
if (tableType === 'org' && props.info.is_global_admin) {
if (tableType === 'org' && isGlobalAdmin.value) {
// Navigate to permissions matrix or actions
if (permMatrixRef.value) {
const firstCheckbox = permMatrixRef.value.querySelector('input[type="checkbox"]')
@@ -232,7 +236,7 @@ function handlePermActionsKeydown(event) {
// Focus helper for external navigation
function focusFirstElement() {
if (props.info.is_global_admin) {
if (isGlobalAdmin.value) {
focusPreferred(orgActionsRef.value, { itemSelector: 'button' })
} else {
const firstFocusable = orgTableRef.value?.querySelector('tbody tr a, tbody tr button:not([disabled])')
@@ -245,9 +249,9 @@ defineExpose({ focusFirstElement })
<template>
<div class="permissions-section" ref="orgSection">
<h2>{{ info.is_global_admin ? 'Organizations' : 'Your Organizations' }}</h2>
<h2>{{ isGlobalAdmin ? 'Organizations' : 'Your Organizations' }}</h2>
<div class="actions" ref="orgActionsRef" @keydown="handleOrgActionsKeydown">
<button v-if="info.is_global_admin" @click="$emit('createOrg')">+ Create Org</button>
<button v-if="isGlobalAdmin" @click="$emit('createOrg')">+ Create Org</button>
</div>
<table class="org-table" ref="orgTableRef" @keydown="e => handleTableKeydown(e, 'org')">
<thead>
@@ -255,18 +259,18 @@ defineExpose({ focusFirstElement })
<th>Name</th>
<th>Roles</th>
<th>Members</th>
<th v-if="info.is_global_admin">Actions</th>
<th v-if="isGlobalAdmin">Actions</th>
</tr>
</thead>
<tbody>
<tr v-for="o in sortedOrgs" :key="o.uuid">
<td>
<a href="#org/{{o.uuid}}" @click.prevent="$emit('openOrg', o)">{{ o.display_name }}</a>
<button v-if="info.is_global_admin || info.is_org_admin" @click="$emit('updateOrg', o)" class="icon-btn edit-org-btn" aria-label="Rename organization" title="Rename organization"></button>
<button v-if="isGlobalAdmin || isOrgAdmin" @click="$emit('updateOrg', o)" class="icon-btn edit-org-btn" aria-label="Rename organization" title="Rename organization"></button>
</td>
<td class="role-names">{{ getRoleNames(o) }}</td>
<td class="center">{{ o.roles.reduce((acc,r)=>acc + r.users.length,0) }}</td>
<td v-if="info.is_global_admin" class="center">
<td v-if="isGlobalAdmin" class="center">
<button @click="$emit('deleteOrg', o)" class="icon-btn delete-icon" aria-label="Delete organization" title="Delete organization"></button>
</td>
</tr>
@@ -274,7 +278,7 @@ defineExpose({ focusFirstElement })
</table>
</div>
<div v-if="info.is_global_admin" class="permissions-section">
<div v-if="isGlobalAdmin" class="permissions-section">
<h2>Permissions</h2>
<div class="matrix-wrapper" ref="permMatrixRef" @keydown="handleMatrixKeydown">
<div class="matrix-scroll">
@@ -292,19 +296,19 @@ defineExpose({ focusFirstElement })
<span>{{ o.display_name }}</span>
</div>
<template v-for="p in sortedPermissions" :key="p.id">
<div class="perm-name" :title="p.id">
<template v-for="p in sortedPermissions" :key="p.scope">
<div class="perm-name" :title="p.scope">
<span class="display-text">{{ p.display_name }}</span>
</div>
<div
v-for="o in sortedOrgs"
:key="o.uuid + '-' + p.id"
:key="o.uuid + '-' + p.scope"
class="matrix-cell"
>
<input
type="checkbox"
:checked="o.permissions.includes(p.id)"
@change="e => $emit('toggleOrgPermission', o, p.id, e.target.checked)"
:checked="o.permissions.includes(p.scope)"
@change="e => $emit('toggleOrgPermission', o, p.scope, e.target.checked)"
/>
</div>
</template>
@@ -313,28 +317,30 @@ defineExpose({ focusFirstElement })
<p class="matrix-hint muted">Toggle which permissions each organization can grant to its members.</p>
</div>
<div class="actions" ref="permActionsRef" @keydown="handlePermActionsKeydown">
<button v-if="info.is_global_admin" @click="$emit('openDialog', 'perm-create', { display_name: '', id: '' })">+ Create Permission</button>
<button v-if="isGlobalAdmin" @click="$emit('openDialog', 'perm-create', { display_name: '', scope: '', domain: '' })">+ Create Permission</button>
</div>
<table class="org-table" ref="permTableRef" @keydown="e => handleTableKeydown(e, 'perm')">
<thead>
<tr>
<th scope="col">Permission</th>
<th scope="col">Domain</th>
<th scope="col" class="center">Members</th>
<th scope="col" class="center">Actions</th>
</tr>
</thead>
<tbody>
<tr v-for="p in sortedPermissions" :key="p.id">
<tr v-for="p in sortedPermissions" :key="p.scope">
<td class="perm-name-cell">
<div class="perm-title">
<span class="display-text">{{ p.display_name }}</span>
<button @click="$emit('renamePermissionDisplay', p)" class="icon-btn edit-display-btn" aria-label="Edit display name" title="Edit display name"></button>
<button @click="$emit('renamePermissionDisplay', p)" class="icon-btn edit-display-btn" aria-label="Edit permission" title="Edit permission"></button>
</div>
<div class="perm-id-info">
<span class="id-text">{{ p.id }}</span>
<span class="id-text">{{ p.scope }}</span>
</div>
</td>
<td class="perm-members center">{{ permissionSummary[p.id]?.userCount || 0 }}</td>
<td class="perm-domain">{{ p.domain || '—' }}</td>
<td class="perm-members center">{{ permissionSummary[p.scope]?.userCount || 0 }}</td>
<td class="perm-actions center">
<button @click="$emit('deletePermission', p)" class="icon-btn delete-icon" aria-label="Delete permission" title="Delete permission"></button>
</td>
@@ -355,7 +361,8 @@ defineExpose({ focusFirstElement })
.org-table .role-names { max-width: 200px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.perm-name-cell { display: flex; flex-direction: column; gap: 0.3rem; }
.perm-title { font-weight: 600; color: var(--color-heading); }
.perm-id-info { font-size: 0.8rem; color: var(--color-text-muted); }
.perm-id-info { font-size: 0.8rem; color: var(--color-text-muted); display: flex; align-items: center; gap: 0.5rem; flex-wrap: wrap; }
.perm-domain { color: var(--color-text-muted); font-size: 0.9rem; }
.icon-btn { background: none; border: none; color: var(--color-text-muted); padding: 0.2rem; border-radius: var(--radius-sm); cursor: pointer; transition: background 0.2s ease, color 0.2s ease; }
.icon-btn:hover { color: var(--color-heading); background: var(--color-surface-muted); }
.delete-icon { color: var(--color-danger); }
+4 -1
View File
@@ -324,7 +324,10 @@ const terminateSession = async (session) => {
const logoutEverywhere = async () => { await authStore.logoutEverywhere() }
const logout = async () => { await authStore.logout() }
const openNameDialog = () => { newName.value = authStore.userInfo?.user?.user_name || ''; showNameDialog.value = true }
const isAdmin = computed(() => !!(authStore.userInfo?.is_global_admin || authStore.userInfo?.is_org_admin))
const isAdmin = computed(() => {
const perms = authStore.userInfo?.permissions ?? []
return perms.includes('auth:admin') || perms.includes('auth:org:admin')
})
const hasMultipleSessions = computed(() => sessions.value.length > 1)
const breadcrumbEntries = computed(() => { const entries = [{ label: 'Auth', href: makeUiHref() }]; if (isAdmin.value) entries.push({ label: 'Admin', href: adminUiPath() }); return entries })
+14 -14
View File
@@ -61,15 +61,17 @@ async def bootstrap_system() -> dict:
dict: Contains information about created entities and reset link
"""
# Create permission first - will fail if already exists
perm0 = Permission(id="auth:admin", display_name="Master Admin")
perm0 = Permission(
uuid=uuid7.create(), scope="auth:admin", display_name="Master Admin"
)
db.create_permission(perm0)
org = Org(uuid7.create(), "Organization")
db.create_organization(org)
# After creation, org.permissions now includes the auto-created org admin permission
# After creation, org.permissions now includes the auto-created org admin permission (auth:org:admin)
# Allow this org to grant global admin explicitly
db.add_permission_to_organization(str(org.uuid), perm0.id)
db.add_permission_to_organization(str(org.uuid), perm0.scope)
# Create an Administration role granting both org and global admin
# Compose permissions for Administration role: global admin + org admin auto-perm
@@ -77,7 +79,7 @@ async def bootstrap_system() -> dict:
uuid7.create(),
org.uuid,
"Administration",
permissions=[perm0.id, *org.permissions],
permissions=[perm0.scope, *org.permissions],
)
db.create_role(role)
@@ -101,7 +103,11 @@ async def bootstrap_system() -> dict:
"role": role,
"permissions": [
perm0,
*[Permission(id=p, display_name="") for p in org.permissions],
*[
db.get_permission_by_scope(p)
for p in org.permissions
if db.get_permission_by_scope(p)
],
],
"reset_link": reset_link,
}
@@ -116,17 +122,13 @@ async def check_admin_credentials() -> bool:
"""
try:
# Get permission organizations to find admin users
permission_orgs = db.get_permission_organizations(
"auth:admin"
)
permission_orgs = db.get_permission_organizations("auth:admin")
if not permission_orgs:
return False
# Get users from the first organization with admin permission
org_users = db.get_organization_users(
str(permission_orgs[0].uuid)
)
org_users = db.get_organization_users(str(permission_orgs[0].uuid))
admin_users = [user for user, role in org_users if role == "Administration"]
if not admin_users:
@@ -134,9 +136,7 @@ async def check_admin_credentials() -> bool:
# Check first admin user for credentials
admin_user = admin_users[0]
credentials = db.get_credentials_by_user_uuid(
admin_user.uuid
)
credentials = db.get_credentials_by_user_uuid(admin_user.uuid)
if not credentials:
# Admin exists but has no credentials, create reset link
+264 -90
View File
@@ -40,11 +40,13 @@ CLEANUP_INTERVAL = 1
# -------------------------------------------------------------------------
class Permission(msgspec.Struct):
class Permission(msgspec.Struct, omit_defaults=True):
"""A permission that can be granted to roles."""
id: str # String primary key (max 128 chars)
uuid: UUID # UUID primary key
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
class Role(msgspec.Struct):
@@ -135,8 +137,10 @@ class SessionContext(msgspec.Struct):
# -------------------------------------------------------------------------
class _PermissionData(msgspec.Struct):
class _PermissionData(msgspec.Struct, omit_defaults=True):
scope: str # Permission scope identifier
display_name: str
domain: str | None = None
orgs: dict[str, bool] = {} # org_uuid -> True (which orgs can grant this)
@@ -477,14 +481,14 @@ class DB:
def _build_org(self, org_uuid: str, include_roles: bool = False) -> Org:
"""Build an Org object from internal storage. Caller must hold lock."""
o = self._data.orgs[org_uuid]
# Get permissions this org can grant
perm_ids = [
pid for pid, p in self._data.permissions.items() if org_uuid in p.orgs
# Get permission scopes this org can grant
perm_scopes = [
p.scope for pid, p in self._data.permissions.items() if org_uuid in p.orgs
]
org = Org(
uuid=UUID(org_uuid),
display_name=o.display_name,
permissions=perm_ids,
permissions=perm_scopes,
)
if include_roles:
org.roles = [
@@ -642,7 +646,9 @@ class DB:
return
raise ValueError("Credential not found")
def delete_credential(self, uuid: UUID, user_uuid: UUID, actor: str = "system") -> None:
def delete_credential(
self, uuid: UUID, user_uuid: UUID, actor: str = "system"
) -> None:
with self.session(actor):
key = str(uuid)
if key not in self._data.credentials:
@@ -784,30 +790,45 @@ class DB:
# -------------------------------------------------------------------------
def create_organization(self, org: Org, actor: str = "system") -> None:
import uuid7
with self.session(actor):
key = str(org.uuid)
self._data.orgs[key] = _OrgData(
display_name=org.display_name,
)
# Update permissions to allow this org to grant them
for perm_id in org.permissions:
if perm_id in self._data.permissions:
self._data.permissions[perm_id].orgs[key] = True
# Update permissions to allow this org to grant them (by scope)
for perm_scope in org.permissions:
# Find permission by scope and add org
for pid, p in self._data.permissions.items():
if p.scope == perm_scope:
p.orgs[key] = True
break
# Automatically create an organization admin permission if not present
auto_perm_id = f"auth:org:{org.uuid}"
if auto_perm_id not in self._data.permissions:
self._data.permissions[auto_perm_id] = _PermissionData(
display_name=f"{org.display_name} Admin",
orgs={key: True}, # This org can grant its own admin permission
# Automatically create or enable the common org admin permission
org_admin_scope = "auth:org:admin"
org_admin_key = None
for pid, p in self._data.permissions.items():
if p.scope == org_admin_scope:
org_admin_key = pid
break
if org_admin_key is None:
# Create the common org admin permission
org_admin_key = str(uuid7.create())
self._data.permissions[org_admin_key] = _PermissionData(
scope=org_admin_scope,
display_name="Organization Admin",
orgs={key: True},
)
else:
# Ensure this org can grant its own admin permission
self._data.permissions[auto_perm_id].orgs[key] = True
# Reflect the automatically added permission in the dataclass instance
if auto_perm_id not in org.permissions:
org.permissions.append(auto_perm_id)
# Ensure this org can grant the org admin permission
self._data.permissions[org_admin_key].orgs[key] = True
# Reflect the org admin permission in the dataclass instance
if org_admin_scope not in org.permissions:
org.permissions.append(org_admin_scope)
def get_organization(self, org_id: str) -> Org:
with self._lock:
@@ -828,15 +849,17 @@ class DB:
if key not in self._data.orgs:
raise ValueError("Organization not found")
self._data.orgs[key].display_name = org.display_name
# Update which permissions this org can grant
# Update which permissions this org can grant (by scope)
# First remove this org from all permissions
for p in self._data.permissions.values():
if key in p.orgs:
del p.orgs[key]
# Then add this org to the specified permissions
for perm_id in org.permissions:
if perm_id in self._data.permissions:
self._data.permissions[perm_id].orgs[key] = True
# Then add this org to the specified permissions (by scope)
for perm_scope in org.permissions:
for pid, p in self._data.permissions.items():
if p.scope == perm_scope:
p.orgs[key] = True
break
def delete_organization(self, org_uuid: UUID, actor: str = "system") -> None:
with self.session(actor):
@@ -889,7 +912,9 @@ class DB:
with self._lock:
# Get all roles for this org
org_role_uuids = {
role_uuid for role_uuid, r in self._data.roles.items() if r.org == org_id
role_uuid
for role_uuid, r in self._data.roles.items()
if r.org == org_id
}
return [
(self._build_user(user_uuid), self._data.roles[u.role].display_name)
@@ -905,9 +930,7 @@ class DB:
if r.org == org_id
]
def get_user_role_in_organization(
self, user_uuid: UUID, org_id: str
) -> str | None:
def get_user_role_in_organization(self, user_uuid: UUID, org_id: str) -> str | None:
with self._lock:
user_key = str(user_uuid)
if user_key not in self._data.users:
@@ -947,83 +970,167 @@ class DB:
def create_permission(self, permission: Permission, actor: str = "system") -> None:
with self.session(actor):
self._data.permissions[permission.id] = _PermissionData(
key = str(permission.uuid)
self._data.permissions[key] = _PermissionData(
scope=permission.scope,
display_name=permission.display_name,
domain=permission.domain,
orgs={}, # Will be populated when orgs are allowed to grant this permission
)
def get_permission(self, permission_id: str) -> Permission:
"""Get a permission by UUID string or scope.
For backwards compatibility, this accepts either:
- A UUID string (the primary key)
- A scope string (searches for matching scope)
"""
with self._lock:
if permission_id not in self._data.permissions:
raise ValueError("Permission not found")
p = self._data.permissions[permission_id]
return Permission(id=permission_id, display_name=p.display_name)
# First try as UUID key
if permission_id in self._data.permissions:
p = self._data.permissions[permission_id]
return Permission(
uuid=UUID(permission_id),
scope=p.scope,
display_name=p.display_name,
domain=p.domain,
)
# Fall back to scope search
for pid, p in self._data.permissions.items():
if p.scope == permission_id:
return Permission(
uuid=UUID(pid),
scope=p.scope,
display_name=p.display_name,
domain=p.domain,
)
raise ValueError("Permission not found")
def get_permission_by_scope(self, scope: str) -> Permission | None:
"""Get a permission by its scope string."""
with self._lock:
for pid, p in self._data.permissions.items():
if p.scope == scope:
return Permission(
uuid=UUID(pid),
scope=p.scope,
display_name=p.display_name,
domain=p.domain,
)
return None
def list_permissions(self) -> list[Permission]:
with self._lock:
return [
Permission(id=pid, display_name=p.display_name)
Permission(
uuid=UUID(pid),
scope=p.scope,
display_name=p.display_name,
domain=p.domain,
)
for pid, p in self._data.permissions.items()
]
def update_permission(self, permission: Permission, actor: str = "system") -> None:
with self.session(actor):
if permission.id not in self._data.permissions:
key = str(permission.uuid)
if key not in self._data.permissions:
raise ValueError("Permission not found")
self._data.permissions[permission.id].display_name = permission.display_name
self._data.permissions[key].scope = permission.scope
self._data.permissions[key].display_name = permission.display_name
self._data.permissions[key].domain = permission.domain
def delete_permission(self, permission_id: str, actor: str = "system") -> None:
"""Delete a permission by UUID string or scope."""
with self.session(actor):
if permission_id in self._data.permissions:
del self._data.permissions[permission_id]
# Remove from roles (permissions is a dict)
for r in self._data.roles.values():
if permission_id in r.permissions:
del r.permissions[permission_id]
# Find the UUID key
key = self._resolve_permission_key(permission_id)
if key and key in self._data.permissions:
scope = self._data.permissions[key].scope
del self._data.permissions[key]
# Remove from roles (roles store scopes, not UUIDs)
for r in self._data.roles.values():
if scope in r.permissions:
del r.permissions[scope]
def _resolve_permission_key(self, permission_id: str) -> str | None:
"""Resolve a permission_id (UUID or scope) to its UUID key."""
if permission_id in self._data.permissions:
return permission_id
for pid, p in self._data.permissions.items():
if p.scope == permission_id:
return pid
return None
def _resolve_permission_scope(self, permission_id: str) -> str | None:
"""Resolve a permission_id (UUID or scope) to its scope."""
if permission_id in self._data.permissions:
return self._data.permissions[permission_id].scope
for pid, p in self._data.permissions.items():
if p.scope == permission_id:
return p.scope
return None
def rename_permission(
self, old_id: str, new_id: str, display_name: str, actor: str = "system"
self,
old_scope: str,
new_scope: str,
display_name: str,
domain: str | None = None,
actor: str = "system",
) -> None:
"""Rename a permission's scope. The UUID remains the same."""
with self.session(actor):
if old_id == new_id:
if old_id in self._data.permissions:
self._data.permissions[old_id].display_name = display_name
return
if old_id not in self._data.permissions:
# Find the permission by scope
key = self._resolve_permission_key(old_scope)
if not key:
raise ValueError("Original permission not found")
if new_id in self._data.permissions:
raise ValueError("New permission id already exists")
# Create new permission with same orgs
old_perm = self._data.permissions[old_id]
self._data.permissions[new_id] = _PermissionData(
display_name=display_name,
orgs=dict(old_perm.orgs),
)
# Update role references (roles store permissions as dict)
for r in self._data.roles.values():
if old_id in r.permissions:
del r.permissions[old_id]
r.permissions[new_id] = True
# Delete old permission
del self._data.permissions[old_id]
# Check if new scope already exists
for pid, p in self._data.permissions.items():
if p.scope == new_scope and pid != key:
raise ValueError("New permission scope already exists")
old_scope_value = self._data.permissions[key].scope
# Update the permission
self._data.permissions[key].scope = new_scope
self._data.permissions[key].display_name = display_name
self._data.permissions[key].domain = domain
# Update role references if scope changed
if old_scope_value != new_scope:
for r in self._data.roles.values():
if old_scope_value in r.permissions:
del r.permissions[old_scope_value]
r.permissions[new_scope] = True
def add_permission_to_organization(
self, org_id: str, permission_id: str, actor: str = "system"
) -> None:
"""Add a permission to an organization (allows org to grant it).
permission_id can be a UUID string or a scope string.
"""
with self.session(actor):
if org_id not in self._data.orgs:
raise ValueError("Organization not found")
if permission_id not in self._data.permissions:
key = self._resolve_permission_key(permission_id)
if not key:
raise ValueError("Permission not found")
self._data.permissions[permission_id].orgs[org_id] = True
self._data.permissions[key].orgs[org_id] = True
def remove_permission_from_organization(
self, org_id: str, permission_id: str, actor: str = "system"
) -> None:
"""Remove a permission from an organization.
permission_id can be a UUID string or a scope string.
"""
with self.session(actor):
if permission_id in self._data.permissions:
orgs = self._data.permissions[permission_id].orgs
key = self._resolve_permission_key(permission_id)
if key and key in self._data.permissions:
orgs = self._data.permissions[key].orgs
if org_id in orgs:
del orgs[org_id]
@@ -1034,14 +1141,26 @@ class DB:
permissions = []
for pid, p in self._data.permissions.items():
if org_id in p.orgs:
permissions.append(Permission(id=pid, display_name=p.display_name))
permissions.append(
Permission(
uuid=UUID(pid),
scope=p.scope,
display_name=p.display_name,
domain=p.domain,
)
)
return permissions
def get_permission_organizations(self, permission_id: str) -> list[Org]:
"""Get organizations that can grant a permission.
permission_id can be a UUID string or a scope string.
"""
with self._lock:
if permission_id not in self._data.permissions:
key = self._resolve_permission_key(permission_id)
if not key or key not in self._data.permissions:
return []
org_ids = self._data.permissions[permission_id].orgs
org_ids = self._data.permissions[key].orgs
return [
self._build_org(org_id)
for org_id in org_ids
@@ -1055,49 +1174,86 @@ class DB:
def add_permission_to_role(
self, role_uuid: UUID, permission_id: str, actor: str = "system"
) -> None:
"""Add a permission to a role.
permission_id can be a UUID string or a scope string.
Stores the scope in the role's permissions dict.
"""
with self.session(actor):
key = str(role_uuid)
if key not in self._data.roles:
raise ValueError("Role not found")
if permission_id not in self._data.permissions:
scope = self._resolve_permission_scope(permission_id)
if not scope:
raise ValueError("Permission not found")
self._data.roles[key].permissions[permission_id] = True
self._data.roles[key].permissions[scope] = True
def remove_permission_from_role(
self, role_uuid: UUID, permission_id: str, actor: str = "system"
) -> None:
"""Remove a permission from a role.
permission_id can be a UUID string or a scope string.
"""
with self.session(actor):
key = str(role_uuid)
if key in self._data.roles:
if permission_id in self._data.roles[key].permissions:
# Try to find the scope
scope = self._resolve_permission_scope(permission_id)
if scope and scope in self._data.roles[key].permissions:
del self._data.roles[key].permissions[scope]
# Also try the raw permission_id in case it's already a scope
elif permission_id in self._data.roles[key].permissions:
del self._data.roles[key].permissions[permission_id]
def get_role_permissions(self, role_uuid: UUID) -> list[Permission]:
"""Get permissions granted by a role.
Note: Roles store scopes, so we need to look up permissions by scope.
"""
with self._lock:
key = str(role_uuid)
if key not in self._data.roles:
return []
perm_ids = list(self._data.roles[key].permissions)
scopes = list(self._data.roles[key].permissions.keys())
permissions = []
for pid in perm_ids:
if pid in self._data.permissions:
p = self._data.permissions[pid]
permissions.append(Permission(id=pid, display_name=p.display_name))
for scope in scopes:
# Find permission with this scope
for pid, p in self._data.permissions.items():
if p.scope == scope:
permissions.append(
Permission(
uuid=UUID(pid),
scope=p.scope,
display_name=p.display_name,
domain=p.domain,
)
)
break
return permissions
def get_permission_roles(self, permission_id: str) -> list[Role]:
"""Get roles that have a permission.
permission_id can be a UUID string or a scope string.
"""
with self._lock:
scope = self._resolve_permission_scope(permission_id)
if not scope:
return []
return [
self._build_role(role_uuid)
for role_uuid, r in self._data.roles.items()
if permission_id in r.permissions
if scope in r.permissions
]
# -------------------------------------------------------------------------
# Combined operations
# -------------------------------------------------------------------------
def login(self, user_uuid: UUID, credential: Credential, actor: str = "system") -> None:
def login(
self, user_uuid: UUID, credential: Credential, actor: str = "system"
) -> None:
with self.session(actor):
# Update credential
for key, c in self._data.credentials.items():
@@ -1277,12 +1433,30 @@ class DB:
else None
)
# Effective permissions: role permissions that the org can grant
effective_permissions = [
Permission(id=pid, display_name=self._data.permissions[pid].display_name)
for pid in role_obj.permissions
if pid in self._data.permissions and pid in org_obj.permissions
]
# Effective permissions: role permissions (scopes) that the org can grant
# role_obj.permissions contains scopes, org_obj.permissions contains scopes
from paskia.util.hostutil import normalize_host
normalized_host = normalize_host(host)
effective_permissions = []
for scope in role_obj.permissions:
if scope not in org_obj.permissions:
continue
# Find the permission by scope
for pid, p in self._data.permissions.items():
if p.scope == scope:
# Check domain restriction
if p.domain is not None and p.domain != normalized_host:
continue
effective_permissions.append(
Permission(
uuid=UUID(pid),
scope=p.scope,
display_name=p.display_name,
domain=p.domain,
)
)
break
return SessionContext(
session=session_obj,
+191 -83
View File
@@ -23,6 +23,30 @@ from paskia.util.tokens import encode_session_key, session_key
app = FastAPI()
def is_global_admin(ctx) -> bool:
"""Check if user has global admin permission."""
return "auth:admin" in ctx.role.permissions
def is_org_admin(ctx, org_uuid: UUID | None = None) -> bool:
"""Check if user has org admin permission.
If org_uuid is provided, checks if user is admin of that specific org.
If org_uuid is None, checks if user is admin of their own org.
"""
if "auth:org:admin" not in ctx.role.permissions:
return False
if org_uuid is None:
return True
# User must belong to the target org (via their role)
return ctx.org.uuid == org_uuid
def can_manage_org(ctx, org_uuid: UUID) -> bool:
"""Check if user can manage the specified organization."""
return is_global_admin(ctx) or is_org_admin(ctx, org_uuid)
@app.exception_handler(ValueError)
async def value_error_handler(_request, exc: ValueError): # pragma: no cover - simple
return JSONResponse(status_code=400, content={"detail": str(exc)})
@@ -55,13 +79,14 @@ async def adminapp(request: Request, auth=AUTH_COOKIE):
async def admin_list_orgs(request: Request, auth=AUTH_COOKIE):
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:*"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
orgs = db.list_organizations()
if "auth:admin" not in ctx.role.permissions:
orgs = [o for o in orgs if f"auth:org:{o.uuid}" in ctx.role.permissions]
if not is_global_admin(ctx):
# Org admins can only see their own organization
orgs = [o for o in orgs if o.uuid == ctx.org.uuid]
def role_to_dict(r):
return {
@@ -110,12 +135,13 @@ async def admin_create_org(
db.create_organization(org)
# Automatically create Administration role with org admin permission
# The auth:org:admin permission is automatically created/enabled by create_organization
role_uuid = uuid4()
admin_role = RoleDC(
uuid=role_uuid,
org_uuid=org_uuid,
display_name="Administration",
permissions=[f"auth:org:{org_uuid}"],
permissions=["auth:org:admin"],
)
db.create_role(admin_role)
@@ -131,10 +157,14 @@ async def admin_update_org(
):
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
from ..db import Org as OrgDC # local import to avoid cycles
current = db.get_organization(str(org_uuid))
@@ -144,13 +174,10 @@ async def admin_update_org(
permissions = current.permissions or []
# Sanity check: prevent removing permissions that would break current user's admin access
org_admin_perm = f"auth:org:{org_uuid}"
org_admin_perm = "auth:org:admin"
# If current user is org admin (not global admin), ensure org admin perm remains
if (
"auth:admin" not in ctx.role.permissions
and f"auth:org:{org_uuid}" in ctx.role.permissions
):
if not is_global_admin(ctx) and is_org_admin(ctx, org_uuid):
if org_admin_perm not in permissions:
raise ValueError(
"Cannot remove organization admin permission from your own organization"
@@ -165,11 +192,15 @@ async def admin_update_org(
async def admin_delete_org(org_uuid: UUID, request: Request, auth=AUTH_COOKIE):
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
max_age="5m",
)
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
if ctx.org.uuid == org_uuid:
raise ValueError("Cannot delete the organization you belong to")
@@ -177,15 +208,15 @@ async def admin_delete_org(org_uuid: UUID, request: Request, auth=AUTH_COOKIE):
org_perm_pattern = f"org:{str(org_uuid).lower()}"
all_permissions = db.list_permissions()
for perm in all_permissions:
perm_id_lower = perm.id.lower()
perm_scope_lower = perm.scope.lower()
# Check if permission contains "org:{uuid}" separated by colons or at boundaries
if (
f":{org_perm_pattern}:" in perm_id_lower
or perm_id_lower.startswith(f"{org_perm_pattern}:")
or perm_id_lower.endswith(f":{org_perm_pattern}")
or perm_id_lower == org_perm_pattern
f":{org_perm_pattern}:" in perm_scope_lower
or perm_scope_lower.startswith(f"{org_perm_pattern}:")
or perm_scope_lower.endswith(f":{org_perm_pattern}")
or perm_scope_lower == org_perm_pattern
):
db.delete_permission(perm.id)
db.delete_permission(str(perm.uuid))
db.delete_organization(org_uuid)
return {"status": "ok"}
@@ -229,12 +260,16 @@ async def admin_create_role(
payload: dict = Body(...),
auth=AUTH_COOKIE,
):
await authz.verify(
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
from ..db import Role as RoleDC
role_uuid = uuid4()
@@ -267,10 +302,14 @@ async def admin_update_role(
# Verify caller is global admin or admin of provided org
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
role = db.get_role(role_uuid)
if role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization")
@@ -291,7 +330,7 @@ async def admin_update_role(
# Sanity check: prevent admin from removing their own access via role update
if ctx.org.uuid == org_uuid and ctx.role.uuid == role_uuid:
has_admin_access = (
"auth:admin" in permissions or f"auth:org:{org_uuid}" in permissions
"auth:admin" in permissions or "auth:org:admin" in permissions
)
if not has_admin_access:
raise ValueError("Cannot update your own role to remove admin permissions")
@@ -315,11 +354,15 @@ async def admin_delete_role(
):
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
max_age="5m",
)
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
role = db.get_role(role_uuid)
if role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization")
@@ -342,12 +385,16 @@ async def admin_create_user(
payload: dict = Body(...),
auth=AUTH_COOKIE,
):
await authz.verify(
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
display_name = payload.get("display_name")
role_name = payload.get("role")
if not display_name or not role_name:
@@ -380,10 +427,14 @@ async def admin_update_user_role(
):
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
new_role = payload.get("role")
if not new_role:
raise ValueError("role is required")
@@ -403,7 +454,7 @@ async def admin_update_user_role(
if new_role_obj: # pragma: no branch - always true, role validated above
has_admin_access = (
"auth:admin" in new_role_obj.permissions
or f"auth:org:{org_uuid}" in new_role_obj.permissions
or "auth:org:admin" in new_role_obj.permissions
)
if not has_admin_access:
raise ValueError(
@@ -429,15 +480,12 @@ async def admin_create_user_registration_link(
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
max_age="5m",
)
if ( # pragma: no cover - defense in depth, authz.verify already checked
"auth:admin" not in ctx.role.permissions
and f"auth:org:{org_uuid}" not in ctx.role.permissions
):
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -480,14 +528,11 @@ async def admin_get_user_detail(
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if ( # pragma: no cover - defense in depth, authz.verify already checked
"auth:admin" not in ctx.role.permissions
and f"auth:org:{org_uuid}" not in ctx.role.permissions
):
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -565,9 +610,7 @@ async def admin_get_user_detail(
"ip": entry.ip,
"user_agent": useragent.compact_user_agent(entry.user_agent),
"last_renewed": (
renewed.astimezone(timezone.utc)
.isoformat()
.replace("+00:00", "Z")
renewed.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
if renewed.tzinfo
else renewed.replace(tzinfo=timezone.utc)
.isoformat()
@@ -631,14 +674,11 @@ async def admin_update_user_display_name(
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if ( # pragma: no cover - defense in depth, authz.verify already checked
"auth:admin" not in ctx.role.permissions
and f"auth:org:{org_uuid}" not in ctx.role.permissions
):
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -667,15 +707,12 @@ async def admin_delete_user_credential(
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
max_age="5m",
)
if ( # pragma: no cover - defense in depth, authz.verify already checked
"auth:admin" not in ctx.role.permissions
and f"auth:org:{org_uuid}" not in ctx.role.permissions
):
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -699,14 +736,11 @@ async def admin_delete_user_session(
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", f"auth:org:{org_uuid}"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if ( # pragma: no cover - defense in depth, authz.verify already checked
"auth:admin" not in ctx.role.permissions
and f"auth:org:{org_uuid}" not in ctx.role.permissions
):
if not can_manage_org(ctx, org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -732,24 +766,46 @@ async def admin_delete_user_session(
# -------------------- Permissions (global) --------------------
def _perm_to_dict(p):
"""Convert Permission to dict, omitting domain if None."""
d = {"uuid": str(p.uuid), "scope": p.scope, "display_name": p.display_name}
if p.domain is not None:
d["domain"] = p.domain
return d
def _validate_permission_domain(domain: str | None) -> None:
"""Validate that domain is rp_id or a subdomain of it."""
if domain is None:
return
from paskia.globals import global_passkey
rp_id = global_passkey.instance.rp_id
if domain == rp_id or domain.endswith(f".{rp_id}"):
return
raise ValueError(
f"Domain '{domain}' must be the same as or a subdomain of rp_id '{rp_id}'"
)
@app.get("/permissions")
async def admin_list_permissions(request: Request, auth=AUTH_COOKIE):
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:*"],
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
perms = db.list_permissions()
# Global admins see all permissions
if "auth:admin" in ctx.role.permissions:
return [{"id": p.id, "display_name": p.display_name} for p in perms]
if is_global_admin(ctx):
return [_perm_to_dict(p) for p in perms]
# Org admins only see permissions their org can grant
grantable = set(ctx.org.permissions or [])
filtered_perms = [p for p in perms if p.id in grantable]
return [{"id": p.id, "display_name": p.display_name} for p in filtered_perms]
filtered_perms = [p for p in perms if p.scope in grantable]
return [_perm_to_dict(p) for p in filtered_perms]
@app.post("/permissions")
@@ -765,34 +821,67 @@ async def admin_create_permission(
match=permutil.has_all,
max_age="5m",
)
import uuid7
from ..db import Permission as PermDC
perm_id = payload.get("id")
scope = payload.get("scope") or payload.get(
"id"
) # Support both for backwards compat
display_name = payload.get("display_name")
if not perm_id or not display_name:
raise ValueError("id and display_name are required")
querysafe.assert_safe(perm_id, field="id")
db.create_permission(PermDC(id=perm_id, display_name=display_name))
domain = payload.get("domain") or None # Treat empty string as None
if not scope or not display_name:
raise ValueError("scope and display_name are required")
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
)
)
return {"status": "ok"}
@app.put("/permission")
async def admin_update_permission(
permission_id: str,
display_name: str,
request: Request,
auth=AUTH_COOKIE,
permission_uuid: str | None = None,
permission_id: str | None = None, # Backwards compat - treated as scope
display_name: str | None = None,
scope: str | None = None,
domain: str | None = None,
):
await authz.verify(
auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
)
# permission_uuid or permission_id (scope) to identify the permission
perm_identifier = permission_uuid or permission_id
if not perm_identifier:
raise ValueError("permission_uuid or permission_id required")
# Get existing permission
perm = db.get_permission(perm_identifier)
# Update fields that were provided
new_scope = scope if scope is not None else perm.scope
new_display_name = display_name if display_name is not None else perm.display_name
domain_value = domain if domain else None
if not new_display_name:
raise ValueError("display_name is required")
querysafe.assert_safe(new_scope, field="scope")
_validate_permission_domain(domain_value)
from ..db import Permission as PermDC
if not display_name:
raise ValueError("display_name is required")
querysafe.assert_safe(permission_id, field="permission_id")
db.update_permission(
PermDC(id=permission_id, display_name=display_name)
PermDC(
uuid=perm.uuid,
scope=new_scope,
display_name=new_display_name,
domain=domain_value,
)
)
return {"status": "ok"}
@@ -806,31 +895,43 @@ async def admin_rename_permission(
await authz.verify(
auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
)
old_id = payload.get("old_id")
new_id = payload.get("new_id")
old_scope = payload.get("old_scope") or payload.get("old_id") # Support both
new_scope = payload.get("new_scope") or payload.get("new_id") # Support both
display_name = payload.get("display_name")
if not old_id or not new_id:
raise ValueError("old_id and new_id required")
domain = payload.get(
"domain"
) # Can be None (not provided), empty string (clear), or value
if not old_scope or not new_scope:
raise ValueError("old_scope and new_scope required")
# Sanity check: prevent renaming critical permissions
if old_id == "auth:admin":
if old_scope == "auth:admin":
raise ValueError("Cannot rename the master admin permission")
querysafe.assert_safe(old_id, field="old_id")
querysafe.assert_safe(new_id, field="new_id")
querysafe.assert_safe(old_scope, field="old_scope")
querysafe.assert_safe(new_scope, field="new_scope")
# Get existing permission to preserve values not being changed
perm = db.get_permission(old_scope)
if display_name is None:
perm = db.get_permission(old_id)
display_name = perm.display_name
# domain=None means "not provided, keep existing", domain="" means "clear it"
if domain is None:
domain_value = perm.domain
else:
domain_value = domain if domain else None
_validate_permission_domain(domain_value)
# All current backends support rename_permission
db.rename_permission(old_id, new_id, display_name)
db.rename_permission(old_scope, new_scope, display_name, domain_value)
return {"status": "ok"}
@app.delete("/permission")
async def admin_delete_permission(
permission_id: str,
request: Request,
auth=AUTH_COOKIE,
permission_uuid: str | None = None,
permission_id: str | None = None, # Backwards compat - treated as scope
):
await authz.verify(
auth,
@@ -839,11 +940,18 @@ async def admin_delete_permission(
match=permutil.has_all,
max_age="5m",
)
querysafe.assert_safe(permission_id, field="permission_id")
perm_identifier = permission_uuid or permission_id
if not perm_identifier:
raise ValueError("permission_uuid or permission_id required")
querysafe.assert_safe(perm_identifier, field="permission_id")
# Get the permission to check its scope
perm = db.get_permission(perm_identifier)
# Sanity check: prevent deleting critical permissions
if permission_id == "auth:admin":
if perm.scope == "auth:admin":
raise ValueError("Cannot delete the master admin permission")
db.delete_permission(permission_id)
db.delete_permission(str(perm.uuid))
return {"status": "ok"}
+3 -5
View File
@@ -147,7 +147,7 @@ async def forward_authentication(
)
role_permissions = set(ctx.role.permissions or [])
if ctx.permissions:
role_permissions.update(permission.id for permission in ctx.permissions)
role_permissions.update(permission.scope for permission in ctx.permissions)
remote_headers: dict[str, str] = {
"Remote-User": str(ctx.user.uuid),
@@ -158,13 +158,11 @@ async def forward_authentication(
"Remote-Role": str(ctx.role.uuid),
"Remote-Role-Name": ctx.role.display_name,
"Remote-Session-Expires": (
ctx.session.expiry
.astimezone(timezone.utc)
ctx.session.expiry.astimezone(timezone.utc)
.isoformat()
.replace("+00:00", "Z")
if ctx.session.expiry.tzinfo
else ctx.session.expiry
.replace(tzinfo=timezone.utc)
else ctx.session.expiry.replace(tzinfo=timezone.utc)
.isoformat()
.replace("+00:00", "Z")
),
+1 -3
View File
@@ -324,9 +324,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
# Fetch and verify credential
try:
stored_cred = db.get_credential_by_id(
credential.raw_id
)
stored_cred = db.get_credential_by_id(credential.raw_id)
except ValueError:
raise ValueError(
f"This passkey is no longer registered with {passkey.instance.rp_name}"
+1 -3
View File
@@ -116,9 +116,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
try:
session = await get_session(auth, host=host)
session_user_uuid = session.user_uuid
credential_ids = db.get_credentials_by_user_uuid(
session_user_uuid
)
credential_ids = db.get_credentials_by_user_uuid(session_user_uuid)
except ValueError:
pass # Invalid/expired session - allow normal authentication
+62 -11
View File
@@ -54,6 +54,9 @@ async def migrate_from_sql(
json_db_path: Path for the destination JSONL file
"""
# Import here to avoid circular imports and to not require JSON db at import time
import re
import uuid7
from sqlalchemy import select
from paskia.db.json import (
@@ -79,14 +82,53 @@ async def migrate_from_sql(
# Build all data directly without saving (we'll save once at the end)
with json_db._lock:
# Migrate permissions
# Track old permission ID -> new scope mapping for migration
# Also track org-specific admin permissions to consolidate
old_org_admin_pattern = re.compile(r"^auth:org:([0-9a-f-]+)$", re.IGNORECASE)
org_admin_uuids = set() # org UUIDs that had org-specific admin permissions
# First pass: identify org-specific admin permissions
permissions = await sql_db.list_permissions()
for perm in permissions:
json_db._data.permissions[perm.id] = _PermissionData(
match = old_org_admin_pattern.match(perm.id)
if match:
org_admin_uuids.add(match.group(1).lower())
# 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 = str(uuid7.create())
json_db._data.permissions[org_admin_perm_uuid] = _PermissionData(
scope="auth:org:admin",
display_name="Organization Admin",
orgs={},
)
# Mapping from old permission ID to new scope
perm_id_to_scope: dict[str, str] = {}
for perm in permissions:
# Skip old org-specific admin permissions (auth:org:{uuid}) - they map to auth:org:admin
match = old_org_admin_pattern.match(perm.id)
if match:
perm_id_to_scope[perm.id] = "auth:org:admin"
continue
# Skip if this is already auth:org:admin - we created one above
if perm.id == "auth:org:admin":
perm_id_to_scope[perm.id] = "auth:org:admin"
continue
# Regular permission - create with UUID key
perm_uuid = str(uuid7.create())
json_db._data.permissions[perm_uuid] = _PermissionData(
scope=perm.id, # Old ID becomes the scope
display_name=perm.display_name,
orgs={},
)
print(f" Migrated {len(permissions)} permissions")
perm_id_to_scope[perm.id] = perm.id # Scope same as old ID
print(
f" Migrated {len(permissions)} permissions (with {len(org_admin_uuids)} org-specific admins consolidated to auth:org:admin)"
)
# Migrate organizations
orgs = await sql_db.list_organizations()
@@ -95,23 +137,32 @@ async def migrate_from_sql(
json_db._data.orgs[key] = _OrgData(
display_name=org.display_name,
)
# Update permissions to allow this org to grant them
for perm_id in org.permissions:
if perm_id in json_db._data.permissions:
json_db._data.permissions[perm_id].orgs[key] = True
# Update permissions to allow this org to grant them (by scope)
for old_perm_id in org.permissions:
new_scope = perm_id_to_scope.get(old_perm_id, old_perm_id)
# Find permission with this scope and add org
for pid, p in json_db._data.permissions.items():
if p.scope == new_scope:
p.orgs[key] = True
break
# Ensure every org can grant auth:org:admin
json_db._data.permissions[org_admin_perm_uuid].orgs[key] = True
print(f" Migrated {len(orgs)} organizations")
# Migrate roles
# Migrate roles - convert old permission IDs to scopes
role_count = 0
for org in orgs:
for role in org.roles:
key = str(role.uuid)
# Convert old permission IDs to scopes
new_permissions = {}
for old_perm_id in role.permissions or []:
new_scope = perm_id_to_scope.get(old_perm_id, old_perm_id)
new_permissions[new_scope] = True
json_db._data.roles[key] = _RoleData(
org=str(role.org_uuid),
display_name=role.display_name,
permissions={p: True for p in role.permissions}
if role.permissions
else {},
permissions=new_permissions,
)
role_count += 1
print(f" Migrated {role_count} roles")
+18 -5
View File
@@ -28,12 +28,21 @@ from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from paskia.db import (
Credential,
Org,
Permission,
ResetToken,
Role,
User,
)
# Local Permission class for SQL schema (uses 'id' not 'uuid' + 'scope')
@dataclass
class SqlPermission:
"""Permission as stored in the old SQL schema with id field."""
id: str
display_name: str
DB_PATH_DEFAULT = "sqlite+aiosqlite:///paskia.sqlite"
@@ -41,6 +50,7 @@ DB_PATH_DEFAULT = "sqlite+aiosqlite:///paskia.sqlite"
@dataclass
class _SqlSession:
"""Session as stored in the old SQL schema with renewed timestamp."""
key: bytes
user_uuid: UUID
credential_uuid: UUID
@@ -249,11 +259,14 @@ class PermissionModel(Base):
display_name: Mapped[str] = mapped_column(String, nullable=False)
def as_dataclass(self):
return Permission(self.id, self.display_name)
return SqlPermission(self.id, self.display_name)
@staticmethod
def from_dataclass(permission: Permission):
return PermissionModel(id=permission.id, display_name=permission.display_name)
def from_dataclass(permission: SqlPermission):
return PermissionModel(
id=permission.id,
display_name=permission.display_name,
)
class OrgPermission(Base):
@@ -320,7 +333,7 @@ class DB:
async with self.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async def list_permissions(self) -> list[Permission]:
async def list_permissions(self) -> list[SqlPermission]:
async with self.session() as session:
result = await session.execute(select(PermissionModel))
return [p.as_dataclass() for p in result.scalars().all()]
+2 -11
View File
@@ -2,9 +2,8 @@
from datetime import timezone
from paskia import aaguid
from paskia import aaguid, db
from paskia.authsession import EXPIRES, session_key
from paskia import db
from paskia.util import hostutil, permutil, tokens, useragent
@@ -76,8 +75,6 @@ async def format_user_info(
role_info = None
org_info = None
effective_permissions: list[str] = []
is_global_admin = False
is_org_admin = False
if ctx:
role_info = {
@@ -90,11 +87,7 @@ async def format_user_info(
"display_name": ctx.org.display_name,
"permissions": ctx.org.permissions,
}
effective_permissions = [p.id for p in (ctx.permissions or [])]
is_global_admin = "auth:admin" in (role_info["permissions"] or [])
is_org_admin = any(
p.startswith("auth:org:") for p in (role_info["permissions"] or [])
)
effective_permissions = [p.scope for p in (ctx.permissions or [])]
# Format sessions
normalized_request_host = hostutil.normalize_host(request_host)
@@ -132,8 +125,6 @@ async def format_user_info(
"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,
"sessions": sessions_payload,
+26 -3
View File
@@ -84,19 +84,42 @@ 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."""
perm = Permission(id="auth:admin", display_name="Master Admin")
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="auth:admin", display_name="Master Admin"
)
test_db.create_permission(perm)
return perm
@pytest_asyncio.fixture(scope="function")
async def test_role(test_db: DB, test_org: Org, admin_permission: Permission) -> Role:
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"
)
test_db.create_permission(perm)
# Make it grantable by the org
test_db.add_permission_to_organization(str(test_org.uuid), "auth:org:admin")
return perm
@pytest_asyncio.fixture(scope="function")
async def test_role(
test_db: DB,
test_org: Org,
admin_permission: Permission,
org_admin_permission: Permission,
) -> Role:
"""Create a test role with admin permission."""
role = Role(
uuid=uuid7.create(),
org_uuid=test_org.uuid,
display_name="Test Admin Role",
permissions=["auth:admin", f"auth:org:{test_org.uuid}"],
permissions=["auth:admin", "auth:org:admin"],
)
test_db.create_role(role)
return role
+73 -40
View File
@@ -108,13 +108,13 @@ async def second_org_session_token(
@pytest_asyncio.fixture(scope="function")
async def org_admin_role(test_db: DB, test_org: Org) -> Role:
async def org_admin_role(test_db: DB, test_org: Org, org_admin_permission) -> Role:
"""Create a role with org admin permission only (no global admin)."""
role = Role(
uuid=uuid7.create(),
org_uuid=test_org.uuid,
display_name="Org Admin Role",
permissions=[f"auth:org:{test_org.uuid}"],
permissions=["auth:org:admin"],
)
test_db.create_role(role)
return role
@@ -176,10 +176,14 @@ 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."""
perm = Permission(id="test:grantable:perm", display_name="Grantable Perm")
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:grantable:perm", display_name="Grantable Perm"
)
test_db.create_permission(perm)
# Add to org's grantable permissions
test_db.add_permission_to_organization(str(test_org.uuid), perm.id)
test_db.add_permission_to_organization(str(test_org.uuid), perm.scope)
return perm
@@ -355,7 +359,7 @@ class TestAdminOrganizations:
f"/auth/api/admin/orgs/{test_org.uuid}",
json={
"display_name": "Org Admin Updated Name",
"permissions": [f"auth:org:{test_org.uuid}"], # Keep org admin perm
"permissions": ["auth:org:admin"], # Keep org admin perm
},
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
@@ -372,18 +376,7 @@ class TestAdminOrganizations:
test_db: DB,
):
"""Org admin cannot remove their org admin permission from org's permissions."""
# First create and add the org admin perm to the org's grantable perms
org_admin_perm_id = f"auth:org:{test_org.uuid}"
perm = Permission(id=org_admin_perm_id, display_name="Org Admin")
try:
test_db.create_permission(perm)
except Exception:
pass # Permission may already exist
# Add it to the org's permissions
test_db.add_permission_to_organization(
str(test_org.uuid), org_admin_perm_id
)
# The auth:org:admin perm is already created and added by org_admin_permission fixture
# Try to remove all permissions including org admin perm
response = await client.put(
@@ -419,6 +412,8 @@ class TestAdminOrganizations:
test_db: DB,
):
"""Admin should be able to delete another organization."""
import uuid7
# Create org to delete
org_to_delete = Org(
uuid=uuid7.create(),
@@ -429,7 +424,9 @@ class TestAdminOrganizations:
# Create some org-specific permissions to test cleanup
org_perm = Permission(
id=f"test:org:{org_to_delete.uuid}:feature", display_name="Org Feature"
uuid=uuid7.create(),
scope=f"test:org:{org_to_delete.uuid}:feature",
display_name="Org Feature",
)
test_db.create_permission(org_perm)
@@ -456,7 +453,7 @@ class TestAdminOrgPermissions:
# First create a permission
await client.post(
"/auth/api/admin/permissions",
json={"id": "test:org:addable", "display_name": "Addable"},
json={"scope": "test:org:addable", "display_name": "Addable"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -491,7 +488,7 @@ class TestAdminOrgPermissions:
# First create and add a permission
await client.post(
"/auth/api/admin/permissions",
json={"id": "test:org:removable", "display_name": "Removable"},
json={"scope": "test:org:removable", "display_name": "Removable"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
await client.post(
@@ -585,7 +582,7 @@ class TestAdminRoles:
f"/auth/api/admin/orgs/{test_org.uuid}/roles",
json={
"display_name": "Role With Perms",
"permissions": [grantable_permission.id],
"permissions": [grantable_permission.scope],
},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -603,7 +600,13 @@ class TestAdminRoles:
):
"""Creating role with non-grantable permission should fail."""
# Create permission but don't add to org
perm = Permission(id="test:not:grantable", display_name="Not Grantable")
import uuid7
perm = Permission(
uuid=uuid7.create(),
scope="test:not:grantable",
display_name="Not Grantable",
)
test_db.create_permission(perm)
response = await client.post(
@@ -658,7 +661,7 @@ class TestAdminRoles:
"""Admin should be able to add grantable permissions to role."""
response = await client.put(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}",
json={"permissions": [grantable_permission.id]},
json={"permissions": [grantable_permission.scope]},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -673,7 +676,13 @@ class TestAdminRoles:
test_db: DB,
):
"""Adding non-grantable permission to role should fail."""
perm = Permission(id="test:not:grantable:update", display_name="Not Grantable")
import uuid7
perm = Permission(
uuid=uuid7.create(),
scope="test:not:grantable:update",
display_name="Not Grantable",
)
test_db.create_permission(perm)
response = await client.put(
@@ -1291,8 +1300,8 @@ class TestAdminPermissions:
data = response.json()
assert isinstance(data, list)
# Should include at least auth:admin
perm_ids = [p["id"] for p in data]
assert "auth:admin" in perm_ids
perm_scopes = [p["scope"] for p in data]
assert "auth:admin" in perm_scopes
@pytest.mark.asyncio
async def test_list_permissions_org_admin(
@@ -1310,12 +1319,12 @@ class TestAdminPermissions:
assert response.status_code == 200
data = response.json()
# Should only see permissions the org can grant
perm_ids = [p["id"] for p in data]
assert grantable_permission.id in perm_ids
perm_scopes = [p["scope"] for p in data]
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_ids
assert "auth:admin" in perm_scopes
# Should also see auto-created org admin permission
assert f"auth:org:{test_org.uuid}" in perm_ids
assert "auth:org:admin" in perm_scopes
@pytest.mark.asyncio
async def test_create_permission(
@@ -1324,7 +1333,7 @@ class TestAdminPermissions:
"""Admin should be able to create new permissions."""
response = await client.post(
"/auth/api/admin/permissions",
json={"id": "test:create:permission", "display_name": "Test Permission"},
json={"scope": "test:create:permission", "display_name": "Test Permission"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1352,7 +1361,7 @@ class TestAdminPermissions:
"""Creating permission without admin should fail."""
response = await client.post(
"/auth/api/admin/permissions",
json={"id": "test:forbidden", "display_name": "Forbidden"},
json={"scope": "test:forbidden", "display_name": "Forbidden"},
headers={
**auth_headers(regular_session_token),
"Host": "localhost:4401",
@@ -1366,7 +1375,11 @@ class TestAdminPermissions:
):
"""Admin should be able to update a permission."""
# Create permission first
perm = Permission(id="test:updateable", display_name="Updateable")
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:updateable", display_name="Updateable"
)
test_db.create_permission(perm)
response = await client.put(
@@ -1379,9 +1392,17 @@ class TestAdminPermissions:
@pytest.mark.asyncio
async def test_update_permission_empty_name(
self, client: httpx.AsyncClient, session_token: str
self, client: httpx.AsyncClient, session_token: str, test_db: DB
):
"""Updating permission with empty name should fail."""
# Create permission first
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:perm", display_name="Test Perm"
)
test_db.create_permission(perm)
response = await client.put(
"/auth/api/admin/permission?permission_id=test:perm&display_name=",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
@@ -1396,12 +1417,16 @@ class TestAdminPermissions:
):
"""Admin should be able to rename a permission."""
# Create permission first
perm = Permission(id="test:renameable2", display_name="Renameable")
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:renameable2", display_name="Renameable"
)
test_db.create_permission(perm)
response = await client.post(
"/auth/api/admin/permission/rename",
json={"old_id": "test:renameable2", "new_id": "test:renamed2"},
json={"old_scope": "test:renameable2", "new_scope": "test:renamed2"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1439,14 +1464,18 @@ class TestAdminPermissions:
self, client: httpx.AsyncClient, session_token: str, test_db: DB
):
"""Renaming permission can also update display name."""
perm = Permission(id="test:rename:withname", display_name="Old Name")
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:rename:withname", display_name="Old Name"
)
test_db.create_permission(perm)
response = await client.post(
"/auth/api/admin/permission/rename",
json={
"old_id": "test:rename:withname",
"new_id": "test:renamed:withname",
"old_scope": "test:rename:withname",
"new_scope": "test:renamed:withname",
"display_name": "New Display Name",
},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
@@ -1459,7 +1488,11 @@ class TestAdminPermissions:
):
"""Admin should be able to delete a permission."""
# Create permission first
perm = Permission(id="test:deleteable", display_name="Deleteable")
import uuid7
perm = Permission(
uuid=uuid7.create(), scope="test:deleteable", display_name="Deleteable"
)
test_db.create_permission(perm)
response = await client.delete(