Phase 0 hardening + batched AI moderation + websocket/perf fixes

Security / correctness
- Privilege escalation: rule/api-key/project-update routes now gate on
  can_manage_members instead of is_member; plain members could previously
  rewrite or disable moderation rules and delete API keys.
- Project.is_owner() AttributeError on non-admin API-user deletion (route
  + template) replaced with user_id comparison.
- Flask-Limiter on /auth/login, /auth/register, /auth/change-password,
  and /api/moderate (per API key id).
- Remove silent SECRET_KEY fallback: docker-compose fails hard if unset;
  production config validator refuses to boot without SECRET_KEY and
  OPENAI_API_KEY.
- Require verified email in Google OAuth auto-link flow.
- User-supplied regex now evaluated via the regex package with a 1s
  timeout, closing the ReDoS vector.
- Cap batched AI chunk count (max 40) and concurrent workers (200 -> 10).
- openai>=2.32 (from 1.55.3 so reasoning_effort kwarg works); add regex,
  Flask-Limiter, gunicorn, eventlet, psycogreen; drop the stdlib-shadowing
  pypi asyncio package.

Production runtime
- New wsgi.py: eventlet.monkey_patch + psycogreen.patch_psycopg so DB
  queries cooperate with the event loop; sets SOCKETIO_ASYNC_MODE=eventlet
  for app startup.
- Dockerfiles run as non-root app user and CMD gunicorn with
  --worker-class eventlet. run.py refuses FLASK_CONFIG=production so a
  misconfigured deploy fails loudly.
- cloudbuild.yaml stops managing env vars/secrets on deploy — it was
  about to clobber secret-backed env vars with plain ones. Runtime config
  is now owned by the Cloud Run service.

SocketIO async-mode fix (was causing full site lockup during API calls)
- async_mode picked via SOCKETIO_ASYNC_MODE env var, default "threading".
  The previous sys.modules auto-detect false-positived because
  flask_socketio / python-engineio import eventlet on load whenever it is
  installed, pinning a non-monkey-patched eventlet server in front of
  blocking stdlib I/O.

Real-time dashboard updates
- New content_received WebSocket event fires when the API accepts a
  submission; UI renders a "Processing…" row immediately and updates it
  in place when moderation_update arrives.
- Fix stale updated_at timestamp in moderation_update payloads.

Batched AI rule evaluation
- Replace the per-rule parallel fan-out (N OpenAI calls) with a single
  batched call evaluating all AI-prompt rules at once. On a 6-rule
  default project this drops safe-content cost from 5-12s to ~1-4s.
- response_format=json_object + max_completion_tokens=4000 so the model
  emits strict JSON and stops at the close-brace.
- OPENAI_REASONING_EFFORT config: minimal/low/medium/high for gpt-5 /
  o-series; accepts "none" on gpt-5.4. Works around the gpt-5-nano case
  where reasoning tokens ate the full output budget and returned empty
  responses.

Client errors
- Malformed / non-UTF-8 JSON bodies now return 400 with a clear message
  about UTF-8 encoding instead of a generic 500.

Logging
- Single INFO line per moderation: the existing
  "Content <id>: decision in Zs" summary.
- Per-stage timing, websocket scheduling, and batched-AI call details
  moved to DEBUG so they are available when diagnosing latency without
  cluttering steady-state logs.

Author: Bentlybro

app/__init__.py

@@ -5,22 +5,53 @@
 
 import sentry_sdk
 from flask import Flask
+from flask_limiter import Limiter
+from flask_limiter.util import get_remote_address
 from flask_login import LoginManager
 from flask_socketio import SocketIO
 from flask_sqlalchemy import SQLAlchemy
 from flask_talisman import Talisman
 from flask_wtf.csrf import CSRFProtect
 
