feat: Enhance security and input validation with environment-based credentials, CSRF protection, and structured logging

Author: fabriziosalmi

backend/Dockerfile

@@ -4,8 +4,7 @@ WORKDIR /app
 
 # Install dependencies with specific order to handle dependencies correctly
 COPY requirements.txt .
-RUN pip install --no-cache-dir werkzeug==2.2.3 && \
-    pip install --no-cache-dir -r requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
 
 # Install curl for healthcheck (Docker CLI removed for security)
 RUN apt-get update && \

backend/app/app.py

@@ -22,6 +22,16 @@
 from contextlib import contextmanager
 import signal
 import sys
+from pythonjsonlogger import jsonlogger
+from marshmallow import Schema, fields, validate, ValidationError
+
+class IPSchema(Schema):
+    ip = fields.String(required=True)
+    description = fields.String(missing='')
+
+class DomainSchema(Schema):
+    domain = fields.String(required=True, validate=validate.Regexp(r'^(\*\.)?(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])\.)+([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]*[A-Za-z0-9])$'))
+    description = fields.String(missing='')
 
 # Rate limiting setup
 auth_attempts = defaultdict(list)
@@ -53,15 +63,13 @@
     os.makedirs('logs', exist_ok=True)
     log_path = 'logs/backend.log'
 
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-    handlers=[
-        logging.FileHandler(log_path),
-        logging.StreamHandler()
-    ]
-)
 logger = logging.getLogger(__name__)
+logHandler = logging.FileHandler(log_path)
+formatter = jsonlogger.JsonFormatter('%(asctime)s %(name)s %(levelname)s %(message)s')
+logHandler.setFormatter(formatter)
+logger.addHandler(logHandler)
+logger.setLevel(logging.INFO)
+logger.addHandler(logging.StreamHandler())
 
 # Database configuration
 DATABASE_PATH = '/data/secure_proxy.db'
