ct/bmc_hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden admin users endpoint fallback on partial schemas

Browse Source
This commit is contained in:
Christian 2026-03-07 03:14:29 +01:00
parent 7e77266d97
commit 7678b58cb4
1 changed files with 42 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
app/auth/backend/admin.py
View File

@ -37,6 +37,20 @@ def _users_column_exists(column_name: str) -> bool:
return bool(result)
def _table_exists(table_name: str) -> bool:
result = execute_query_single(
"""
SELECT 1
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_name = %s
LIMIT 1
""",
(table_name,)
)
return bool(result)
@router.get("/admin/users", dependencies=[Depends(require_permission("users.manage"))])
async def list_users():
is_2fa_expr = "u.is_2fa_enabled" if _users_column_exists("is_2fa_enabled") else "FALSE AS is_2fa_enabled"
@ -45,6 +59,15 @@ async def list_users():
telefoni_ip_expr = "u.telefoni_phone_ip" if _users_column_exists("telefoni_phone_ip") else "NULL::varchar AS telefoni_phone_ip"
telefoni_username_expr = "u.telefoni_phone_username" if _users_column_exists("telefoni_phone_username") else "NULL::varchar AS telefoni_phone_username"
last_login_expr = "u.last_login_at" if _users_column_exists("last_login_at") else "NULL::timestamp AS last_login_at"
has_user_groups = _table_exists("user_groups")
has_groups = _table_exists("groups")
if has_user_groups and has_groups:
groups_join = "LEFT JOIN user_groups ug ON u.user_id = ug.user_id LEFT JOIN groups g ON ug.group_id = g.id"
groups_select = "COALESCE(array_remove(array_agg(g.name), NULL), ARRAY[]::varchar[]) AS groups"
else:
groups_join = ""
groups_select = "ARRAY[]::varchar[] AS groups"
try:
users = execute_query(
@ -53,18 +76,32 @@ async def list_users():
u.is_active, u.is_superadmin, {is_2fa_expr},
{telefoni_extension_expr}, {telefoni_active_expr}, {telefoni_ip_expr}, {telefoni_username_expr},
u.created_at, {last_login_expr},
COALESCE(array_remove(array_agg(g.name), NULL), ARRAY[]::varchar[]) AS groups
{groups_select}
FROM users u
LEFT JOIN user_groups ug ON u.user_id = ug.user_id
LEFT JOIN groups g ON ug.group_id = g.id
{groups_join}
GROUP BY u.user_id
ORDER BY u.user_id
"""
)
return users
except Exception as exc:
logger.error("❌ Failed to load admin users: %s", exc)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Could not load users") from exc
logger.warning("⚠️ Admin user query fallback triggered: %s", exc)
try:
users = execute_query(
f"""
SELECT u.user_id, u.username, u.email, u.full_name,
u.is_active, u.is_superadmin, {is_2fa_expr},
{telefoni_extension_expr}, {telefoni_active_expr}, {telefoni_ip_expr}, {telefoni_username_expr},
u.created_at, {last_login_expr},
ARRAY[]::varchar[] AS groups
FROM users u
ORDER BY u.user_id
"""
)
return users
except Exception as fallback_exc:
logger.error("❌ Failed to load admin users (fallback): %s", fallback_exc)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Could not load users") from fallback_exc
@router.post("/admin/users", status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_permission("users.manage"))])