From b980157e59ad7c8b2d2857645b0e070f8f8e1812 Mon Sep 17 00:00:00 2001
From: Yadd <yadd@debian.org>
Date: Sun, 19 Apr 2026 11:35:17 +0200
Subject: [PATCH] feat(users): allow users to regenerate their own API token
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The POST /users/{user_id}/regenerate_token endpoint used to be
documented as "admin or self" but had no permission check at all —
any authenticated caller could rotate any other user's token. This
adds a `require_admin_or_self` dependency that enforces the
documented contract: admins may regenerate any user's token; non-admins
may only regenerate their own.

Also:
- Return 404 instead of crashing when the target user does not exist.
- Bump the indexer-ui submodule to pull in the companion UI branch
  that surfaces this action as a button in the NavBar.
---
 extern/indexer-ui                            |  2 +-
 openrag/components/indexer/vectordb/utils.py |  4 ++-
 openrag/routers/users.py                     | 15 ++++++++--
 openrag/routers/utils.py                     | 31 ++++++++++++++++++++
 4 files changed, 47 insertions(+), 5 deletions(-)

diff --git a/extern/indexer-ui b/extern/indexer-ui
index 3a999914f..b84052e7b 160000
--- a/extern/indexer-ui
+++ b/extern/indexer-ui
@@ -1 +1 @@
-Subproject commit 3a999914f5e57e404cbf9d60cd3ebd1073a43ead
+Subproject commit b84052e7b5d959f4e3bc936731d5843e899c5852
diff --git a/openrag/components/indexer/vectordb/utils.py b/openrag/components/indexer/vectordb/utils.py
index a497fbbb0..fcc9f3716 100644
--- a/openrag/components/indexer/vectordb/utils.py
+++ b/openrag/components/indexer/vectordb/utils.py
@@ -425,9 +425,11 @@ def delete_user(self, user_id: int) -> bool:
             s.commit()
             return True
 
-    def regenerate_user_token(self, user_id: int) -> dict:
+    def regenerate_user_token(self, user_id: int) -> dict | None:
         with self.Session() as s:
             user = s.query(User).filter(User.id == user_id).first()
+            if not user:
+                return None
             new_token = f"or-{secrets.token_hex(16)}"
             hashed_token = self.hash_token(new_token)
             user.token = hashed_token
diff --git a/openrag/routers/users.py b/openrag/routers/users.py
index dadc76a41..494086c6f 100644
--- a/openrag/routers/users.py
+++ b/openrag/routers/users.py
@@ -4,7 +4,7 @@
 from utils.dependencies import get_task_state_manager, get_vectordb
 from utils.logger import get_logger
 
-from .utils import DEFAULT_FILE_QUOTA, current_user, require_admin
+from .utils import DEFAULT_FILE_QUOTA, current_user, require_admin, require_admin_or_self
 
 logger = get_logger()
 router = APIRouter()
@@ -207,7 +207,7 @@ async def delete_user(user_id: int, vectordb=Depends(get_vectordb), admin_user=D
 - `user_id`: User identifier
 
 **Permissions:**
-- Requires admin role (or user can regenerate their own token)
+- Requires admin role, or the caller must be regenerating their own token.
 
 **Behavior:**
 - Generates a new authentication token
@@ -223,11 +223,20 @@ async def delete_user(user_id: int, vectordb=Depends(get_vectordb), admin_user=D
 **Note:** Store the new token securely - the old token is now invalid.
 """,
 )
-async def regenerate_user_token(user_id: int, vectordb=Depends(get_vectordb)):
+async def regenerate_user_token(
+    user_id: int,
+    vectordb=Depends(get_vectordb),
+    _auth=Depends(require_admin_or_self),
+):
     """
     Regenerate a user's token.
     """
     user = await vectordb.regenerate_user_token.remote(user_id)
+    if user is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"User '{user_id}' not found",
+        )
     logger.info("Regenerated user token", user_id=user_id)
     return JSONResponse(status_code=status.HTTP_200_OK, content=user)
 
diff --git a/openrag/routers/utils.py b/openrag/routers/utils.py
index 1e1d3e6e6..bfd8a908c 100644
--- a/openrag/routers/utils.py
+++ b/openrag/routers/utils.py
@@ -182,6 +182,37 @@ def require_admin(user=Depends(current_user)):
     return user
 
 
+def request_user_id(request: Request) -> int | None:
+    """Return the user_id from path params (as int), or None."""
+    raw = request.path_params.get("user_id", None)
+    if raw is None:
+        return None
+    try:
+        return int(raw)
+    except (TypeError, ValueError):
+        return None
+
+
+def require_admin_or_self(
+    target_user_id: int | None = Depends(request_user_id),
+    user=Depends(current_user),
+):
+    """Ensure the caller is admin or is acting on their own account."""
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Authentication required",
+        )
+    if user.get("is_admin", False):
+        return user
+    if target_user_id is not None and user.get("id") == target_user_id:
+        return user
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail="Admin privileges or self-access required",
+    )
+
+
 async def check_user_file_quota(
     user=Depends(current_user),
 ):