diff --git a/frontend/auth/admin/AdminApp.vue b/frontend/auth/admin/AdminApp.vue index bca60ee..c3cc961 100644 --- a/frontend/auth/admin/AdminApp.vue +++ b/frontend/auth/admin/AdminApp.vue @@ -11,9 +11,10 @@ import AdminOrgDetail from '@/admin/AdminOrgDetail.vue' import AdminUserDetail from '@/admin/AdminUserDetail.vue' import AdminDialogs from '@/admin/AdminDialogs.vue' import { useAuthStore } from '@/stores/auth' -import { getSettings, adminUiPath, makeUiHref } from '@/utils/settings' +import { adminUiPath, makeUiHref } from '@/utils/settings' import { apiJson } from '@/utils/api' import { getDirection } from '@/utils/keynav' +import { goBack } from '@/utils/helpers' const info = ref(null) const loading = ref(true) @@ -64,8 +65,8 @@ function handleGlobalClick(e) { onMounted(async () => { document.addEventListener('click', handleGlobalClick) window.addEventListener('hashchange', parseHash) - const settings = await getSettings() - if (settings?.rp_name) document.title = settings.rp_name + ' Admin' + await authStore.loadSettings() + if (authStore.settings?.rp_name) document.title = authStore.settings.rp_name + ' Admin' await load() }) @@ -418,7 +419,7 @@ async function toggleOrgPermission(org, permId, checked) { await apiJson(`/auth/api/admin/orgs/${org.uuid}/permission?${params.toString()}`, { method: checked ? 'POST' : 'DELETE' }) await loadOrgs() } catch (e) { - authStore.showMessage(e.message || 'Failed to update organization permission') + authStore.showMessage(e.message || 'Failed to update organization permission', 'error') org.permissions = prev // revert } } @@ -680,7 +681,17 @@ async function submitDialog() { }) return // Don't call closeDialog() again } else if (t === 'confirm') { - const action = dialog.value.data.action; if (action) await action() + const action = dialog.value.data.action + // Close dialog first, then perform action (errors shown via showMessage) + closeDialog() + if (action) { + try { + await action() + } catch (e) { + authStore.showMessage(e.message || 'Action failed', 'error') + } + } + return // Already closed } closeDialog() } catch (e) { @@ -698,6 +709,18 @@ async function submitDialog() { v-else-if="showBackMessage" @reload="reloadPage" /> + +
{{ error }}
+You do not have admin permissions for this application.
+ +{{ dialog.data.message }}
diff --git a/frontend/src/stores/auth.js b/frontend/src/stores/auth.js index 1d28199..1a174e6 100644 --- a/frontend/src/stores/auth.js +++ b/frontend/src/stores/auth.js @@ -26,16 +26,18 @@ export const useAuthStore = defineStore('auth', { setLoading(flag) { this.isLoading = !!flag }, - showMessage(message, type = 'info', duration = 3000) { + showMessage(message, type = 'info', duration = null) { + // Default duration: 5 seconds for errors, 3 seconds for others + const effectiveDuration = duration ?? (type === 'error' ? 5000 : 3000) this.status = { message, type, show: true } - if (duration > 0) { + if (effectiveDuration > 0) { setTimeout(() => { this.status.show = false - }, duration) + }, effectiveDuration) } }, async setSessionCookie(result) { diff --git a/paskia/bootstrap.py b/paskia/bootstrap.py index c9f36cd..041e784 100644 --- a/paskia/bootstrap.py +++ b/paskia/bootstrap.py @@ -66,20 +66,25 @@ async def bootstrap_system() -> dict: ) db.create_permission(perm0) + # Create org admin permission - allows managing users within an org + perm_org_admin = Permission( + uuid=uuid7.create(), scope="auth:org:admin", display_name="Org Admin" + ) + db.create_permission(perm_org_admin) + org = Org(uuid7.create(), "Organization") db.create_organization(org) - # After creation, org.permissions now includes the auto-created org admin permission (auth:org:admin) - # Allow this org to grant global admin explicitly + # Allow this org to grant global admin and org admin permissions db.add_permission_to_organization(str(org.uuid), perm0.scope) + db.add_permission_to_organization(str(org.uuid), perm_org_admin.scope) # Create an Administration role granting both org and global admin - # Compose permissions for Administration role: global admin + org admin auto-perm role = Role( uuid7.create(), org.uuid, "Administration", - permissions=[perm0.scope, *org.permissions], + permissions=[perm0.scope, perm_org_admin.scope], ) db.create_role(role) diff --git a/paskia/db/json.py b/paskia/db/json.py index ef21f25..b5343fc 100644 --- a/paskia/db/json.py +++ b/paskia/db/json.py @@ -790,8 +790,6 @@ 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( @@ -806,30 +804,15 @@ class DB: p.orgs[key] = True break - # Automatically create or enable the common org admin permission + # Automatically allow the org to grant the org admin permission if it exists 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 + p.orgs[key] = True + if org_admin_scope not in org.permissions: + org.permissions.append(org_admin_scope) 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 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: if org_id not in self._data.orgs: @@ -1438,6 +1421,10 @@ class DB: from paskia.util.hostutil import normalize_host normalized_host = normalize_host(host) + # Strip port for domain matching (e.g., localhost:4401 -> localhost) + host_without_port = ( + normalized_host.rsplit(":", 1)[0] if normalized_host else None + ) effective_permissions = [] for scope in role_obj.permissions: if scope not in org_obj.permissions: @@ -1445,8 +1432,8 @@ class DB: # 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: + # Check domain restriction (compare without port) + if p.domain is not None and p.domain != host_without_port: continue effective_permissions.append( Permission( diff --git a/paskia/fastapi/admin.py b/paskia/fastapi/admin.py index 27db238..336497c 100644 --- a/paskia/fastapi/admin.py +++ b/paskia/fastapi/admin.py @@ -5,10 +5,10 @@ from uuid import UUID, uuid4 from fastapi import Body, FastAPI, HTTPException, Request, Response from fastapi.responses import JSONResponse +from paskia import db from paskia.authsession import EXPIRES, reset_expires from paskia.fastapi import authz from paskia.fastapi.session import AUTH_COOKIE -from paskia import db from paskia.util import ( frontend, hostutil, @@ -243,9 +243,19 @@ async def admin_remove_org_permission( request: Request, auth=AUTH_COOKIE, ): - await authz.verify( + ctx = await authz.verify( auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all ) + + # Guard rail: prevent removing auth:admin from your own org if it would lock you out + if permission_id == "auth:admin" and ctx.org.uuid == org_uuid: + # Check if any other org grants auth:admin that we're a member of + # (we only know our current org, so this effectively means we can't remove it from our own org) + raise ValueError( + "Cannot remove auth:admin from your own organization. " + "This would lock you out of admin access." + ) + db.remove_permission_from_organization(str(org_uuid), permission_id) return {"status": "ok"} @@ -778,13 +788,84 @@ 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 + from paskia.globals import passkey - rp_id = global_passkey.instance.rp_id + rp_id = passkey.instance.rp_id if domain == rp_id or domain.endswith(f".{rp_id}"): return + raise ValueError(f"Domain '{domain}' must be '{rp_id}' or its subdomain") + + +def _check_admin_lockout( + perm_uuid: str, new_domain: str | None, current_host: str | None +) -> None: + """Check if setting domain on auth:admin would lock out the admin. + + Raises ValueError if this change would result in no auth:admin permissions + being accessible from the current host. + """ + from paskia.util.hostutil import normalize_host + + normalized_host = normalize_host(current_host) + host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None + + # Get all auth:admin permissions + all_perms = db.list_permissions() + admin_perms = [p for p in all_perms if p.scope == "auth:admin"] + + # Check if at least one auth:admin would remain accessible + for p in admin_perms: + # If this is the permission being modified, use the new domain + domain = new_domain if str(p.uuid) == perm_uuid else p.domain + + # No domain restriction = accessible from anywhere + if domain is None: + return + + # Domain matches current host + if host_without_port and domain == host_without_port: + return + + # No accessible auth:admin permission would remain raise ValueError( - f"Domain '{domain}' must be the same as or a subdomain of rp_id '{rp_id}'" + "Cannot set this domain restriction: it would lock you out of admin access. " + "Ensure at least one auth:admin permission remains accessible from your current host." + ) + + +def _check_admin_lockout_on_delete(perm_uuid: str, current_host: str | None) -> None: + """Check if deleting an auth:admin permission would lock out the admin. + + Raises ValueError if this deletion would result in no auth:admin permissions + being accessible from the current host. + """ + from paskia.util.hostutil import normalize_host + + normalized_host = normalize_host(current_host) + host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None + + # Get all auth:admin permissions + all_perms = db.list_permissions() + admin_perms = [p for p in all_perms if p.scope == "auth:admin"] + + # Check if at least one auth:admin would remain accessible after deletion + for p in admin_perms: + # Skip the permission being deleted + if str(p.uuid) == perm_uuid: + continue + + # No domain restriction = accessible from anywhere + if p.domain is None: + return + + # Domain matches current host + if host_without_port and p.domain == host_without_port: + return + + # No accessible auth:admin permission would remain + raise ValueError( + "Cannot delete this permission: it would lock you out of admin access. " + "Ensure at least one auth:admin permission remains accessible from your current host." ) @@ -822,6 +903,7 @@ async def admin_create_permission( max_age="5m", ) import uuid7 + from ..db import Permission as PermDC scope = payload.get("scope") or payload.get( @@ -873,6 +955,10 @@ async def admin_update_permission( querysafe.assert_safe(new_scope, field="scope") _validate_permission_domain(domain_value) + # Safety check: prevent admin lockout when setting domain on auth:admin + if perm.scope == "auth:admin" or new_scope == "auth:admin": + _check_admin_lockout(str(perm.uuid), domain_value, request.headers.get("host")) + from ..db import Permission as PermDC db.update_permission( @@ -921,6 +1007,11 @@ async def admin_rename_permission( else: domain_value = domain if domain else None _validate_permission_domain(domain_value) + + # Safety check: prevent admin lockout when setting domain on auth:admin + if perm.scope == "auth:admin" or new_scope == "auth:admin": + _check_admin_lockout(str(perm.uuid), domain_value, request.headers.get("host")) + # All current backends support rename_permission db.rename_permission(old_scope, new_scope, display_name, domain_value) return {"status": "ok"} @@ -949,9 +1040,9 @@ async def admin_delete_permission( # Get the permission to check its scope perm = db.get_permission(perm_identifier) - # Sanity check: prevent deleting critical permissions + # Sanity check: prevent deleting critical permissions if it would lock out admin if perm.scope == "auth:admin": - raise ValueError("Cannot delete the master admin permission") + _check_admin_lockout_on_delete(str(perm.uuid), request.headers.get("host")) db.delete_permission(str(perm.uuid)) return {"status": "ok"} diff --git a/paskia/migrate/__init__.py b/paskia/migrate/__init__.py index dfe119c..e73ee45 100644 --- a/paskia/migrate/__init__.py +++ b/paskia/migrate/__init__.py @@ -61,6 +61,8 @@ async def migrate_from_sql( from paskia.db.json import ( DB as JSONDB, + ) + from paskia.db.json import ( _CredentialData, _OrgData, _PermissionData, @@ -99,7 +101,7 @@ async def migrate_from_sql( org_admin_perm_uuid = str(uuid7.create()) json_db._data.permissions[org_admin_perm_uuid] = _PermissionData( scope="auth:org:admin", - display_name="Organization Admin", + display_name="Org Admin", orgs={}, ) diff --git a/tests/test_admin.py b/tests/test_admin.py index b0259b2..13ec40e 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -1504,17 +1504,81 @@ class TestAdminPermissions: assert data["status"] == "ok" @pytest.mark.asyncio - async def test_delete_permission_auth_admin_fails( + async def test_delete_permission_auth_admin_last_one_fails( self, client: httpx.AsyncClient, session_token: str ): - """Cannot delete the auth:admin permission.""" + """Cannot delete the only auth:admin permission (would lock out admin).""" response = await client.delete( "/auth/api/admin/permission?permission_id=auth:admin", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() - assert "Cannot delete the master admin" in data["detail"] + assert "lock you out of admin access" in data["detail"] + + @pytest.mark.asyncio + async def test_delete_permission_auth_admin_with_another_succeeds( + self, client: httpx.AsyncClient, session_token: str, test_db: DB + ): + """Can delete an auth:admin permission if another accessible one exists.""" + import uuid7 + + from paskia.db import Permission + + # Create a second auth:admin permission (no domain restriction) + perm2 = Permission( + uuid=uuid7.create(), scope="auth:admin", display_name="Secondary Admin" + ) + test_db.create_permission(perm2) + + # Now we can delete the original one + response = await client.delete( + "/auth/api/admin/permission?permission_id=auth:admin", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert data["status"] == "ok" + + @pytest.mark.asyncio + async def test_delete_permission_auth_admin_domain_mismatch_fails( + self, client: httpx.AsyncClient, session_token: str, test_db: DB + ): + """Cannot delete auth:admin if remaining one has mismatched domain.""" + import uuid7 + + from paskia.db import Permission + + # Create a second auth:admin permission with a different domain + perm2 = Permission( + uuid=uuid7.create(), + scope="auth:admin", + display_name="Other Domain Admin", + domain="other.example.com", + ) + test_db.create_permission(perm2) + + # Cannot delete the original one because the remaining one is not accessible + response = await client.delete( + "/auth/api/admin/permission?permission_id=auth:admin", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "lock you out of admin access" in data["detail"] + + @pytest.mark.asyncio + async def test_remove_auth_admin_from_own_org_fails( + self, client: httpx.AsyncClient, session_token: str, test_org + ): + """Cannot remove auth:admin permission from your own organization.""" + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_id=auth:admin", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "lock you out of admin access" in data["detail"] # -------------------- Edge Cases for AuthException in Org-Admin Checks --------------------