fix: address DeepSource audit findings

server/pssm_gremlin_server/auth.py

@@ -23,7 +23,7 @@
 from typing import Any
 
 import sqlalchemy as sa
-from flask import current_app, g, jsonify, redirect, render_template, request, url_for
+from flask import current_app, g, jsonify, redirect, request, url_for
 from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
 from werkzeug.security import check_password_hash, generate_password_hash
 
@@ -471,21 +471,27 @@ def load_current_user() -> dict[str, Any] | None:
         payload = validate_token(token)
         if payload is not None:
             user = db.get_user(payload["uid"])
-            if user is not None and _is_account_blocked(user) is None:
-                if payload.get("ver", 0) == user.get("token_version", 0):
-                    g.auth_method = "bearer"
-                    return user
+            if (
+                user is not None
+                and _is_account_blocked(user) is None
+                and payload.get("ver", 0) == user.get("token_version", 0)
+            ):
+                g.auth_method = "bearer"
+                return user
     # 2. Cookie — browser page navigations after login (read-only; CSRF-prone)
     #    Guest accounts ARE permitted to use cookie-based web access.
     token = request.cookies.get("auth_token")
     if token:
         payload = validate_token(token)
         if payload is not None:
             user = db.get_user(payload["uid"])
-            if user is not None and _is_account_blocked(user) is None:
-                if payload.get("ver", 0) == user.get("token_version", 0):
-                    g.auth_method = "cookie"
-                    return user
+            if (
+                user is not None
+                and _is_account_blocked(user) is None
+                and payload.get("ver", 0) == user.get("token_version", 0)
+            ):
+                g.auth_method = "cookie"
+                return user
 
     # 3. API key (programmatic access — restricted privileges)
     #    Guest accounts are not permitted to use API keys.

server/pssm_gremlin_server/pssm_gremlin.py

@@ -2,9 +2,6 @@
 # Distributed under the terms of the GNU General Public License v3.0.
 # SPDX-License-Identifier: GPL-3.0-only
 
-
-#! /mnt/data/envs/conda_env/envs/REvoDesign/bin/python
-
 from __future__ import annotations
 
 import grp
@@ -26,7 +23,7 @@
 from celery import Celery
 from celery.result import AsyncResult
 from docker import types
-from flask import Flask, current_app, g, jsonify, request
+from flask import Flask, g, jsonify, request
 from pssm_gremlin_server.db import TaskDatabase
 from sqlalchemy.exc import IntegrityError
 from werkzeug.utils import secure_filename
@@ -39,7 +36,6 @@
 
 from pssm_gremlin_server.auth import UserDatabase  # noqa: E402
 from pssm_gremlin_server.auth import _env_bool  # noqa: E402
-from pssm_gremlin_server.auth import _env_int  # noqa: E402
 from pssm_gremlin_server.auth import _env_str  # noqa: E402
 from pssm_gremlin_server.auth import send_admin_digest  # noqa: E402
 
