Security fixes

Author: azvoleff

gefapi/__init__.py

@@ -298,14 +298,47 @@ def set_security_headers(response):
 # Transfer rate limiting configuration to Flask app config
 app.config["RATE_LIMITING"] = SETTINGS.get("RATE_LIMITING", {})
 
-# Ensure JWT_SECRET_KEY is set with proper fallback
+# Ensure JWT_SECRET_KEY is set with proper fallback and validation
 jwt_secret = (
     SETTINGS.get("JWT_SECRET_KEY")
     or SETTINGS.get("SECRET_KEY")
     or os.getenv("JWT_SECRET_KEY")
     or os.getenv("SECRET_KEY")
 )
 
+# Security: Validate JWT secret key to prevent insecure configurations
+# This list includes actual defaults from .env.example and develop/test configs
+INSECURE_KEY_PATTERNS = [
+    "your_jwt_secret_key_here",
+    "your_secret_key_here",
+    "develop_jwt_secret",
+    "develop_secret_key",
+    "test_jwt_secret_key",
+    "test_secret_key",
+]
+
+if not jwt_secret:
+    if os.getenv("ENVIRONMENT") == "prod":
+        raise RuntimeError(
+            "CRITICAL SECURITY ERROR: JWT_SECRET_KEY must be set in production. "
+            "Set the JWT_SECRET_KEY or SECRET_KEY environment variable."
+        )
+    logger.warning(
+        "JWT_SECRET_KEY is not set. This is insecure and must be fixed before "
+        "deploying to production. Set JWT_SECRET_KEY or SECRET_KEY."
+    )
+elif jwt_secret.lower() in INSECURE_KEY_PATTERNS or len(jwt_secret) < 32:
+    if os.getenv("ENVIRONMENT") == "prod":
+        raise RuntimeError(
+            "CRITICAL SECURITY ERROR: JWT_SECRET_KEY is insecure (too short or "
+            "uses a known default value). Use a cryptographically secure random "
+            "string of at least 32 characters."
+        )
+    logger.warning(
+        "JWT_SECRET_KEY appears insecure (too short or uses a default value). "
+        "This must be fixed before deploying to production."
+    )
+
 app.config["JWT_SECRET_KEY"] = jwt_secret
 app.config["JWT_ACCESS_TOKEN_EXPIRES"] = SETTINGS.get("JWT_ACCESS_TOKEN_EXPIRES")
 app.config["JWT_TOKEN_LOCATION"] = SETTINGS.get("JWT_TOKEN_LOCATION")
@@ -431,13 +464,21 @@ def ping():
 
 @app.route("/debug/routes", methods=["GET"])
 def debug_routes():
-    """Debug endpoint to show all registered routes"""
+    """Debug endpoint to show all registered routes.
+
+    Security: This endpoint is only available in development/test environments.
+    It is completely disabled in production and staging unless explicitly enabled.
+    """
     import os
 
-    if (
-        os.getenv("ENVIRONMENT") == "prod"
-        and os.getenv("DEBUG_ROUTES_ENABLED", "false").lower() != "true"
-    ):
+    environment = os.getenv("ENVIRONMENT", "dev")
+    debug_routes_enabled = os.getenv("DEBUG_ROUTES_ENABLED", "false").lower() == "true"
+
+    # Security: Only allow in dev/test environments, or if explicitly enabled
+    if environment not in ("dev", "test") and not debug_routes_enabled:
+        logger.warning(
+            f"[SECURITY]: Blocked debug/routes access in {environment} environment"
+        )
         abort(404)
 
     routes_info = []

gefapi/routes/api/v1/executions.py

@@ -103,6 +103,30 @@ def run_script(script):
         if request.get_json(silent=True):
             params.update(request.get_json())
         if "token" in params:
+            # Security: Log deprecated token query parameter usage
+            # Tokens in URLs are logged in server logs, browser history, and referrer
+            # headers, which is a security risk (CWE-598)
+            import rollbar
+
+            rollbar.report_message(
+                "SECURITY WARNING: Deprecated token query parameter used in "
+                f"script execution request for script '{script}' by user "
+                f"'{user.email}'. Token-based authentication via query "
+                "parameters is deprecated and will be removed in a future "
+                "version. Use Authorization header instead.",
+                level="warning",
+                extra_data={
+                    "script": script,
+                    "user_id": str(user.id),
+                    "user_email": user.email,
+                    "endpoint": "run_script",
+                    "security_issue": "CWE-598",
+                },
+            )
+            logger.warning(
+                f"[SECURITY]: Deprecated token query parameter used by {user.email} "
+                f"for script {script}. This should be migrated to header-based auth."
+            )
             del params["token"]
         execution = ExecutionService.create_execution(script, params, user)
     except ScriptNotFound as e:

gefapi/services/user_service.py

@@ -83,6 +83,20 @@ def _validate_password_strength(password: str) -> None:
 class UserService:
     """User Class"""
 
+    # Security: Explicitly allowed fields for filter and sort operations
+    # to prevent unauthorized access to sensitive model fields
+    USER_ALLOWED_FILTER_FIELDS = {
+        "id",
+        "email",
+        "name",
+        "role",
+        "country",
+        "institution",
+        "created_at",
+        "updated_at",
+    }
+    USER_ALLOWED_SORT_FIELDS = USER_ALLOWED_FILTER_FIELDS
+
     @staticmethod
     def create_user(user, legacy=True):
         """Create a new user account.
@@ -295,7 +309,7 @@ def get_users(
 
         query = db.session.query(User)
 
-        # SQL-style filter_param
+        # SQL-style filter_param with field allowlist for security
         if filter_param:
             import re
 
@@ -312,6 +326,12 @@ def get_users(
                     field = field.strip().lower()
                     op = op.strip().lower()
                     value = value.strip().strip("'\"")
+                    # Security: Only allow filtering on explicitly allowed fields
+                    if field not in UserService.USER_ALLOWED_FILTER_FIELDS:
+                        logger.warning(
+                            f"[SERVICE]: Rejected filter on disallowed field: {field}"
+                        )
+                        continue
                     col = getattr(User, field, None)
                     if col is not None:
                         # Check if this is a string column for case-insensitive compare
@@ -357,6 +377,12 @@ def get_users(
                 parts = sort_expr.split()
                 field = parts[0].lower()
                 direction = parts[1].lower() if len(parts) > 1 else "asc"
+                # Security: Only allow sorting on explicitly allowed fields
+                if field not in UserService.USER_ALLOWED_SORT_FIELDS:
+                    logger.warning(
+                        f"[SERVICE]: Rejected sort on disallowed field: {field}"
+                    )
+                    continue
                 col = getattr(User, field, None)
                 if col is not None:
                     if direction == "desc":