-from config.config import config
+from config.config import config, validate_production_config
+
+# SocketIO async mode selection.
+#
+# Default is 'threading' so the Werkzeug dev server (run.py) and any plain
+# WSGI host work correctly. Production uses eventlet — but ONLY when wsgi.py
+# has run first and called eventlet.monkey_patch() + psycogreen.patch_psycopg().
+# wsgi.py signals that by setting SOCKETIO_ASYNC_MODE=eventlet in the
+# environment before importing this module.
+#
+# DO NOT auto-detect from sys.modules. flask_socketio / python-engineio
+# import eventlet at load time whenever it's installed, which would false-
+# positive the detection and put a non-monkey-patched eventlet server in
+# front of blocking stdlib I/O (that was the "site locks up during API
+# requests" bug).
+_SOCKETIO_ASYNC_MODE = os.environ.get('SOCKETIO_ASYNC_MODE', 'threading')
 
 # SQLAlchemy - database interface
 db = SQLAlchemy()
 login_manager = LoginManager()
 socketio = SocketIO(cors_allowed_origins="*")
 csrf = CSRFProtect()
 
+# Rate limiter. Uses in-memory storage; when scaling to multiple workers or
+# processes, set storage_uri to a Redis URL so limits are shared. Per-route
+# limits are declared in the route files with @limiter.limit(...).
+limiter = Limiter(
+    key_func=get_remote_address,
+    default_limits=[],
+    storage_uri="memory://",
+    strategy="fixed-window",
+)
+
 
 def create_app(config_name: str = 'default') -> Flask:
+    # Fail fast in production if required secrets are missing
+    if config_name == 'production':
+        validate_production_config()
+
     app = Flask(__name__)
     app.config.from_object(config[config_name])
 
@@ -77,9 +108,11 @@ def before_send(event, hint):
     # Initialize SocketIO with increased timeouts to handle browser tab throttling
     # ping_timeout: Time to wait for client response before considering connection dead
     # ping_interval: Time between server pings to check client connection
