From 9cec3d001feabb4e5103714287e3073ff343c201 Mon Sep 17 00:00:00 2001 From: JiriUhlir <149317995+JiriUhlir@users.noreply.github.com> Date: Wed, 10 Jun 2026 07:56:02 +0200 Subject: [PATCH] sprava uzivatelu --- app/auth.py | 53 +++++++++--- app/db/migrations.py | 21 +++++ app/db/users.py | 125 +++++++++++++++++++++++++++ app/main.py | 3 +- app/routes/auth.py | 10 ++- app/routes/users.py | 186 ++++++++++++++++++++++++++++++++++++++++ app/templates/layout.py | 6 +- 7 files changed, 391 insertions(+), 13 deletions(-) create mode 100644 app/db/users.py create mode 100644 app/routes/users.py diff --git a/app/auth.py b/app/auth.py index e9fe271..2511657 100644 --- a/app/auth.py +++ b/app/auth.py @@ -4,10 +4,39 @@ from fastapi import HTTPException, Request, status from passlib.context import CryptContext from app.db.database import get_connection +from app.db.migrations import run_migrations pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +def _table_columns(con, table_name: str) -> set[str]: + rows = con.execute(f"PRAGMA table_info({table_name})").fetchall() + return {row["name"] for row in rows} + + +def _user_select_columns(con) -> str: + columns = _table_columns(con, "users") + enabled_select = "COALESCE(is_enabled, 1) AS is_enabled" if "is_enabled" in columns else "1 AS is_enabled" + return f"id, username, display_name, email, password_hash, role, is_active, {enabled_select}, created_at, updated_at" + + +def mark_last_login(user_id: int) -> None: + con = get_connection() + columns = _table_columns(con, "users") + if "last_login_at" in columns: + con.execute( + """ + UPDATE users + SET last_login_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + (user_id,), + ) + con.commit() + con.close() + + def hash_password(password: str) -> str: return pwd_context.hash(password) @@ -20,11 +49,12 @@ def verify_password(plain_password: str, password_hash: str) -> bool: def get_user_by_username(username: str) -> dict[str, Any] | None: + run_migrations() con = get_connection() row = con.execute( - """ - SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at + f""" + SELECT {_user_select_columns(con)} FROM users WHERE username = ? """, @@ -40,10 +70,11 @@ def find_or_create_oauth_user(username: str, display_name: str, email: str) -> d display_name = (display_name or username).strip() or username email = (email or "").strip() + run_migrations() con = get_connection() row = con.execute( - """ - SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at + f""" + SELECT {_user_select_columns(con)} FROM users WHERE username = ? """, @@ -82,8 +113,8 @@ def find_or_create_oauth_user(username: str, display_name: str, email: str) -> d con.commit() user_id = row["id"] if row else con.execute("SELECT last_insert_rowid() AS id").fetchone()["id"] user = con.execute( - """ - SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at + f""" + SELECT {_user_select_columns(con)} FROM users WHERE id = ? """, @@ -95,11 +126,12 @@ def find_or_create_oauth_user(username: str, display_name: str, email: str) -> d def get_user_by_id(user_id: int) -> dict[str, Any] | None: + run_migrations() con = get_connection() row = con.execute( - """ - SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at + f""" + SELECT {_user_select_columns(con)} FROM users WHERE id = ? """, @@ -113,12 +145,13 @@ def get_user_by_id(user_id: int) -> dict[str, Any] | None: def authenticate_user(username: str, password: str) -> dict[str, Any] | None: user = get_user_by_username(username.strip()) - if not user or not user.get("is_active"): + if not user or not user.get("is_active") or not user.get("is_enabled", 1): return None if not verify_password(password, user.get("password_hash") or ""): return None + mark_last_login(int(user["id"])) return user @@ -134,7 +167,7 @@ def current_user(request: Request) -> dict[str, Any] | None: return None user = get_user_by_id(user_id) - if not user or not user.get("is_active"): + if not user or not user.get("is_active") or not user.get("is_enabled", 1): request.session.pop("user_id", None) return None diff --git a/app/db/migrations.py b/app/db/migrations.py index 2e07bd4..2287d1f 100644 --- a/app/db/migrations.py +++ b/app/db/migrations.py @@ -1,8 +1,29 @@ from app.db.database import get_connection +def _table_columns(con, table_name: str) -> set[str]: + rows = con.execute(f"PRAGMA table_info({table_name})").fetchall() + return {row["name"] for row in rows} + + +def ensure_user_columns(con) -> None: + columns = _table_columns(con, "users") + if not columns: + return + + if "is_enabled" not in columns: + default_expression = "COALESCE(is_active, 1)" if "is_active" in columns else "1" + con.execute("ALTER TABLE users ADD COLUMN is_enabled INTEGER NOT NULL DEFAULT 1") + con.execute(f"UPDATE users SET is_enabled = {default_expression} WHERE is_enabled IS NULL OR is_enabled = 1") + + if "updated_at" not in columns: + con.execute("ALTER TABLE users ADD COLUMN updated_at TEXT") + con.execute("UPDATE users SET updated_at = COALESCE(created_at, CURRENT_TIMESTAMP) WHERE updated_at IS NULL") + + def run_migrations(): con = get_connection() + ensure_user_columns(con) con.execute( """ diff --git a/app/db/users.py b/app/db/users.py new file mode 100644 index 0000000..f48326c --- /dev/null +++ b/app/db/users.py @@ -0,0 +1,125 @@ +from typing import Any + +from app.db.database import get_connection +from app.db.migrations import run_migrations + +ROLES = ("admin", "developer", "viewer") + + +def _table_columns(con, table_name: str) -> set[str]: + rows = con.execute(f"PRAGMA table_info({table_name})").fetchall() + return {row["name"] for row in rows} + + +def list_users() -> list[dict[str, Any]]: + run_migrations() + con = get_connection() + columns = _table_columns(con, "users") + last_login_select = "last_login_at" if "last_login_at" in columns else "NULL AS last_login_at" + + rows = con.execute( + f""" + SELECT + id, + username, + display_name, + email, + role, + COALESCE(is_enabled, 1) AS is_enabled, + created_at, + updated_at, + {last_login_select} + FROM users + ORDER BY id + """ + ).fetchall() + + con.close() + return [dict(row) for row in rows] + + +def get_user(user_id: int) -> dict[str, Any] | None: + run_migrations() + con = get_connection() + columns = _table_columns(con, "users") + last_login_select = "last_login_at" if "last_login_at" in columns else "NULL AS last_login_at" + + row = con.execute( + f""" + SELECT + id, + username, + display_name, + email, + role, + COALESCE(is_enabled, 1) AS is_enabled, + created_at, + updated_at, + {last_login_select} + FROM users + WHERE id = ? + """, + (user_id,), + ).fetchone() + + con.close() + return dict(row) if row else None + + +def count_enabled_admins(excluding_user_id: int | None = None) -> int: + run_migrations() + con = get_connection() + params: list[Any] = [] + exclusion = "" + if excluding_user_id is not None: + exclusion = "AND id != ?" + params.append(excluding_user_id) + + row = con.execute( + f""" + SELECT COUNT(*) AS count + FROM users + WHERE LOWER(COALESCE(role, '')) = 'admin' + AND COALESCE(is_enabled, 1) = 1 + {exclusion} + """, + params, + ).fetchone() + + con.close() + return int(row["count"] or 0) + + +def update_user_role(user_id: int, role: str) -> None: + if role not in ROLES: + raise ValueError("Invalid user role") + + run_migrations() + con = get_connection() + con.execute( + """ + UPDATE users + SET role = ?, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + (role, user_id), + ) + con.commit() + con.close() + + +def set_user_enabled(user_id: int, enabled: bool) -> None: + run_migrations() + con = get_connection() + con.execute( + """ + UPDATE users + SET is_enabled = ?, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + (1 if enabled else 0, user_id), + ) + con.commit() + con.close() diff --git a/app/main.py b/app/main.py index ecdd5fe..aad3124 100644 --- a/app/main.py +++ b/app/main.py @@ -5,7 +5,7 @@ from fastapi.staticfiles import StaticFiles from starlette.middleware.sessions import SessionMiddleware from .config import read_env_bool, read_env_value -from .routes import alerting, apps, audit, auth, backups, deployments, health, incidents, jobs, migration_readiness, operations, scheduled_scripts, workers +from .routes import alerting, apps, audit, auth, backups, deployments, health, incidents, jobs, migration_readiness, operations, scheduled_scripts, users, workers def create_app() -> FastAPI: @@ -34,6 +34,7 @@ def create_app() -> FastAPI: app.include_router(scheduled_scripts.router) app.include_router(jobs.router) app.include_router(workers.router) + app.include_router(users.router) app.include_router(audit.router) return app diff --git a/app/routes/auth.py b/app/routes/auth.py index b903108..d5bfae3 100644 --- a/app/routes/auth.py +++ b/app/routes/auth.py @@ -9,7 +9,7 @@ from urllib.request import urlopen from fastapi import APIRouter, Form, Request from fastapi.responses import HTMLResponse, RedirectResponse -from app.auth import authenticate_user, current_user, find_or_create_oauth_user +from app.auth import authenticate_user, current_user, find_or_create_oauth_user, mark_last_login from app.config import ( get_gitea_public_url, get_gitea_redirect_uri, @@ -114,12 +114,16 @@ def gitea_callback(request: Request, code: str = "", state: str = "", error: str display_name = (gitea_user.get("full_name") or username).strip() email = (gitea_user.get("email") or "").strip() user = find_or_create_oauth_user(username, display_name, email) + if not user.get("is_enabled", 1): + _log_gitea_failure("disabled_user", username=username) + return _render_login("User is disabled in the portal.") if not user.get("is_active"): _log_gitea_failure("disabled_user", username=username) return _render_login("Uživatel je v portálu vypnutý.") request.session.clear() request.session["user_id"] = user["id"] + mark_last_login(int(user["id"])) log_audit_event( user, action="auth.gitea.login.success", @@ -201,12 +205,16 @@ def google_callback(request: Request, code: str = "", state: str = "", error: st username = email display_name = (google_user.get("name") or email).strip() user = find_or_create_oauth_user(username, display_name, email) + if not user.get("is_enabled", 1): + _log_google_failure("disabled_user", username=username) + return _render_login("User is disabled in the portal.") if not user.get("is_active"): _log_google_failure("disabled_user", username=username) return _render_login("UĹľivatel je v portálu vypnutĂ˝.") request.session.clear() request.session["user_id"] = user["id"] + mark_last_login(int(user["id"])) log_audit_event( user, action="auth.google.login.success", diff --git a/app/routes/users.py b/app/routes/users.py new file mode 100644 index 0000000..d26d66e --- /dev/null +++ b/app/routes/users.py @@ -0,0 +1,186 @@ +import html +from urllib.parse import quote + +from fastapi import APIRouter, Depends, Form, HTTPException, Request +from fastapi.responses import HTMLResponse, RedirectResponse + +from app.auth import require_user +from app.db.audit import log_audit_event +from app.db.users import ROLES, count_enabled_admins, get_user, list_users, set_user_enabled, update_user_role +from app.templates.layout import page + +router = APIRouter() + + +def require_admin(user: dict) -> None: + if (user.get("role") or "").lower() != "admin": + raise HTTPException(status_code=403, detail="Only admins can manage users") + + +def render_role_options(selected: str) -> str: + options = [] + for role in ROLES: + selected_attr = " selected" if selected == role else "" + options.append(f'') + return "".join(options) + + +def render_enabled_pill(enabled) -> str: + if enabled: + return 'enabled' + return 'disabled' + + +@router.get("/admin/users", response_class=HTMLResponse) +def users_page(request: Request, message: str = "", error: str = "", user=Depends(require_user)): + require_admin(user) + + notice = "" + if message: + notice = f'

{html.escape(message)}

' + if error: + notice = f'

{html.escape(error)}

' + + rows = "" + for managed_user in list_users(): + user_id = int(managed_user.get("id")) + user_id_html = html.escape(str(user_id)) + username = html.escape(managed_user.get("username", "") or "") + display_name = html.escape(managed_user.get("display_name", "") or "") + email = html.escape(managed_user.get("email", "") or "") + role = html.escape(managed_user.get("role", "") or "") + created_at = html.escape(managed_user.get("created_at", "") or "") + last_login_at = html.escape(managed_user.get("last_login_at", "") or "") + enabled = bool(managed_user.get("is_enabled", True)) + toggle_label = "Disable" if enabled else "Enable" + toggle_action = "disable" if enabled else "enable" + + rows += f""" + + {user_id_html} + {username} + {display_name} + {email} + +
+ + +
+ + {render_enabled_pill(enabled)} + {created_at} + {last_login_at} + +
+ +
+ + + """ + + if not rows: + rows = 'No users found.' + + return page( + "Users", + f""" +
+

Users

+

Administration of portal users and roles.

+ {notice} +
+ +
+

User List

+ + + + + + + + + + + + + {rows} +
IDUsernameDisplay nameEmailRoleStatusCreatedLast loginActions
+
+ """, + user=user, + ) + + +@router.post("/admin/users/{user_id}/role") +def update_role_action(user_id: int, role: str = Form(...), user=Depends(require_user)): + require_admin(user) + target = get_user(user_id) + if not target: + raise HTTPException(status_code=404, detail="User not found") + if role not in ROLES: + raise HTTPException(status_code=400, detail="Invalid role") + + old_role = (target.get("role") or "").lower() + if old_role == "admin" and role != "admin" and count_enabled_admins(excluding_user_id=user_id) == 0: + return RedirectResponse( + url="/portal/admin/users?error=" + quote("Cannot remove admin role from the last enabled admin."), + status_code=303, + ) + + update_user_role(user_id, role) + if old_role != role: + log_audit_event( + user, + action="user.role.updated", + target_type="user", + target_id=user_id, + metadata={ + "username": target.get("username"), + "old_role": old_role, + "new_role": role, + }, + ) + + return RedirectResponse(url="/portal/admin/users?message=" + quote("User role updated."), status_code=303) + + +@router.post("/admin/users/{user_id}/disable") +def disable_user_action(user_id: int, user=Depends(require_user)): + require_admin(user) + target = get_user(user_id) + if not target: + raise HTTPException(status_code=404, detail="User not found") + if (target.get("role") or "").lower() == "admin" and count_enabled_admins(excluding_user_id=user_id) == 0: + return RedirectResponse( + url="/portal/admin/users?error=" + quote("Cannot disable the last enabled admin."), + status_code=303, + ) + + set_user_enabled(user_id, False) + log_audit_event( + user, + action="user.disabled", + target_type="user", + target_id=user_id, + metadata={"username": target.get("username")}, + ) + return RedirectResponse(url="/portal/admin/users?message=" + quote("User disabled."), status_code=303) + + +@router.post("/admin/users/{user_id}/enable") +def enable_user_action(user_id: int, user=Depends(require_user)): + require_admin(user) + target = get_user(user_id) + if not target: + raise HTTPException(status_code=404, detail="User not found") + + set_user_enabled(user_id, True) + log_audit_event( + user, + action="user.enabled", + target_type="user", + target_id=user_id, + metadata={"username": target.get("username")}, + ) + return RedirectResponse(url="/portal/admin/users?message=" + quote("User enabled."), status_code=303) diff --git a/app/templates/layout.py b/app/templates/layout.py index df44222..28fd287 100644 --- a/app/templates/layout.py +++ b/app/templates/layout.py @@ -12,7 +12,10 @@ def page(title: str, body: str, user=None) -> str: if user: username = html.escape(user.get("username", "")) display_name = html.escape(user.get("display_name") or user.get("username", "")) - nav = """ + admin_links = "" + if (user.get("role") or "").lower() == "admin": + admin_links = 'Users' + nav = f"""