@@ -253,7 +249,7 @@ def _digest_loop() -> None:
                 with open(_digest_lock, "w") as _lock_fh:
                     fcntl.flock(_lock_fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                     send_admin_digest()
-            except (BlockingIOError, OSError):
+            except OSError:
                 pass  # another worker has the lock — try next cycle
             except Exception:
                 logging.exception("Admin digest thread failed")
@@ -660,7 +656,7 @@ def _create_mount(mount_name: str, path: str, read_only=True) -> tuple[types.Mou
     target_path = os.path.join(_ROOT_MOUNT_DIRECTORY, mount_name)
 
     if not read_only:
-        logging.warning(f"{mount_name} is not read-only!")
+        logging.warning("%s is not read-only!", mount_name)
 
     if os.path.isdir(path):
         source_path = path
@@ -670,7 +666,7 @@ def _create_mount(mount_name: str, path: str, read_only=True) -> tuple[types.Mou
         mounted_path = os.path.join(target_path, os.path.basename(path))
     if not os.path.exists(source_path):
         os.makedirs(source_path)
-    logging.info(f"Mounting {source_path} -> {target_path}")
+    logging.info("Mounting %s -> %s", source_path, target_path)
     mount = types.Mount(
         target=str(target_path),
         source=str(source_path),
@@ -818,8 +814,7 @@ def _pack_failed_results_archive(task: dict, error: Any) -> None:
 def format_times(timestamp):
     if timestamp:
         return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S")
-    else:
-        return None
+    return None
 
 
 def format_walltime(seconds: Any) -> str:

server/pssm_gremlin_server/routes.py

@@ -17,6 +17,7 @@
 import os
 import shutil
 import time
+from typing import Any
 
 from celery.result import AsyncResult
 from flask import current_app, g, jsonify, redirect, render_template, request, send_from_directory, url_for
@@ -735,23 +736,21 @@ def auth_forgot_password():
     return jsonify({"message": "If that email is registered, a reset link has been sent."}), 200
 
 
-@app.route("/PSSM_GREMLIN/reset_password", methods=["GET", "POST"])
-def auth_reset_password():
-    """Password-reset page.
+@app.route("/PSSM_GREMLIN/reset_password", methods=["GET"])
+def auth_reset_password_page():
+    """Render the password-reset page for a valid reset token."""
+    token = request.args.get("c", "").strip()
+    if not token:
+        return render_template("error.html", code=400, message="Missing reset token."), 400
+    user_id = validate_reset_token(token)
+    if user_id is None:
+        return render_template("error.html", code=400, message="Invalid or expired reset token."), 400
+    return render_template("reset-password.html", token=token), 200
 
-    ``GET`` — renders the new-password form (requires ``?c=`` token).
-    ``POST`` — sets the new password.
-    """
-    if request.method == "GET":
-        token = request.args.get("c", "").strip()
-        if not token:
-            return render_template("error.html", code=400, message="Missing reset token."), 400
-        user_id = validate_reset_token(token)
-        if user_id is None:
-            return render_template("error.html", code=400, message="Invalid or expired reset token."), 400
-        return render_template("reset-password.html", token=token), 200
-
-    # POST
+
+@app.route("/PSSM_GREMLIN/reset_password", methods=["POST"])
+def auth_reset_password():
+    """Set a new password using a password-reset token."""
     req = _parse_body(ResetPasswordRequest)
     if isinstance(req, tuple):
         return req
@@ -1069,27 +1068,29 @@ def auth_revoke_api_key():
     return jsonify({"message": "API key revoked"}), 200
 
 
-@app.route("/PSSM_GREMLIN/api/auth/admin/users", methods=["GET", "POST"])
+@app.route("/PSSM_GREMLIN/api/auth/admin/users", methods=["GET"])
 @login_required
 def admin_users():
-    """Admin-only user management.
-
-    ``GET`` — list all users (safe fields only).
-    ``POST`` — create a new user (auto-verified, immediately active).
-    """
+    """Admin-only user listing with safe fields only."""
     if _blocked := require_admin():
         return _blocked
 
     db = _get_user_db()
+    users = db.list_users()
+    safe = [UserResponse.model_validate(u).model_dump() for u in users]
+    return jsonify({"users": safe}), 200
 
-    if request.method == "GET":
-        users = db.list_users()
-        safe = [UserResponse.model_validate(u).model_dump() for u in users]
-        return jsonify({"users": safe}), 200
 
-    # POST — state-changing, require Bearer token for CSRF protection
+@app.route("/PSSM_GREMLIN/api/auth/admin/users", methods=["POST"])
+@login_required
+def admin_create_user():
+    """Admin-only user creation for pre-approved accounts."""
+    if _blocked := require_admin():
+        return _blocked
     if _blocked := require_bearer_auth():
         return _blocked
+
+    db = _get_user_db()
     req = _parse_body(AdminCreateUserRequest)
     if isinstance(req, tuple):
         return req