+    # async_mode: auto-detected above — 'eventlet' under gunicorn, 'threading' under werkzeug dev
+    app.logger.info(f"Initialising SocketIO with async_mode={_SOCKETIO_ASYNC_MODE}")
     socketio.init_app(
         app,
-        async_mode='threading',
+        async_mode=_SOCKETIO_ASYNC_MODE,
         ping_timeout=120,  # Increased from default 60s to 2 minutes
         ping_interval=25,  # Keep default 25s interval
         logger=False,      # Disable SocketIO logger to reduce noise
@@ -106,6 +139,9 @@ def filtered_werkzeug_log(msg, *args, **kwargs):
     # Initialize CSRF protection
     csrf.init_app(app)
 
+    # Initialize rate limiter (limits declared on individual routes)
+    limiter.init_app(app)
+
     # Configure CSRF header names for AJAX requests
     app.config['WTF_CSRF_HEADERS'] = ['X-CSRFToken', 'X-CSRF-Token']
 

app/routes/api.py

@@ -3,7 +3,9 @@
 from typing import Callable
 
 from flask import Blueprint, current_app, jsonify, render_template, request
+from flask_limiter.util import get_remote_address
 
+from app import limiter
 from app.schemas import ContentListRequest, ModerateContentRequest
 from app.services.database_service import db_service
 from app.services.error_tracker import error_tracker
@@ -19,6 +21,19 @@
 api_bp = Blueprint('api', __name__)
 
 
+def _api_rate_limit_key() -> str:
+    """Rate-limit key for /api/moderate.
+
+    Prefers the API key id (set by @require_api_key) so a leaked key cannot be
+    replayed at unbounded QPS from many IPs. Falls back to remote address when
+    the key hasn't been resolved yet (malformed/missing key -> 401 path).
+    """
+    api_key = getattr(request, 'api_key', None)
+    if api_key is not None:
+        return f"api_key:{api_key.id}"
+    return get_remote_address()
+
+
 def require_api_key(f: Callable) -> Callable:
     """Decorator to require valid API key for API endpoints"""
     @wraps(f)
@@ -52,6 +67,7 @@ async def decorated_function(*args, **kwargs):
 
 
 @api_bp.route('/moderate', methods=['POST'])
+@limiter.limit("120 per minute; 5000 per hour", key_func=_api_rate_limit_key)
 @require_api_key
 @validate_json_request(ModerateContentRequest)
 @handle_api_error
@@ -75,7 +91,7 @@ async def moderate_content(validated_data=None):
 
     max_content_size = 5000000  # 5MB limit (increased from 1MB)
     content_size_kb = len(content_data) // 1000
-    current_app.logger.info(f'Content moderation request: {content_size_kb}KB')
+    current_app.logger.debug(f'Content moderation request: {content_size_kb}KB')
 
     if len(content_data) > max_content_size:
         current_app.logger.warning(f'Content too large: {content_size_kb}KB > {max_content_size // 1000}KB')
@@ -141,6 +157,14 @@ async def moderate_content(validated_data=None):
 
     # Start moderation process
     moderation_orchestrator = ModerationOrchestrator()
+
+    # Emit "content_received" before the (potentially slow) moderation pass
+    # so the dashboard can render a pending row immediately. Fetch the fresh
+    # record so the notifier has real timestamps.
+    created_content = await db_service.get_content_by_id(content_id)
+    if created_content is not None:
+        moderation_orchestrator.websocket_notifier.send_content_created(created_content)
+
     result = await moderation_orchestrator.moderate_content(
         content_id, request_start_time)
 

app/routes/auth.py

@@ -4,6 +4,7 @@
 from flask import Blueprint, current_app, flash, jsonify, redirect, render_template, request, url_for
 from flask_login import current_user, login_required, login_user, logout_user
 
+from app import limiter
 from app.models.system_settings import SystemSettings
 from app.services.database_service import db_service
 
@@ -12,6 +13,7 @@
 
 
 @auth_bp.route('/login', methods=['GET', 'POST'])
+@limiter.limit("10 per minute; 50 per hour", methods=["POST"])
 async def login():
     # Redirect to dashboard if already logged in
     if current_user.is_authenticated:
@@ -100,6 +102,7 @@ async def login():
 
 
 @auth_bp.route('/register', methods=['GET', 'POST'])
+@limiter.limit("5 per hour; 20 per day", methods=["POST"])
 async def register():
     # Redirect to dashboard if already logged in
     if current_user.is_authenticated:
@@ -294,6 +297,7 @@ async def google_callback():
 
         google_id = user_info.get('sub')
         email = user_info.get('email')
+        email_verified = user_info.get('email_verified') is True
 
         if not google_id or not email:
             flash('Invalid user information from Google', 'error')
@@ -308,11 +312,18 @@ async def google_callback():
             flash('Login successful!', 'success')
             return redirect(url_for('dashboard.index'))
 
+        # Reject unverified emails before any account-linking or account-creation branch.
+        # An attacker who controls a Google workspace can register arbitrary unverified
+        # addresses and would otherwise hijack a matching local account here.
+        if not email_verified:
+            flash('Your Google account email must be verified before you can sign in here.', 'error')
+            return redirect(url_for('auth.login'))
+
         # Check if user exists with this email
         user = await db_service.get_user_by_email(email.lower())
 
         if user:
-            # Link Google account to existing user
+            # Link Google account to existing user (email is verified per check above)
             await db_service.link_google_account(user.id, google_id)
             login_user(user)
             flash('Google account linked successfully!', 'success')
@@ -357,6 +368,7 @@ async def google_callback():
 
 @auth_bp.route('/change-password', methods=['POST'])
 @login_required
+@limiter.limit("10 per hour")
 async def change_password():
     # Check if this is an AJAX request by looking for specific headers or content type
     # is_ajax = request.headers.get('X-Requested-With') == 'XMLHttpRequest' or \

app/routes/dashboard.py

@@ -196,10 +196,10 @@ async def create_rule(project_id):
     """Create new moderation rule"""
     project = Project.query.filter_by(id=project_id).first_or_404()
 
-    # Check if user has access to this project
-    if not project.is_member(current_user.id):
-        flash('You do not have access to this project', 'error')
-        return redirect(url_for('dashboard.projects'))
+    # Mutating rules requires owner/admin role; plain members are read-only
+    if not project.can_manage_members(current_user.id):
+        flash('You do not have permission to create rules for this project', 'error')
+        return redirect(url_for('dashboard.project_rules', project_id=project_id))
 
     if request.method == 'POST':
         name = request.form.get('name')
@@ -251,8 +251,8 @@ async def update_rule(project_id, rule_id):
     """Update existing moderation rule"""
     project = Project.query.filter_by(id=project_id).first_or_404()
 
-    # Check if user has access to this project
-    if not project.is_member(current_user.id):
+    # Mutating rules requires owner/admin role; plain members are read-only
+    if not project.can_manage_members(current_user.id):
         return jsonify({'success': False, 'error': 'Access denied'}), 403
     rule = ModerationRule.query.filter_by(
         id=rule_id, project_id=project.id).first_or_404()
@@ -301,8 +301,8 @@ async def toggle_rule(project_id, rule_id):
     """Toggle rule active/inactive status"""
     project = Project.query.filter_by(id=project_id).first_or_404()
 
-    # Check if user has access to this project
-    if not project.is_member(current_user.id):
+    # Mutating rules requires owner/admin role; plain members are read-only
+    if not project.can_manage_members(current_user.id):
         return jsonify({'success': False, 'error': 'Access denied'}), 403
     rule = ModerationRule.query.filter_by(
         id=rule_id, project_id=project.id).first_or_404()
@@ -337,8 +337,8 @@ async def delete_rule(project_id, rule_id):
     """Delete moderation rule"""
     project = Project.query.filter_by(id=project_id).first_or_404()
 
-    # Check if user has access to this project
-    if not project.is_member(current_user.id):
+    # Deleting rules requires owner/admin role; plain members are read-only
+    if not project.can_manage_members(current_user.id):
         return jsonify({'success': False, 'error': 'Access denied'}), 403
     rule = ModerationRule.query.filter_by(
         id=rule_id, project_id=project.id).first_or_404()
@@ -464,8 +464,8 @@ async def toggle_api_key(project_id, key_id):
     """Toggle API key active/inactive status"""
     project = Project.query.filter_by(id=project_id).first_or_404()
 
-    # Check if user has access to this project
-    if not project.is_member(current_user.id):
+    # Mutating API keys requires owner/admin role; plain members are read-only
+    if not project.can_manage_members(current_user.id):
         return jsonify({'success': False, 'error': 'Access denied'}), 403
     api_key = APIKey.query.filter_by(
         id=key_id, project_id=project.id).first_or_404()
@@ -500,8 +500,8 @@ async def delete_api_key(project_id, key_id):
     """Delete API key"""
     project = Project.query.filter_by(id=project_id).first_or_404()
 
-    # Check if user has access to this project
-    if not project.is_member(current_user.id):
+    # Deleting API keys requires owner/admin role; plain members are read-only
+    if not project.can_manage_members(current_user.id):
         return jsonify({'success': False, 'error': 'Access denied'}), 403
     api_key = APIKey.query.filter_by(
         id=key_id, project_id=project.id).first_or_404()
@@ -573,10 +573,10 @@ async def update_project(project_id):
     """Update project information"""
     project = Project.query.filter_by(id=project_id).first_or_404()
 
-    # Check if user has access to this project
-    if not project.is_member(current_user.id):
-        flash('You do not have access to this project', 'error')
-        return redirect(url_for('dashboard.projects'))
+    # Updating project identity requires owner/admin role; plain members are read-only
+    if not project.can_manage_members(current_user.id):
+        flash('You do not have permission to modify project settings', 'error')
+        return redirect(url_for('dashboard.project_settings', project_id=project_id))
 
     name = request.form.get('name')
     description = request.form.get('description', '')

app/routes/manual_review.py

@@ -500,7 +500,7 @@ async def delete_user_data(user_id):
         api_user = APIUser.query.get_or_404(user_id)
 
         # Check if user has permission (admin or project owner)
-        if not current_user.is_admin and not api_user.project.is_owner(current_user.id):
+        if not current_user.is_admin and api_user.project.user_id != current_user.id:
             flash('You do not have permission to delete user data', 'error')
             return redirect(url_for('manual_review.api_user_detail', user_id=user_id))