@@ -146,19 +154,31 @@ def init_db():
     )
     ''')
 
-    # Insert default admin user if not exists
-    cursor.execute("SELECT COUNT(*) FROM users WHERE username = 'admin'")
-    if cursor.fetchone()[0] == 0:
-        # Use a known password hash for 'admin' to ensure UI can authenticate
-        admin_password_hash = generate_password_hash('admin')
-        logger.info(f"Creating default admin user with password hash")
-        cursor.execute("INSERT INTO users (username, password) VALUES (?, ?)", 
-                      ('admin', admin_password_hash))
+    # Handle admin user creation/update based on environment variables
+    env_username = os.environ.get('BASIC_AUTH_USERNAME')
+    env_password = os.environ.get('BASIC_AUTH_PASSWORD')
+
+    if env_username and env_password:
+        # Check if user exists
+        cursor.execute("SELECT COUNT(*) FROM users WHERE username = ?", (env_username,))
+        if cursor.fetchone()[0] == 0:
+            logger.info(f"Creating admin user '{env_username}' from environment variables")
+            cursor.execute("INSERT INTO users (username, password) VALUES (?, ?)", 
+                          (env_username, generate_password_hash(env_password)))
+        else:
+            logger.info(f"Updating admin user '{env_username}' from environment variables")
+            cursor.execute("UPDATE users SET password = ? WHERE username = ?", 
+                         (generate_password_hash(env_password), env_username))
     else:
-        # For existing installations, ensure the admin password hash is correct
-        cursor.execute("UPDATE users SET password = ? WHERE username = ?", 
-                     (generate_password_hash('admin'), 'admin'))
-        logger.info("Updated admin password hash to ensure authentication works")
+        # Check if any user exists
+        cursor.execute("SELECT COUNT(*) FROM users")
+        if cursor.fetchone()[0] == 0:
+            # No users and no env vars - generate random credentials
+            gen_username = 'admin'
+            gen_password = secrets.token_urlsafe(16)
+            logger.warning(f"No credentials provided. Created default user '{gen_username}' with password: {gen_password}")
+            cursor.execute("INSERT INTO users (username, password) VALUES (?, ?)", 
+                          (gen_username, generate_password_hash(gen_password)))
 
     # Insert default settings if not exists
     default_settings = [
@@ -434,13 +454,14 @@ def get_ip_blacklist():
 @auth.login_required
 def add_ip_to_blacklist():
     """Add an IP to the blacklist"""
-    data = request.get_json()
-    if not data or 'ip' not in data:
-        return jsonify({"status": "error", "message": "No IP provided"}), 400
+    try:
+        data = IPSchema().load(request.get_json())
+    except ValidationError as err:
+        return jsonify({"status": "error", "message": err.messages}), 400
 
     # Validate IP address format
     ip = data['ip'].strip()
-    description = data.get('description', '')
+    description = data['description']
 
     # Validate CIDR notation or single IP address
     try:
@@ -492,18 +513,13 @@ def get_domain_blacklist():
 @auth.login_required
 def add_domain_to_blacklist():
     """Add a domain to the blacklist"""
-    data = request.get_json()
-    if not data or 'domain' not in data:
-        return jsonify({"status": "error", "message": "No domain provided"}), 400
+    try:
+        data = DomainSchema().load(request.get_json())
+    except ValidationError as err:
+        return jsonify({"status": "error", "message": err.messages}), 400
 
     domain = data['domain'].strip()
-    description = data.get('description', '')
-    
-    # Basic domain validation
-    # Allow wildcard domains (*.example.com) and regular domains
-    domain_pattern = r'^(\*\.)?(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])\.)+([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]*[A-Za-z0-9])$'
-    if not re.match(domain_pattern, domain):
-        return jsonify({"status": "error", "message": "Invalid domain format"}), 400
+    description = data['description']
 
     conn = get_db()
     cursor = conn.cursor()

backend/requirements.txt

@@ -9,4 +9,6 @@ sqlalchemy==2.0.36
 python-dotenv==1.0.1
 pytz==2024.2
 werkzeug==3.1.3
-markupsafe==2.1.5
+markupsafe==2.1.5
+python-json-logger==2.0.7
+marshmallow==3.20.1

docker-compose.yml

@@ -4,7 +4,6 @@ services:
     ports:
       - "8011:8011"
     volumes:
-      - ./ui:/app
       - ./config:/config
       - ./data:/data
       - ./logs:/logs
@@ -14,33 +13,45 @@ services:
     environment:
       - FLASK_ENV=production
       - BACKEND_URL=http://backend:5000
-      - BASIC_AUTH_USERNAME=admin
-      - BASIC_AUTH_PASSWORD=admin
       - REQUEST_TIMEOUT=30
       - MAX_RETRIES=5
       - BACKOFF_FACTOR=1.0
       - RETRY_WAIT_AFTER_STARTUP=10
+      # Credentials must be provided via .env file or environment variables
+      - BASIC_AUTH_USERNAME
+      - BASIC_AUTH_PASSWORD
+      - SECRET_KEY
+    deploy:
+      resources:
+        limits:
+          cpus: '0.50'
+          memory: 512M
     networks:
       - proxy-network
     restart: unless-stopped
 
   backend:
     build: ./backend
     ports:
-      - "5001:5000"  # Map container port 5000 to host port 5001
+      - "5001:5000"
     volumes:
-      - ./backend:/app
       - ./config:/config
       - ./data:/data
       - ./logs:/logs
-      # Docker socket mount removed for security - see SECURITY.md
     environment:
       - FLASK_ENV=production
       - PROXY_HOST=proxy
       - PROXY_PORT=3128
-      - BASIC_AUTH_USERNAME=admin
-      - BASIC_AUTH_PASSWORD=admin
-      - PROXY_CONTAINER_NAME=secure-proxy-proxy-1  # Add container name for restart commands
+      - PROXY_CONTAINER_NAME=secure-proxy-proxy-1
+      # Credentials must be provided via .env file or environment variables
+      - BASIC_AUTH_USERNAME
+      - BASIC_AUTH_PASSWORD
+      - SECRET_KEY
+    deploy:
+      resources:
+        limits:
+          cpus: '0.50'
+          memory: 512M
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
       interval: 5s
@@ -56,11 +67,15 @@ services:
     ports:
       - "3128:3128"
     volumes:
-      # Removed the direct mount of squid.conf
       - ./config:/config
       - ./data:/data
       - ./logs:/var/log/squid
       - squid-cache:/var/spool/squid
+    deploy:
+      resources:
+        limits:
+          cpus: '0.50'
+          memory: 512M
     networks:
       - proxy-network
     restart: unless-stopped

tests/test_security_improvements.py

@@ -0,0 +1,76 @@
+import unittest
+import json
+import os
+import sys
+from unittest.mock import patch, MagicMock
+
+# Add backend to path
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../backend')))
+
+# Set env vars before importing app
+os.environ['BASIC_AUTH_USERNAME'] = 'admin'
+os.environ['BASIC_AUTH_PASSWORD'] = 'admin'
+os.environ['SECRET_KEY'] = 'test_secret'
+
+from app.app import app, init_db, get_db
+
+class SecurityTests(unittest.TestCase):
+    def setUp(self):
+        app.config['TESTING'] = True
+        self.client = app.test_client()
+        
+        # Setup in-memory db
+        with app.app_context():
+            # Mock database path to use in-memory
+            with patch('app.app.DATABASE_PATH', ':memory:'):
+                init_db()
+
+    def test_input_validation_ip(self):
+        # Test invalid IP
+        response = self.client.post('/api/ip-blacklist', 
+            headers={'Authorization': 'Basic YWRtaW46YWRtaW4='}, # admin:admin
+            json={'ip': 'invalid-ip'}
+        )
+        self.assertEqual(response.status_code, 400)
+        
+        # Test valid IP
+        response = self.client.post('/api/ip-blacklist', 
+            headers={'Authorization': 'Basic YWRtaW46YWRtaW4='},
+            json={'ip': '1.2.3.4', 'description': 'test'}
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_input_validation_domain(self):
+        # Test invalid domain
+        response = self.client.post('/api/domain-blacklist', 
+            headers={'Authorization': 'Basic YWRtaW46YWRtaW4='},
+            json={'domain': '-invalid-domain'}
+        )
+        self.assertEqual(response.status_code, 400)
+        
+        # Test valid domain
+        response = self.client.post('/api/domain-blacklist', 
+            headers={'Authorization': 'Basic YWRtaW46YWRtaW4='},
+            json={'domain': 'example.com', 'description': 'test'}
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_settings_update(self):
+        # Test update setting
+        response = self.client.put('/api/settings/log_level', 
+            headers={'Authorization': 'Basic YWRtaW46YWRtaW4='},
+            json={'value': 'debug'}
+        )
+        self.assertEqual(response.status_code, 200)
+        
+        # Test invalid setting value
+        response = self.client.put('/api/settings/log_level', 
+            headers={'Authorization': 'Basic YWRtaW46YWRtaW4='},
+            json={'value': 'invalid'}
+        )
+        # validate_setting returns False, so it should be 400
+        # In app.py: if not validate_setting(...): return ..., 400
+        self.assertEqual(response.status_code, 400)
+
+if __name__ == '__main__':
+    unittest.main()