Add comprehensive test and CI coverage

.github/workflows/tests.yml

@@ -0,0 +1,47 @@
+name: Tests
+
+on:
+  push:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  unit-and-route-tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+      - run: python -m pip install --requirement requirements.txt
+      - name: Check Python syntax
+        run: python -m compileall -q app review tests
+      - name: Run fast tests
+        run: pytest --cov=app --cov=review --cov-branch --cov-report=term-missing --cov-fail-under=75
+
+  mongodb-integration-tests:
+    runs-on: ubuntu-latest
+    services:
+      mongodb:
+        image: mongo:7
+        ports:
+          - 27017:27017
+        options: >-
+          --health-cmd "mongosh --quiet --eval 'db.runCommand({ ping: 1 })'"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      TEST_MONGODB_URL: mongodb://127.0.0.1:27017
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+      - run: python -m pip install --requirement requirements.txt
+      - name: Run tests against MongoDB
+        run: pytest

app/__init__.py

@@ -17,20 +17,25 @@
 from app.services.view import register_filters
 
 
-def create_app(config_path=None):
-    """Create and configure the Flask application."""
+def create_app(config_path=None, config=None):
+    """Create and configure the Flask application.
+
+    ``config`` allows tests and other callers to supply configuration without
+    creating or modifying the production JSON file.  Production continues to
+    load ``config_path`` when no configuration mapping is supplied.
+    """
     app = Flask(__name__,
                 template_folder='../templates',
                 static_folder='../static')
     app.url_map.strict_slashes = False
 
-    # Load configuration
-    if config_path is None:
-        config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)),
-                                   'config', 'config.json')
+    if config is None:
+        if config_path is None:
+            config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)),
+                                       'config', 'config.json')
 
-    with open(config_path, 'r') as f:
-        config = json.load(f)
+        with open(config_path, 'r') as f:
+            config = json.load(f)
 
     # Configure Flask app
     app.config['SECRET_KEY'] = config['Session']['SecretKey']

app/services/database.py

@@ -18,7 +18,9 @@ def init_db(app, config):
         mongo_url = config.get('URL', 'mongodb://127.0.0.1:27017')
         database_name = config.get('Name', 'crackmesone')
 
-        mongo_client = MongoClient(mongo_url)
+        # Tests can inject a disposable compatible client (for example,
+        # mongomock) while production always constructs a real MongoClient.
+        mongo_client = config.get('Client') or MongoClient(mongo_url)
         # Verify connection
         mongo_client.admin.command('ping')
 

app/services/passhash.py

@@ -23,7 +23,7 @@ def match_string(hashed: str, password: str) -> bool:
     """Check if the password matches the hash."""
     try:
         return bcrypt.checkpw(password.encode('utf-8'), hashed.encode('utf-8'))
-    except (ValueError, TypeError):
+    except (AttributeError, ValueError, TypeError):
         return False
 
 

requirements.txt

@@ -12,6 +12,7 @@ resend==2.0.0
 rustyzipper==1.0.6
 
 # Testing
+setuptools<81  # mongomock 4.1.2 imports pkg_resources
 pytest==7.4.3
 pytest-cov==4.1.0
 mongomock==4.1.2

review/auth.py

@@ -0,0 +1,87 @@
+"""Reviewer authentication, authorization, and CSRF helpers.
+
+This module deliberately owns only reviewer-session security.  Main-site user
+authentication remains separate.
+"""
+
+from functools import wraps
+import hashlib
+import os
+
+from flask import abort, redirect, request, session, url_for
+
+
+REVIEWER_SESSION_KEY = '_reviewer_user'
+REVIEWER_ADMIN_KEY = '_reviewer_is_admin'
+REVIEWER_CSRF_KEY = '_reviewer_csrf_token'
+
+_users = {}
+
+
+def configure(users):
+    """Use the reviewer credential mapping loaded by the reviewer package."""
+    global _users
+    _users = users
+
+
+def hash_string(input_string):
+    """Return a SHA-256 hexadecimal password digest input."""
+    return hashlib.sha256(input_string.encode('utf-8')).hexdigest()
+
+
+def get_current_reviewer():
+    """Return the authenticated reviewer, or ``None`` for a stale session."""
+    username = session.get(REVIEWER_SESSION_KEY)
+    if not username or username not in _users:
+        return None
+    return {
+        'username': username,
+        'is_admin': _users[username].get('is_admin', False),
+    }
+
+
+def clear_reviewer_session():
+    """Remove reviewer authentication without touching main-site auth."""
+    session.pop(REVIEWER_SESSION_KEY, None)
+    session.pop(REVIEWER_ADMIN_KEY, None)
+
+
+def token_required(view):
+    """Require a current reviewer account."""
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        current_user = get_current_reviewer()
+        if not current_user:
+            clear_reviewer_session()
+            return redirect(url_for('reviewer.login'))
+        return view(current_user, *args, **kwargs)
+    return decorated
+
+
+def admin_required(view):
+    """Require a current reviewer account with administrator privileges."""
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        current_user = get_current_reviewer()
+        if not current_user:
+            clear_reviewer_session()
+            return redirect(url_for('reviewer.login'))
+        if not current_user['is_admin']:
+            abort(403)
+        return view(current_user, *args, **kwargs)
+    return decorated
+
+
+def generate_csrf_token():
+    """Generate or retrieve the reviewer-specific CSRF token."""
+    if REVIEWER_CSRF_KEY not in session:
+        session[REVIEWER_CSRF_KEY] = hashlib.sha256(os.urandom(32)).hexdigest()
+    return session[REVIEWER_CSRF_KEY]
+
+
+def validate_csrf_token():
+    """Reject a missing or mismatched reviewer CSRF token."""
+    token = request.form.get('csrf_token')
+    expected = session.get(REVIEWER_CSRF_KEY)
+    if not token or not expected or token != expected:
+        abort(403, description='CSRF token validation failed')

