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

API/DB cleanup for flat URLs that don't include org where users etc. are referred to. Implement user deletion in admin app and API, UI improvement. Reset token DB factory function revised to create passphrase and key internally. Removed unneeded functions and args, using update_user_role instead of a separate deleted _in_organization function.

Browse Source
This commit is contained in:
LeoVasanko
2026-02-05 13:57:07 +00:00
parent 632278d4ce
commit af5a48f565
12 changed files with 448 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/auth/admin/AdminApp.vue
+44 -11
View File
@@ -239,12 +239,44 @@ function deleteOrg(org) {
function createUserInRole(org, role) { openDialog('user-create', { org, role }) }
async function moveUserToRole(org, user, targetRoleDisplayName) {
if (user.role === targetRoleDisplayName) return
function deleteUser(user, userDetail) {
const credentialCount = userDetail?.credentials?.length || 0
const userUuid = user.uuid
const userName = user.display_name
const orgUuid = user.org // org UUID is stored in selectedUser
if (credentialCount === 0) {
// No credentials, safe to delete directly
performUserDeletion(userUuid, userName, orgUuid)
return
}
const passkeys = credentialCount === 1 ? '1 passkey' : `${credentialCount} passkeys`
openDialog('confirm', {
message: `Delete user "${userName}" with ${passkeys}? This action cannot be undone.`,
action: async () => {
await performUserDeletion(userUuid, userName, orgUuid)
}
})
}
async function performUserDeletion(userUuid, userName, orgUuid) {
try {
await apiJson(`/auth/api/admin/orgs/${org.uuid}/users/${user.uuid}/role`, {
await apiJson(`/auth/api/admin/users/${userUuid}`, { method: 'DELETE' })
authStore.showMessage(`User "${userName}" deleted.`, 'success', 2500)
await loadOrgs()
window.location.hash = `#org/${orgUuid}`
} catch (e) {
authStore.showMessage(e.message || 'Failed to delete user', 'error')
}
}
async function moveUserToRole(user, targetRoleUuid) {
if (user.role_uuid === targetRoleUuid) return
try {
await apiJson(`/auth/api/admin/users/${user.uuid}/role`, {
method: 'PATCH',
body: { role: targetRoleDisplayName }
body: { role_uuid: targetRoleUuid }
})
await loadOrgs()
} catch (e) {
@@ -268,7 +300,7 @@ function onRoleDrop(e, org, role) {
const data = JSON.parse(e.dataTransfer.getData('text/plain'))
if (data.org !== org.uuid) return // only within same org
const user = org.roles.flatMap(r => r.users).find(u => u.uuid === data.user_uuid)
if (user) moveUserToRole(org, user, role.display_name)
if (user) moveUserToRole(user, role.uuid)
} catch (_) { /* ignore */ }
}
@@ -279,7 +311,7 @@ function updateRole(role) { openDialog('role-update', { role, name: role.display
function deleteRole(role) {
// UI only allows deleting empty roles, so no confirmation needed
apiJson(`/auth/api/admin/orgs/${role.org}/roles/${role.uuid}`, { method: 'DELETE' })
apiJson(`/auth/api/admin/roles/${role.uuid}`, { method: 'DELETE' })
.then(() => {
authStore.showMessage(`Role "${role.display_name}" deleted.`, 'success', 2500)
loadOrgs()
@@ -299,7 +331,7 @@ async function toggleRolePermission(role, pid, checked) {
try {
const method = checked ? 'POST' : 'DELETE'
await apiJson(`/auth/api/admin/orgs/${role.org}/roles/${role.uuid}/permissions/${pid}`, {
await apiJson(`/auth/api/admin/roles/${role.uuid}/permissions/${pid}`, {
method
})
await loadOrgs()
@@ -406,7 +438,7 @@ const breadcrumbEntries = computed(() => {
watch(selectedUser, async (u) => {
if (!u) { userDetail.value = null; return }
try {
userDetail.value = await apiJson(`/auth/api/admin/orgs/${u.org}/users/${u.uuid}`)
userDetail.value = await apiJson(`/auth/api/admin/users/${u.uuid}`)
} catch (e) {
userDetail.value = { error: e.message }
}
@@ -542,7 +574,7 @@ async function refreshUserDetail() {
await loadOrgs()
if (selectedUser.value) {
try {
userDetail.value = await apiJson(`/auth/api/admin/orgs/${selectedUser.value.org}/users/${selectedUser.value.uuid}`)
userDetail.value = await apiJson(`/auth/api/admin/users/${selectedUser.value.uuid}`)
} catch (e) { authStore.showMessage(e.message || 'Failed to reload user', 'error') }
}
}
@@ -604,7 +636,7 @@ async function submitDialog() {
// Close dialog immediately, then perform async operation
closeDialog()
apiJson(`/auth/api/admin/orgs/${role.org}/roles/${role.uuid}`, { method: 'PATCH', body: { display_name: name } })
apiJson(`/auth/api/admin/roles/${role.uuid}`, { method: 'PATCH', body: { display_name: name } })
.then(() => {
authStore.showMessage(`Role renamed to "${name}".`, 'success', 2500)
loadOrgs()
@@ -632,7 +664,7 @@ async function submitDialog() {
// Close dialog immediately, then perform async operation
closeDialog()
apiJson(`/auth/api/admin/orgs/${user.org}/users/${user.uuid}/display-name`, { method: 'PATCH', body: { display_name: name } })
apiJson(`/auth/api/admin/users/${user.uuid}/display-name`, { method: 'PATCH', body: { display_name: name } })
.then(() => {
authStore.showMessage(`User renamed to "${name}".`, 'success', 2500)
onUserNameSaved()
@@ -771,6 +803,7 @@ async function submitDialog() {
@edit-user-name="editUserName"
@close-reg-modal="showRegModal = false"
@navigate-out="handlePanelNavigateOut"
@delete-user="deleteUser(selectedUser, userDetail)"
/>
<AdminOrgDetail
v-else-if="selectedOrg"
frontend/src/admin/AdminUserDetail.vue
+25 -20
View File
@@ -17,7 +17,7 @@ const props = defineProps({
navigationDisabled: { type: Boolean, default: false }
})
const emit = defineEmits(['generateUserRegistrationLink', 'goOverview', 'openOrg', 'onUserNameSaved', 'closeRegModal', 'editUserName', 'refreshUserDetail', 'navigateOut'])
const emit = defineEmits(['generateUserRegistrationLink', 'goOverview', 'openOrg', 'onUserNameSaved', 'closeRegModal', 'editUserName', 'refreshUserDetail', 'navigateOut', 'deleteUser'])
const authStore = useAuthStore()
const terminatingSessions = ref({})
@@ -45,7 +45,7 @@ function handleEditName() {
async function handleDelete(credential) {
try {
const data = await apiJson(`/auth/api/admin/orgs/${props.selectedUser.org}/users/${props.selectedUser.uuid}/credentials/${credential.credential}`, { method: 'DELETE' })
const data = await apiJson(`/auth/api/admin/users/${props.selectedUser.uuid}/credentials/${credential.credential}`, { method: 'DELETE' })
if (data.status === 'ok') {
emit('onUserNameSaved') // Reuse to refresh user detail
} else {
@@ -61,7 +61,7 @@ async function handleTerminateSession(session) {
if (!sessionId) return
terminatingSessions.value = { ...terminatingSessions.value, [sessionId]: true }
try {
const data = await apiJson(`/auth/api/admin/orgs/${props.selectedUser.org}/users/${props.selectedUser.uuid}/sessions/${sessionId}`, { method: 'DELETE' })
const data = await apiJson(`/auth/api/admin/users/${props.selectedUser.uuid}/sessions/${sessionId}`, { method: 'DELETE' })
if (data.status === 'ok') {
if (data.current_session_terminated) {
sessionStorage.clear()
@@ -83,6 +83,10 @@ async function handleTerminateSession(session) {
}
}
async function handleDeleteUser() {
emit('deleteUser')
}
// Handle user info section keynav
function handleUserInfoKeydown(event) {
if (hasActiveModal.value || props.navigationDisabled) return
@@ -183,24 +187,27 @@ defineExpose({ focusFirstElement })
:loading="loading"
:org-display-name="userDetail.org.display_name"
:role-name="userDetail.role"
:update-endpoint="`/auth/api/admin/orgs/${selectedUser.org}/users/${selectedUser.uuid}/display-name`"
:update-endpoint="`/auth/api/admin/users/${selectedUser.uuid}/display-name`"
@saved="$emit('onUserNameSaved')"
@edit-name="handleEditName"
/>
>
<div class="admin-actions">
<button
class="btn-primary"
@click="$emit('generateUserRegistrationLink', selectedUser)"
:disabled="loading"
>Registration Link</button>
<button
class="btn-danger"
@click="handleDeleteUser"
:disabled="loading"
title="Delete this user"
>Delete User</button>
</div>
</UserBasicInfo>
</div>
<div v-if="userDetail?.error" class="error small">{{ userDetail.error }}</div>
<template v-if="userDetail && !userDetail.error">
<div class="registration-actions" ref="regActionsRef" @keydown="handleRegActionsKeydown">
<button
class="btn-secondary reg-token-btn"
@click="$emit('generateUserRegistrationLink', selectedUser)"
:disabled="loading"
>Generate Registration Token</button>
<p class="matrix-hint muted">
Generate a one-time registration link so this user can register or add another passkey.
Copy the link from the dialog and send it to the user, or have the user scan the QR code on their device.
</p>
</div>
<section class="section-block" data-section="registered-passkeys">
<div class="section-header">
<h2>Registered Passkeys</h2>
@@ -238,7 +245,7 @@ defineExpose({ focusFirstElement })
</div>
<RegistrationLinkModal
v-if="showRegModal"
:endpoint="`/auth/api/admin/orgs/${selectedUser.org}/users/${selectedUser.uuid}/create-link`"
:endpoint="`/auth/api/admin/users/${selectedUser.uuid}/create-link`"
:user-name="userDetail?.display_name || selectedUser.display_name"
@close="$emit('closeRegModal')"
@copied="onLinkCopied"
@@ -248,13 +255,11 @@ defineExpose({ focusFirstElement })
<style scoped>
.user-detail { display: flex; flex-direction: column; gap: var(--space-lg); }
.admin-actions { display: flex; gap: 0.5rem; }
.actions { display: flex; flex-wrap: wrap; gap: var(--space-sm); align-items: center; }
.ancillary-actions { margin-top: -0.5rem; }
.reg-token-btn { align-self: flex-start; }
.registration-actions { display: flex; flex-direction: column; gap: 0.5rem; }
.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); }
.matrix-hint { font-size: 0.8rem; color: var(--color-text-muted); }
.error { color: var(--color-danger-text); }
.small { font-size: 0.9rem; }
.muted { color: var(--color-text-muted); }
paskia/bootstrap.py
+2 -4
View File
@@ -10,7 +10,7 @@ import asyncio
import logging
from paskia import authsession, db, globals
from paskia.util import hostutil, passphrase
from paskia.util import hostutil
logger = logging.getLogger(__name__)
@@ -74,11 +74,9 @@ async def check_admin_credentials() -> bool:
# Admin exists but has no credentials, create reset link
logger.info("⚠️ Admin user has no credentials!")
token = passphrase.generate()
expiry = authsession.reset_expires()
db.create_reset_token(
token = db.create_reset_token(
user_uuid=admin_user.uuid,
passphrase=token,
expiry=expiry,
token_type="admin registration",
)
paskia/db/__init__.py
-2
View File
@@ -63,7 +63,6 @@ from paskia.db.operations import (
update_session,
update_user_display_name,
update_user_role,
update_user_role_in_organization,
update_user_theme,
)
from paskia.db.structs import (
@@ -147,6 +146,5 @@ __all__ = [
"update_session",
"update_user_display_name",
"update_user_role",
"update_user_role_in_organization",
"update_user_theme",
]
paskia/db/operations.py
+22 -43
View File
@@ -31,7 +31,6 @@ from paskia.db.structs import (
SessionContext,
User,
)
from paskia.util.passphrase import generate as generate_passphrase
from paskia.util.passphrase import is_well_formed as _is_passphrase
_logger = logging.getLogger(__name__)
@@ -68,12 +67,12 @@ def get_user_organization(user_uuid: UUID) -> tuple[Org, str]:
Raises ValueError if user not found.
Call sites:
- update_user_role_in_organization: org only
- admin_create_user_registration_link: org only
- admin_get_user_detail: org and role
- admin_update_user_display_name: org only
- admin_delete_user_credential: org only
- admin_delete_user_session: org only
- admin_update_user_role: org only
"""
if user_uuid not in _db.users:
raise ValueError(f"User {user_uuid} not found")
@@ -384,29 +383,6 @@ def update_user_role(
_db.users[uuid].role_uuid = role_uuid
def update_user_role_in_organization(
user_uuid: UUID,
role_name: str,
*,
ctx: SessionContext | None = None,
) -> None:
"""Update user's role by role name within their current organization."""
if user_uuid not in _db.users:
raise ValueError(f"User {user_uuid} not found")
user = _db.users[user_uuid]
org = user.org
# Find role by name in the same org
new_role_uuid = None
for r in org.roles:
if r.display_name == role_name:
new_role_uuid = r.uuid
break
if new_role_uuid is None:
raise ValueError(f"Role '{role_name}' not found in organization")
with _db.transaction("admin:update_user_role", ctx):
_db.users[user_uuid].role_uuid = new_role_uuid
def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
"""Delete user and their credentials/sessions."""
if uuid not in _db.users:
@@ -419,9 +395,10 @@ def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
# Delete sessions
for sess in user.sessions:
del _db.sessions[sess.key]
# Delete reset tokens
for token in user.reset_tokens:
del _db.reset_tokens[token.key]
# Delete reset tokens (iterate over dict items to get correct keys)
for key, token in list(_db.reset_tokens.items()):
if token.user_uuid == uuid:
del _db.reset_tokens[key]
del _db.users[uuid]
@@ -567,31 +544,34 @@ def delete_sessions_for_user(
def create_reset_token(
passphrase: str,
user_uuid: UUID,
expiry: datetime,
token_type: str,
*,
ctx: SessionContext | None = None,
user: str | None = None,
) -> None:
"""Create a reset token from a passphrase.
) -> str:
"""Create a reset token and return the passphrase.
The acting user should be logged via ctx.
For self-service (user creating own recovery link), pass user's ctx.
For admin operations, pass admin's ctx.
For system operations (bootstrap), pass neither to log no user.
For API operations where ctx is not available but user is known, pass user.
Returns:
The passphrase to give to the user.
"""
key = _reset_key(passphrase)
if key in _db.reset_tokens:
raise ValueError("Reset token already exists")
if user_uuid not in _db.users:
raise ValueError(f"User {user_uuid} not found")
token, passphrase = ResetToken.create(
user=user_uuid, expiry=expiry, token_type=token_type
)
if token.key in _db.reset_tokens:
raise ValueError("Reset token already exists")
with _db.transaction("create_reset_token", ctx, user=user):
_db.reset_tokens[key] = ResetToken(
user_uuid=user_uuid, expiry=expiry, token_type=token_type
)
_db.reset_tokens[token.key] = token
return passphrase
def delete_reset_token(key: bytes, *, ctx: SessionContext | None = None) -> None:
@@ -781,14 +761,11 @@ def bootstrap(
role_uuid = uuid7.create()
user_uuid = uuid7.create()
# Generate reset token components
if reset_passphrase is None:
reset_passphrase = generate_passphrase()
# Set reset token expiry (passphrase generated by ResetToken.create)
if reset_expiry is None:
from paskia.authsession import reset_expires # noqa: PLC0415
reset_expiry = reset_expires()
reset_key = _reset_key(reset_passphrase)
now = datetime.now(UTC)
@@ -837,10 +814,12 @@ def bootstrap(
_db.users[user_uuid] = admin_user
# Create reset token
_db.reset_tokens[reset_key] = ResetToken(
user_uuid=user_uuid,
reset_token, reset_passphrase = ResetToken.create(
user=user_uuid,
expiry=reset_expiry,
token_type="admin bootstrap",
passphrase=reset_passphrase,
)
_db.reset_tokens[reset_token.key] = reset_token
return reset_passphrase
paskia/db/structs.py
+36
View File
@@ -353,6 +353,42 @@ class ResetToken(msgspec.Struct, dict=True):
"""Get the User object for this reset token."""
return db.data().users[self.user_uuid]
@classmethod
def create(
cls,
user: UUID | User,
expiry: datetime,
token_type: str,
passphrase: str | None = None,
) -> tuple[ResetToken, str]:
"""Create a new ResetToken with auto-generated or provided passphrase.
Args:
user: User UUID or User object
expiry: Token expiration datetime
token_type: Type of token (e.g., "device addition", "account recovery")
passphrase: Optional passphrase to use (auto-generated if not provided)
Returns:
Tuple of (token, passphrase) where passphrase is the human-readable
code to give to the user.
"""
import hashlib
from paskia.util.passphrase import generate as generate_passphrase
if passphrase is None:
passphrase = generate_passphrase()
key = hashlib.sha512(passphrase.encode()).digest()[:9]
user_uuid = user if isinstance(user, UUID) else user.uuid
token = cls(
user_uuid=user_uuid,
expiry=expiry,
token_type=token_type,
)
token.key = key
return token, passphrase
class SessionContext(msgspec.Struct):
session: Session
paskia/fastapi/admin.py
+89 -84
View File
@@ -17,7 +17,6 @@ from paskia.fastapi.session import AUTH_COOKIE
from paskia.globals import passkey
from paskia.util import (
hostutil,
passphrase,
permutil,
querysafe,
vitedev,
@@ -103,6 +102,7 @@ async def admin_list_orgs(request: Request, auth=AUTH_COOKIE):
"uuid": u.uuid,
"display_name": u.display_name,
"role": role_name,
"role_uuid": u.role_uuid,
"visits": u.visits,
"last_seen": u.last_seen,
}
@@ -281,28 +281,27 @@ async def admin_create_role(
return {"uuid": str(role.uuid)}
@app.patch("/orgs/{org_uuid}/roles/{role_uuid}")
@app.patch("/roles/{role_uuid}")
async def admin_update_role_name(
org_uuid: UUID,
role_uuid: UUID,
request: Request,
payload: dict = Body(...),
auth=AUTH_COOKIE,
):
"""Update role display name only."""
role = db.data().roles.get(role_uuid)
if not role:
raise HTTPException(status_code=404, detail="Role not found")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, role.org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
role = db.data().roles.get(role_uuid)
if not role or role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization")
display_name = payload.get("display_name")
if not display_name:
@@ -312,68 +311,64 @@ async def admin_update_role_name(
return {"status": "ok"}
@app.post("/orgs/{org_uuid}/roles/{role_uuid}/permissions/{permission_uuid}")
@app.post("/roles/{role_uuid}/permissions/{permission_uuid}")
async def admin_add_role_permission(
org_uuid: UUID,
role_uuid: UUID,
permission_uuid: UUID,
request: Request,
auth=AUTH_COOKIE,
):
"""Add a permission to a role (intent-based API)."""
role = db.data().roles.get(role_uuid)
if not role:
raise HTTPException(status_code=404, detail="Role not found")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, role.org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
role = db.data().roles.get(role_uuid)
if not role or role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization")
# Verify permission exists and org can grant it
perm = db.data().permissions.get(permission_uuid)
if not perm:
raise HTTPException(status_code=404, detail="Permission not found")
if org_uuid not in perm.orgs:
if role.org_uuid not in perm.orgs:
raise ValueError("Permission not grantable by organization")
db.add_permission_to_role(role_uuid, permission_uuid, ctx=ctx)
return {"status": "ok"}
@app.delete("/orgs/{org_uuid}/roles/{role_uuid}/permissions/{permission_uuid}")
@app.delete("/roles/{role_uuid}/permissions/{permission_uuid}")
async def admin_remove_role_permission(
org_uuid: UUID,
role_uuid: UUID,
permission_uuid: UUID,
request: Request,
auth=AUTH_COOKIE,
):
"""Remove a permission from a role (intent-based API)."""
role = db.data().roles.get(role_uuid)
if not role:
raise HTTPException(status_code=404, detail="Role not found")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, role.org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
role = db.data().roles.get(role_uuid)
if not role or role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization")
# Sanity check: prevent admin from removing their own access
perm = db.data().permissions.get(permission_uuid)
if ctx.org.uuid == org_uuid and ctx.role.uuid == role_uuid:
if ctx.org.uuid == role.org_uuid and ctx.role.uuid == role_uuid:
if perm and perm.scope in ["auth:admin", "auth:org:admin"]:
# Check if removing this permission would leave no admin access
remaining_perms = role.permission_set - {permission_uuid}
@@ -390,13 +385,15 @@ async def admin_remove_role_permission(
return {"status": "ok"}
@app.delete("/orgs/{org_uuid}/roles/{role_uuid}")
@app.delete("/roles/{role_uuid}")
async def admin_delete_role(
org_uuid: UUID,
role_uuid: UUID,
request: Request,
auth=AUTH_COOKIE,
):
role = db.data().roles.get(role_uuid)
if not role:
raise HTTPException(status_code=404, detail="Role not found")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
@@ -404,13 +401,10 @@ async def admin_delete_role(
host=request.headers.get("host"),
max_age="5m",
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, role.org_uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
role = db.data().roles.get(role_uuid)
if not role or role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization")
# Sanity check: prevent admin from deleting their own role
if ctx.role.uuid == role_uuid:
@@ -460,60 +454,58 @@ async def admin_create_user(
return {"uuid": str(user.uuid)}
@app.patch("/orgs/{org_uuid}/users/{user_uuid}/role")
@app.patch("/users/{user_uuid}/role")
async def admin_update_user_role(
org_uuid: UUID,
user_uuid: UUID,
request: Request,
payload: dict = Body(...),
auth=AUTH_COOKIE,
):
try:
user_org, _current_role = db.get_user_organization(user_uuid)
except ValueError:
raise HTTPException(status_code=404, detail="User not found")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, user_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")
role_uuid_str = payload.get("role_uuid")
if not role_uuid_str:
raise ValueError("role_uuid is required")
try:
user_org, _current_role = db.get_user_organization(user_uuid)
except ValueError:
raise ValueError("User not found")
if user_org.uuid != org_uuid:
raise ValueError("User does not belong to this organization")
roles = user_org.roles
if not any(r.display_name == new_role for r in roles):
new_role_uuid = UUID(role_uuid_str)
except (ValueError, TypeError):
raise ValueError("Invalid role UUID")
new_role = db.data().roles.get(new_role_uuid)
if not new_role or new_role.org_uuid != user_org.uuid:
raise ValueError("Role not found in organization")
# Sanity check: prevent admin from removing their own access
if ctx.user.uuid == user_uuid:
new_role_obj = next((r for r in roles if r.display_name == new_role), None)
if new_role_obj: # pragma: no branch - always true, role validated above
# Check if any permission in the new role is an admin permission
has_admin_access = False
for perm_uuid in new_role_obj.permissions:
perm = db.data().permissions.get(perm_uuid)
if perm and perm.scope in ["auth:admin", "auth:org:admin"]:
has_admin_access = True
break
if not has_admin_access:
raise ValueError(
"Cannot change your own role to one without admin permissions"
)
# Check if any permission in the new role is an admin permission
has_admin_access = False
for perm_uuid in new_role.permissions:
perm = db.data().permissions.get(perm_uuid)
if perm and perm.scope in ["auth:admin", "auth:org:admin"]:
has_admin_access = True
break
if not has_admin_access:
raise ValueError(
"Cannot change your own role to one without admin permissions"
)
db.update_user_role_in_organization(user_uuid, new_role, ctx=ctx)
db.update_user_role(user_uuid, new_role_uuid, ctx=ctx)
return {"status": "ok"}
@app.post("/orgs/{org_uuid}/users/{user_uuid}/create-link")
@app.post("/users/{user_uuid}/create-link")
async def admin_create_user_registration_link(
org_uuid: UUID,
user_uuid: UUID,
request: Request,
auth=AUTH_COOKIE,
@@ -522,8 +514,6 @@ async def admin_create_user_registration_link(
user_org, _role_name = db.get_user_organization(user_uuid)
except ValueError:
raise HTTPException(status_code=404, detail="User not found")
if user_org.uuid != org_uuid:
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
@@ -531,7 +521,7 @@ async def admin_create_user_registration_link(
host=request.headers.get("host"),
max_age="5m",
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, user_org.uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -540,11 +530,9 @@ async def admin_create_user_registration_link(
has_credentials = db.get_user_credential_ids(user_uuid)
token_type = "user registration" if not has_credentials else "account recovery"
token = passphrase.generate()
expiry = reset_expires()
db.create_reset_token(
token = db.create_reset_token(
user_uuid=user_uuid,
passphrase=token,
expiry=expiry,
token_type=token_type,
ctx=ctx,
@@ -556,9 +544,8 @@ async def admin_create_user_registration_link(
}
@app.get("/orgs/{org_uuid}/users/{user_uuid}")
@app.get("/users/{user_uuid}")
async def admin_get_user_detail(
org_uuid: UUID,
user_uuid: UUID,
request: Request,
auth=AUTH_COOKIE,
@@ -567,15 +554,13 @@ async def admin_get_user_detail(
user_org, role_name = db.get_user_organization(user_uuid)
except ValueError:
raise HTTPException(status_code=404, detail="User not found")
if user_org.uuid != org_uuid:
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, user_org.uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -615,9 +600,8 @@ async def admin_get_user_detail(
)
@app.patch("/orgs/{org_uuid}/users/{user_uuid}/display-name")
@app.patch("/users/{user_uuid}/display-name")
async def admin_update_user_display_name(
org_uuid: UUID,
user_uuid: UUID,
request: Request,
payload: dict = Body(...),
@@ -627,15 +611,13 @@ async def admin_update_user_display_name(
user_org, _role_name = db.get_user_organization(user_uuid)
except ValueError:
raise HTTPException(status_code=404, detail="User not found")
if user_org.uuid != org_uuid:
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, user_org.uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -648,9 +630,37 @@ async def admin_update_user_display_name(
return {"status": "ok"}
@app.delete("/orgs/{org_uuid}/users/{user_uuid}/credentials/{credential_uuid}")
@app.delete("/users/{user_uuid}")
async def admin_delete_user(
user_uuid: UUID,
request: Request,
auth=AUTH_COOKIE,
):
"""Delete a user and all their credentials/sessions."""
try:
user_org, _role_name = db.get_user_organization(user_uuid)
except ValueError:
raise HTTPException(status_code=404, detail="User not found")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
max_age="5m",
)
if not can_manage_org(ctx, user_org.uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
# Prevent admin from deleting themselves
if ctx.user.uuid == user_uuid:
raise ValueError("Cannot delete your own account")
db.delete_user(user_uuid, ctx=ctx)
return {"status": "ok"}
@app.delete("/users/{user_uuid}/credentials/{credential_uuid}")
async def admin_delete_user_credential(
org_uuid: UUID,
user_uuid: UUID,
credential_uuid: UUID,
request: Request,
@@ -660,8 +670,6 @@ async def admin_delete_user_credential(
user_org, _role_name = db.get_user_organization(user_uuid)
except ValueError:
raise HTTPException(status_code=404, detail="User not found")
if user_org.uuid != org_uuid:
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
@@ -669,7 +677,7 @@ async def admin_delete_user_credential(
host=request.headers.get("host"),
max_age="5m",
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, user_org.uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
@@ -677,9 +685,8 @@ async def admin_delete_user_credential(
return {"status": "ok"}
@app.delete("/orgs/{org_uuid}/users/{user_uuid}/sessions/{session_id}")
@app.delete("/users/{user_uuid}/sessions/{session_id}")
async def admin_delete_user_session(
org_uuid: UUID,
user_uuid: UUID,
session_id: str,
request: Request,
@@ -689,15 +696,13 @@ async def admin_delete_user_session(
user_org, _role_name = db.get_user_organization(user_uuid)
except ValueError:
raise HTTPException(status_code=404, detail="User not found")
if user_org.uuid != org_uuid:
raise HTTPException(status_code=404, detail="User not found in organization")
ctx = await authz.verify(
auth,
["auth:admin", "auth:org:admin"],
match=permutil.has_any,
host=request.headers.get("host"),
)
if not can_manage_org(ctx, org_uuid):
if not can_manage_org(ctx, user_org.uuid):
raise authz.AuthException(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
paskia/fastapi/remote.py
+2 -5
View File
@@ -20,7 +20,7 @@ from paskia.authsession import expires
from paskia.fastapi.session import AUTH_COOKIE, infodict
from paskia.fastapi.wschat import authenticate_and_login
from paskia.fastapi.wsutil import validate_origin, websocket_error_handler
from paskia.util import passphrase, pow, useragent
from paskia.util import pow, useragent
# Create a FastAPI subapp for remote auth WebSocket endpoints
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
@@ -317,16 +317,13 @@ async def websocket_remote_auth_permit(ws: WebSocket, auth=AUTH_COOKIE):
if request.action == "register":
# For registration, create a reset token for device addition
token_str = passphrase.generate()
expiry = expires()
db.create_reset_token(
reset_token = db.create_reset_token(
user_uuid=ctx.user.uuid,
passphrase=token_str,
expiry=expiry,
token_type="device addition",
user=str(ctx.user.uuid),
)
reset_token = token_str
# Complete the remote auth request (notifies the waiting device)
cred = db.data().credentials[ctx.session.credential_uuid]
paskia/fastapi/reset.py
+2 -4
View File
@@ -15,7 +15,7 @@ from uuid import UUID
from paskia import authsession as _authsession
from paskia import db
from paskia.util import hostutil, passphrase
from paskia.util import hostutil
async def _resolve_targets(query: str | None):
@@ -69,10 +69,8 @@ async def _resolve_targets(query: str | None):
async def _create_reset(user, role_name: str):
token = passphrase.generate()
expiry = _authsession.reset_expires()
db.create_reset_token(
passphrase=token,
token = db.create_reset_token(
user_uuid=user.uuid,
expiry=expiry,
token_type="manual reset",
paskia/fastapi/user.py
+2 -4
View File
@@ -17,7 +17,7 @@ from paskia.authsession import (
)
from paskia.fastapi import authz, session
from paskia.fastapi.session import AUTH_COOKIE
from paskia.util import hostutil, passphrase
from paskia.util import hostutil
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
@@ -149,11 +149,9 @@ async def api_create_link(
):
# Require recent authentication for sensitive operation
ctx = await authz.verify(auth, [], host=request.headers.get("host"), max_age="5m")
token = passphrase.generate()
expiry = expires()
db.create_reset_token(
token = db.create_reset_token(
user_uuid=ctx.user.uuid,
passphrase=token,
expiry=expiry,
token_type="device addition",
ctx=ctx,
paskia/migrate/__init__.py
+3 -22
View File
@@ -28,7 +28,6 @@ from paskia.db.structs import (
Credential,
Org,
Permission,
ResetToken,
Role,
Session,
User,
@@ -39,7 +38,6 @@ from .sql import (
)
from .sql import (
CredentialModel,
ResetTokenModel,
SessionModel,
UserModel,
)
@@ -226,26 +224,9 @@ async def migrate_from_sql(
)
print(f" Migrated {len(session_models)} sessions")
# Migrate reset tokens
# Old format: b"rset" + 16 bytes hash -> New format: 9 bytes (truncated hash)
async with sql_db.session() as session:
result = await session.execute(select(ResetTokenModel))
token_models = result.scalars().all()
for tm in token_models:
token = tm.as_dataclass()
old_key: bytes = token.key
# Strip b"rset" prefix and take first 9 bytes of hash
if old_key.startswith(b"rset"):
token_key = old_key[4:13] # 9 bytes after prefix
else:
# Already in new format or unknown - truncate to 9 bytes
token_key = old_key[:9]
db.reset_tokens[token_key] = ResetToken(
user_uuid=token.user_uuid,
expiry=token.expiry,
token_type=token.token_type,
)
print(f" Migrated {len(token_models)} reset tokens")
# Reset tokens are not migrated - they will expire naturally
# and users can generate new ones as needed
print(" Reset tokens dropped (not migrated)")
# Queue and flush all changes using the transaction mechanism
with db.transaction("migrate:sql"):
tests/test_admin.py
+221 -93
View File
@@ -604,7 +604,7 @@ class TestAdminRoles:
):
"""Admin should be able to update a role."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}",
f"/auth/api/admin/roles/{test_role.uuid}",
json={"display_name": "Updated Role Name"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -614,17 +614,19 @@ class TestAdminRoles:
@pytest.mark.asyncio
async def test_update_role_wrong_org(
self, client: httpx.AsyncClient, session_token: str, test_org, second_org_role
self,
client: httpx.AsyncClient,
org_admin_session_token: str,
test_org,
second_org_role,
):
"""Cannot update role from another org."""
"""Org admin cannot update role from another org."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{second_org_role.uuid}",
f"/auth/api/admin/roles/{second_org_role.uuid}",
json={"display_name": "Try Update Wrong Org"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
data = response.json()
assert "Role not found" in data["detail"]
assert response.status_code == 403
@pytest.mark.asyncio
async def test_update_role_add_grantable_permission(
@@ -637,7 +639,7 @@ class TestAdminRoles:
):
"""Admin should be able to add grantable permissions to role."""
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}/permissions/{grantable_permission.uuid}",
f"/auth/api/admin/roles/{user_role.uuid}/permissions/{grantable_permission.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -659,7 +661,7 @@ class TestAdminRoles:
create_permission(perm)
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}/permissions/{perm.uuid}",
f"/auth/api/admin/roles/{user_role.uuid}/permissions/{perm.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
@@ -680,14 +682,14 @@ class TestAdminRoles:
# test_role has both auth:admin and auth:org:admin
# Remove auth:admin first (should succeed since org:admin remains)
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}/permissions/{admin_permission.uuid}",
f"/auth/api/admin/roles/{test_role.uuid}/permissions/{admin_permission.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
# Now try to remove auth:org:admin (should fail - would leave no admin access)
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}/permissions/{org_admin_permission.uuid}",
f"/auth/api/admin/roles/{test_role.uuid}/permissions/{org_admin_permission.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
@@ -700,7 +702,7 @@ class TestAdminRoles:
):
"""Admin should be able to delete a role."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}",
f"/auth/api/admin/roles/{user_role.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -709,16 +711,18 @@ class TestAdminRoles:
@pytest.mark.asyncio
async def test_delete_role_wrong_org(
self, client: httpx.AsyncClient, session_token: str, test_org, second_org_role
self,
client: httpx.AsyncClient,
org_admin_session_token: str,
test_org,
second_org_role,
):
"""Cannot delete role from another org."""
"""Org admin cannot delete role from another org."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{second_org_role.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
f"/auth/api/admin/roles/{second_org_role.uuid}",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
data = response.json()
assert "Role not found" in data["detail"]
assert response.status_code == 403
@pytest.mark.asyncio
async def test_delete_own_role_fails(
@@ -726,7 +730,7 @@ class TestAdminRoles:
):
"""Admin cannot delete their own role."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}",
f"/auth/api/admin/roles/{test_role.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
@@ -788,7 +792,7 @@ class TestAdminUsersInOrg:
):
"""Admin should be able to get user details within an org."""
response = await client.get(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}",
f"/auth/api/admin/users/{test_user.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -807,7 +811,7 @@ class TestAdminUsersInOrg:
"""Getting non-existent user should return 404."""
fake_uuid = uuid7.create()
response = await client.get(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{fake_uuid}",
f"/auth/api/admin/users/{fake_uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
@@ -816,16 +820,18 @@ class TestAdminUsersInOrg:
@pytest.mark.asyncio
async def test_get_user_wrong_org(
self, client: httpx.AsyncClient, session_token: str, test_org, second_org_user
self,
client: httpx.AsyncClient,
org_admin_session_token: str,
test_org,
second_org_user,
):
"""Getting user from another org should return 404."""
"""Org admin cannot get user from another org."""
response = await client.get(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
f"/auth/api/admin/users/{second_org_user.uuid}",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
data = response.json()
assert "User not found" in data["detail"]
assert response.status_code == 403
@pytest.mark.asyncio
async def test_get_user_with_org_admin(
@@ -837,7 +843,7 @@ class TestAdminUsersInOrg:
):
"""Org admin should be able to get user details."""
response = await client.get(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{org_admin_user.uuid}",
f"/auth/api/admin/users/{org_admin_user.uuid}",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -850,7 +856,7 @@ class TestAdminUsersInOrg:
):
"""Admin should be able to update user display name."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/display-name",
f"/auth/api/admin/users/{test_user.uuid}/display-name",
json={"display_name": "Updated Admin Name"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -863,7 +869,7 @@ class TestAdminUsersInOrg:
"""Updating non-existent user should return 404."""
fake_uuid = uuid7.create()
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{fake_uuid}/display-name",
f"/auth/api/admin/users/{fake_uuid}/display-name",
json={"display_name": "New Name"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -873,15 +879,19 @@ class TestAdminUsersInOrg:
@pytest.mark.asyncio
async def test_update_user_display_name_wrong_org(
self, client: httpx.AsyncClient, session_token: str, test_org, second_org_user
self,
client: httpx.AsyncClient,
org_admin_session_token: str,
test_org,
second_org_user,
):
"""Updating user from another org should return 404."""
"""Org admin cannot update user from another org."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/display-name",
f"/auth/api/admin/users/{second_org_user.uuid}/display-name",
json={"display_name": "New Name"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
assert response.status_code == 403
@pytest.mark.asyncio
async def test_update_user_display_name_empty(
@@ -889,7 +899,7 @@ class TestAdminUsersInOrg:
):
"""Updating user with empty display name should fail."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/display-name",
f"/auth/api/admin/users/{test_user.uuid}/display-name",
json={"display_name": " "},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -903,7 +913,7 @@ class TestAdminUsersInOrg:
):
"""Updating user with too long display name should fail."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/display-name",
f"/auth/api/admin/users/{test_user.uuid}/display-name",
json={"display_name": "x" * 100},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -923,8 +933,8 @@ class TestAdminUsersInOrg:
"""Admin should be able to change user's role within org."""
# Use regular_user who is in the same org but not the session owner
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{regular_user.uuid}/role",
json={"role": user_role.display_name},
f"/auth/api/admin/users/{regular_user.uuid}/role",
json={"role_uuid": str(user_role.uuid)},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -935,13 +945,13 @@ class TestAdminUsersInOrg:
):
"""Updating user role without specifying role should fail."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/role",
f"/auth/api/admin/users/{test_user.uuid}/role",
json={},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
data = response.json()
assert "role is required" in data["detail"]
assert "role_uuid is required" in data["detail"]
@pytest.mark.asyncio
async def test_update_user_role_user_not_found(
@@ -950,27 +960,29 @@ class TestAdminUsersInOrg:
"""Updating role for non-existent user should fail."""
fake_uuid = uuid7.create()
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{fake_uuid}/role",
json={"role": "User Role"},
f"/auth/api/admin/users/{fake_uuid}/role",
json={"role_uuid": str(uuid7.create())},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
assert response.status_code == 404
data = response.json()
assert "User not found" in data["detail"]
@pytest.mark.asyncio
async def test_update_user_role_wrong_org(
self, client: httpx.AsyncClient, session_token: str, test_org, second_org_user
self,
client: httpx.AsyncClient,
org_admin_session_token: str,
test_org,
second_org_user,
):
"""Updating role for user in another org should fail."""
"""Org admin cannot update role for user in another org."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/role",
json={"role": "User Role"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
f"/auth/api/admin/users/{second_org_user.uuid}/role",
json={"role_uuid": str(uuid7.create())},
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
data = response.json()
assert "does not belong" in data["detail"]
assert response.status_code == 403
@pytest.mark.asyncio
async def test_update_user_role_invalid_role(
@@ -978,8 +990,8 @@ class TestAdminUsersInOrg:
):
"""Updating user to non-existent role should fail."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/role",
json={"role": "Nonexistent Role"},
f"/auth/api/admin/users/{test_user.uuid}/role",
json={"role_uuid": str(uuid7.create())},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
@@ -997,8 +1009,8 @@ class TestAdminUsersInOrg:
):
"""Admin cannot change their own role to non-admin role."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{org_admin_user.uuid}/role",
json={"role": user_role.display_name},
f"/auth/api/admin/users/{org_admin_user.uuid}/role",
json={"role_uuid": str(user_role.uuid)},
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
@@ -1018,8 +1030,8 @@ class TestAdminUsersInOrg:
# test_user is already on test_role which has auth:admin
# Changing to the same role should succeed (no permission loss)
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/role",
json={"role": test_role.display_name},
f"/auth/api/admin/users/{test_user.uuid}/role",
json={"role_uuid": str(test_role.uuid)},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1030,7 +1042,7 @@ class TestAdminUsersInOrg:
):
"""Admin should be able to create reset links for users."""
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/create-link",
f"/auth/api/admin/users/{test_user.uuid}/create-link",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1045,7 +1057,7 @@ class TestAdminUsersInOrg:
"""Creating reset link for non-existent user should fail."""
fake_uuid = uuid7.create()
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{fake_uuid}/create-link",
f"/auth/api/admin/users/{fake_uuid}/create-link",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
@@ -1054,16 +1066,18 @@ class TestAdminUsersInOrg:
@pytest.mark.asyncio
async def test_create_user_reset_link_wrong_org(
self, client: httpx.AsyncClient, session_token: str, test_org, second_org_user
self,
client: httpx.AsyncClient,
org_admin_session_token: str,
test_org,
second_org_user,
):
"""Creating reset link for user in another org should fail."""
"""Org admin cannot create reset link for user in another org."""
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/create-link",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
f"/auth/api/admin/users/{second_org_user.uuid}/create-link",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
data = response.json()
assert "not found in organization" in data["detail"]
assert response.status_code == 403
@pytest.mark.asyncio
async def test_create_user_registration_link_without_credentials(
@@ -1083,7 +1097,7 @@ class TestAdminUsersInOrg:
create_user(user_no_cred)
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{user_no_cred.uuid}/create-link",
f"/auth/api/admin/users/{user_no_cred.uuid}/create-link",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1091,6 +1105,120 @@ class TestAdminUsersInOrg:
assert "url" in data
# -------------------- User Deletion Tests --------------------
class TestAdminUserDeletion:
"""Tests for admin user deletion"""
@pytest.mark.asyncio
async def test_delete_user_success(
self,
client: httpx.AsyncClient,
session_token: str,
test_org,
user_role,
test_db: DB,
):
"""Admin should be able to delete a user."""
# Create a user to delete
user_to_delete = User.create(
display_name="User To Delete",
role=user_role.uuid,
)
create_user(user_to_delete)
response = await client.delete(
f"/auth/api/admin/users/{user_to_delete.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "ok"
# Verify user is actually deleted
assert user_to_delete.uuid not in db.data().users
@pytest.mark.asyncio
async def test_delete_user_not_found(
self, client: httpx.AsyncClient, session_token: str
):
"""Deleting non-existent user should return 404."""
fake_uuid = uuid7.create()
response = await client.delete(
f"/auth/api/admin/users/{fake_uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
data = response.json()
assert "User not found" in data["detail"]
@pytest.mark.asyncio
async def test_delete_own_user_fails(
self, client: httpx.AsyncClient, session_token: str, test_user
):
"""Admin cannot delete their own account."""
response = await client.delete(
f"/auth/api/admin/users/{test_user.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
data = response.json()
assert "Cannot delete your own account" in data["detail"]
@pytest.mark.asyncio
async def test_delete_user_wrong_org(
self,
client: httpx.AsyncClient,
org_admin_session_token: str,
second_org_user,
):
"""Org admin cannot delete user from another org."""
response = await client.delete(
f"/auth/api/admin/users/{second_org_user.uuid}",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_delete_user_org_admin_success(
self,
client: httpx.AsyncClient,
org_admin_session_token: str,
test_org,
user_role,
test_db: DB,
):
"""Org admin should be able to delete users in their org."""
# Create a user in the same org to delete
user_to_delete = User.create(
display_name="Org User To Delete",
role=user_role.uuid,
)
create_user(user_to_delete)
response = await client.delete(
f"/auth/api/admin/users/{user_to_delete.uuid}",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "ok"
@pytest.mark.asyncio
async def test_delete_user_regular_user_forbidden(
self,
client: httpx.AsyncClient,
regular_session_token: str,
test_user,
):
"""Regular user trying to delete user should get 403."""
response = await client.delete(
f"/auth/api/admin/users/{test_user.uuid}",
headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 403
# -------------------- Credential Tests --------------------
@@ -1108,7 +1236,7 @@ class TestAdminCredentials:
):
"""Admin should be able to delete a user's credential."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/credentials/{test_credential.uuid}",
f"/auth/api/admin/users/{test_user.uuid}/credentials/{test_credential.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1123,7 +1251,7 @@ class TestAdminCredentials:
fake_user_uuid = uuid7.create()
fake_cred_uuid = uuid7.create()
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{fake_user_uuid}/credentials/{fake_cred_uuid}",
f"/auth/api/admin/users/{fake_user_uuid}/credentials/{fake_cred_uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
@@ -1134,17 +1262,17 @@ class TestAdminCredentials:
async def test_delete_credential_wrong_org(
self,
client: httpx.AsyncClient,
session_token: str,
org_admin_session_token: str,
test_org,
second_org_user,
second_org_credential,
):
"""Deleting credential for user in another org should fail."""
"""Org admin cannot delete credential for user in another org."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/credentials/{second_org_credential.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
f"/auth/api/admin/users/{second_org_user.uuid}/credentials/{second_org_credential.uuid}",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
assert response.status_code == 403
# -------------------- Session Tests --------------------
@@ -1175,7 +1303,7 @@ class TestAdminSessions:
)
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/{extra_token}",
f"/auth/api/admin/users/{test_user.uuid}/sessions/{extra_token}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1193,7 +1321,7 @@ class TestAdminSessions:
):
"""Admin can delete their own current session."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/{session_token}",
f"/auth/api/admin/users/{test_user.uuid}/sessions/{session_token}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1207,7 +1335,7 @@ class TestAdminSessions:
"""Deleting session for non-existent user should fail."""
fake_uuid = uuid7.create()
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{fake_uuid}/sessions/fake-session-id",
f"/auth/api/admin/users/{fake_uuid}/sessions/fake-session-id",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
@@ -1218,16 +1346,16 @@ class TestAdminSessions:
async def test_delete_session_wrong_org(
self,
client: httpx.AsyncClient,
session_token: str,
org_admin_session_token: str,
test_org,
second_org_user,
):
"""Deleting session for user in another org should fail."""
"""Org admin cannot delete session for user in another org."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/sessions/fake-session",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
f"/auth/api/admin/users/{second_org_user.uuid}/sessions/fake-session",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
assert response.status_code == 403
@pytest.mark.asyncio
async def test_delete_session_invalid_id(
@@ -1235,7 +1363,7 @@ class TestAdminSessions:
):
"""Deleting session with invalid/non-existent ID should fail."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/invalid!!id",
f"/auth/api/admin/users/{test_user.uuid}/sessions/invalid!!id",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
@@ -1250,7 +1378,7 @@ class TestAdminSessions:
# Use a valid format but non-existent key
fake_token = secrets.token_urlsafe(12)
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/{fake_token}",
f"/auth/api/admin/users/{test_user.uuid}/sessions/{fake_token}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 404
@@ -1550,7 +1678,7 @@ class TestOrgAdminAuthExceptions:
):
"""Regular user (not org admin) trying to create reset link should get 403."""
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/create-link",
f"/auth/api/admin/users/{test_user.uuid}/create-link",
headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 403
@@ -1565,7 +1693,7 @@ class TestOrgAdminAuthExceptions:
):
"""Regular user trying to get user details should get 403."""
response = await client.get(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}",
f"/auth/api/admin/users/{test_user.uuid}",
headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 403
@@ -1580,7 +1708,7 @@ class TestOrgAdminAuthExceptions:
):
"""Regular user trying to update display name should get 403."""
response = await client.patch(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/display-name",
f"/auth/api/admin/users/{test_user.uuid}/display-name",
json={"display_name": "New Name"},
headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
)
@@ -1597,7 +1725,7 @@ class TestOrgAdminAuthExceptions:
):
"""Regular user trying to delete credential should get 403."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/credentials/{test_credential.uuid}",
f"/auth/api/admin/users/{test_user.uuid}/credentials/{test_credential.uuid}",
headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 403
@@ -1612,7 +1740,7 @@ class TestOrgAdminAuthExceptions:
):
"""Regular user trying to delete session should get 403."""
response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/some-session",
f"/auth/api/admin/users/{test_user.uuid}/sessions/some-session",
headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
)
assert response.status_code == 403