feat(users): allow users to regenerate their own API token

openrag/components/indexer/vectordb/utils.py

@@ -413,9 +413,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

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)
 

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),
 ):