review/routes.py

@@ -12,11 +12,9 @@
     Blueprint, render_template, request, redirect,
     url_for, abort, send_file, session, jsonify
 )
-from functools import wraps
 from html import escape as html_escape
 import datetime
 from datetime import timezone
-import hashlib
 import json
 import os
 import random
@@ -31,6 +29,19 @@
 from bson.objectid import ObjectId
 
 from review.logger import log_reviewer_operation
+from review.auth import (
+    REVIEWER_ADMIN_KEY,
+    REVIEWER_CSRF_KEY,
+    REVIEWER_SESSION_KEY,
+    admin_required,
+    clear_reviewer_session,
+    configure as configure_reviewer_auth,
+    generate_csrf_token,
+    get_current_reviewer,
+    hash_string,
+    token_required,
+    validate_csrf_token,
+)
 from app.services.crypto import get_obfuscation_salt
 from app.services.view import is_valid_hexid
 
@@ -56,11 +67,6 @@
 users = {}
 USERS_FILE = os.path.join(os.path.dirname(__file__), 'users.json')
 
-# Session keys for reviewer authentication (prefixed to avoid conflicts)
-REVIEWER_SESSION_KEY = '_reviewer_user'
-REVIEWER_ADMIN_KEY = '_reviewer_is_admin'
-REVIEWER_CSRF_KEY = '_reviewer_csrf_token'
-
 # Archive password for approved submissions
 ARCHIVE_PASSWORD = 'crackmes.one'
 
@@ -104,6 +110,8 @@ def init_reviewer(app):
         with open(USERS_FILE, 'r') as f:
             users.update(json.load(f))
 
+    configure_reviewer_auth(users)
+
 
 def save_users():
     """
@@ -116,110 +124,6 @@ def save_users():
         json.dump(users, f, indent=2)
 
 
-# =============================================================================
-# Authentication Helpers
-# =============================================================================
-
-def hash_string(input_string):
-    """
-    Hash a string using SHA256.
-
-    Args:
-        input_string: Plain text string to hash
-
-    Returns:
-        Hexadecimal string representation of the SHA256 hash
-    """
-    return hashlib.sha256(input_string.encode('utf-8')).hexdigest()
-
-
-def get_current_reviewer():
-    """
-    Get the current authenticated reviewer from session.
-
-    Returns:
-        Dict with 'username' and 'is_admin' keys if authenticated,
-        None if not authenticated or user no longer exists.
-    """
-    username = session.get(REVIEWER_SESSION_KEY)
-    if not username or username not in users:
-        return None
-    return {
-        'username': username,
-        'is_admin': users[username].get('is_admin', False)
-    }
-
-
-def clear_reviewer_session():
-    """Remove reviewer authentication from session."""
-    session.pop(REVIEWER_SESSION_KEY, None)
-    session.pop(REVIEWER_ADMIN_KEY, None)
-
-
-def token_required(f):
-    """
-    Decorator requiring reviewer authentication.
-
-    Redirects to login page if not authenticated. Passes current_user
-    dict as first argument to the decorated function.
-    """
-    @wraps(f)
-    def decorated(*args, **kwargs):
-        current_user = get_current_reviewer()
-        if not current_user:
-            clear_reviewer_session()
-            return redirect(url_for('reviewer.login'))
-        return f(current_user, *args, **kwargs)
-    return decorated
-
-
-def admin_required(f):
-    """
-    Decorator requiring admin authentication.
-
-    Redirects to login if not authenticated, returns 403 if authenticated
-    but not an admin. Passes current_user dict as first argument.
-    """
-    @wraps(f)
-    def decorated(*args, **kwargs):
-        current_user = get_current_reviewer()
-        if not current_user:
-            clear_reviewer_session()
-            return redirect(url_for('reviewer.login'))
-        if not current_user['is_admin']:
-            abort(403)
-        return f(current_user, *args, **kwargs)
-    return decorated
-
-
-# =============================================================================
-# CSRF Protection
-# =============================================================================
-
-def generate_csrf_token():
-    """
-    Generate or retrieve CSRF token for the current session.
-
-    Returns:
-        32-byte hex string CSRF token
-    """
-    if REVIEWER_CSRF_KEY not in session:
-        session[REVIEWER_CSRF_KEY] = hashlib.sha256(os.urandom(32)).hexdigest()
-    return session[REVIEWER_CSRF_KEY]
-
-
-def validate_csrf_token():
-    """
-    Validate CSRF token from form submission.
-
-    Aborts with 403 if token is missing or invalid.
-    """
-    token = request.form.get('csrf_token')
-    expected = session.get(REVIEWER_CSRF_KEY)
-    if not token or not expected or token != expected:
-        abort(403, description="CSRF token validation failed")
-
-
 @reviewer_bp.context_processor
 def inject_csrf_token():
     """Make csrf_token function available in all templates."""