Content Security Policy: Complete Implementation Guide

Complete Production-Ready CSP System:

#!/usr/bin/env python3
"""
Production-ready Content Security Policy (CSP) System
Comprehensive CSP header management with violation reporting and monitoring
"""

import json
import logging
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class CSPDirective(Enum):
    """CSP directive types."""
    DEFAULT_SRC = "default-src"
    SCRIPT_SRC = "script-src"
    STYLE_SRC = "style-src"
    IMG_SRC = "img-src"
    FONT_SRC = "font-src"
    CONNECT_SRC = "connect-src"
    FRAME_SRC = "frame-src"
    FRAME_ANCESTORS = "frame-ancestors"
    BASE_URI = "base-uri"
    FORM_ACTION = "form-action"
    UPGRADE_INSECURE_REQUESTS = "upgrade-insecure-requests"
    BLOCK_ALL_MIXED_CONTENT = "block-all-mixed-content"


@dataclass
class CSPPolicy:
    """CSP policy configuration."""
    directives: Dict[str, List[str]]
    report_uri: Optional[str] = None
    report_only: bool = False


class CSPManager:
    """Content Security Policy manager."""
    
    def __init__(self, policy: Optional[CSPPolicy] = None):
        """Initialize CSP manager.
        
        Args:
            policy: CSP policy configuration
        """
        if policy is None:
            policy = self._default_policy()
        self.policy = policy
        self.violations: List[Dict] = []
    
    def _default_policy(self) -> CSPPolicy:
        """Create default CSP policy.
        
        Returns:
            Default CSP policy
        """
        return CSPPolicy(
            directives={
                'default-src': ["'self'"],
                'script-src': ["'self'", "'strict-dynamic'"],
                'style-src': ["'self'", "'unsafe-inline'"],
                'img-src': ["'self'", "data:", "https:"],
                'font-src': ["'self'", "data:"],
                'connect-src': ["'self'"],
                'frame-src': ["'none'"],
                'frame-ancestors': ["'none'"],
                'base-uri': ["'self'"],
                'form-action': ["'self'"],
                'upgrade-insecure-requests': [],
            },
            report_uri='/csp-report',
            report_only=False
        )
    
    def generate_header(self) -> str:
        """Generate CSP header string.
        
        Returns:
            CSP header value
        """
        parts = []
        
        for directive, sources in self.policy.directives.items():
            if sources:
                parts.append(f"{directive} {' '.join(sources)}")
            else:
                parts.append(directive)
        
        csp_string = '; '.join(parts)
        
        if self.policy.report_uri:
            csp_string += f"; report-uri {self.policy.report_uri}"
        
        return csp_string
    
    def strict_policy(self) -> CSPPolicy:
        """Create strict CSP policy.
        
        Returns:
            Strict CSP policy
        """
        return CSPPolicy(
            directives={
                'default-src': ["'none'"],
                'script-src': ["'self'", "'nonce-{nonce}'"],  # Nonce-based
                'style-src': ["'self'"],
                'img-src': ["'self'", "data:", "https:"],
                'font-src': ["'self'"],
                'connect-src': ["'self'"],
                'frame-src': ["'none'"],
                'frame-ancestors': ["'none'"],
                'base-uri': ["'self'"],
                'form-action': ["'self'"],
                'upgrade-insecure-requests': [],
                'block-all-mixed-content': [],
            },
            report_uri='/csp-report'
        )
    
    def record_violation(self, violation: Dict):
        """Record CSP violation.
        
        Args:
            violation: Violation report data
        """
        violation['timestamp'] = datetime.utcnow().isoformat()
        self.violations.append(violation)
        logger.warning(f"CSP violation: {violation.get('violated-directive')}")
    
    def analyze_violations(self) -> Dict:
        """Analyze CSP violations.
        
        Returns:
            Analysis results
        """
        if not self.violations:
            return {'total': 0, 'by_directive': {}, 'recommendations': []}
        
        by_directive = {}
        sources = []
        
        for violation in self.violations:
            directive = violation.get('violated-directive', 'unknown')
            source = violation.get('blocked-uri', 'unknown')
            
            if directive not in by_directive:
                by_directive[directive] = 0
            by_directive[directive] += 1
            
            sources.append(source)
        
        # Generate recommendations
        recommendations = []
        unique_sources = list(set(sources))
        
        for directive, count in by_directive.items():
            if count > 10:  # Frequent violations
                directive_sources = [
                    v.get('blocked-uri') for v in self.violations
                    if v.get('violated-directive') == directive
                ]
                unique_dirs = list(set(directive_sources))[:5]  # Top 5 sources
                
                if len(unique_dirs) < 10:  # Few unique sources - can whitelist
                    recommendations.append({
                        'directive': directive,
                        'action': 'Add to allowlist',
                        'sources': unique_dirs,
                        'violations': count
                    })
        
        return {
            'total': len(self.violations),
            'by_directive': by_directive,
            'recommendations': recommendations
        }


# Flask integration
from flask import Flask, request, jsonify

app = Flask(__name__)
csp_manager = CSPManager()


@app.after_request
def set_csp_header(response):
    """Set CSP header on all responses."""
    csp_header = csp_manager.generate_header()
    header_name = 'Content-Security-Policy-Report-Only' if csp_manager.policy.report_only else 'Content-Security-Policy'
    response.headers[header_name] = csp_header
    
    # Also set security headers
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    
    return response


@app.route('/csp-report', methods=['POST'])
def csp_report():
    """CSP violation reporting endpoint."""
    try:
        data = request.get_json()
        
        if data and 'csp-report' in data:
            violation = data['csp-report']
            csp_manager.record_violation(violation)
            
            logger.warning(
                f"CSP violation: {violation.get('violated-directive')} "
                f"blocked {violation.get('blocked-uri')}"
            )
        
        return jsonify({'status': 'received'}), 204
    except Exception as e:
        logger.error(f"Error processing CSP report: {e}")
        return jsonify({'error': 'Invalid report'}), 400


@app.route('/api/csp/analyze', methods=['GET'])
def analyze_csp():
    """Analyze CSP violations."""
    analysis = csp_manager.analyze_violations()
    return jsonify(analysis)


@app.route('/api/csp/violations', methods=['GET'])
def get_violations():
    """Get recent CSP violations."""
    limit = request.args.get('limit', 100, type=int)
    violations = csp_manager.violations[-limit:]
    return jsonify({'violations': violations})


# Express.js integration example
EXPRESS_CSP_EXAMPLE = """
const express = require('express');
const helmet = require('helmet');

const app = express();

app.use(helmet.contentSecurityPolicy({
    directives: {
        defaultSrc: ["'self'"],
        scriptSrc: ["'self'", "'strict-dynamic'"],
        styleSrc: ["'self'", "'unsafe-inline'"],
        imgSrc: ["'self'", "data:", "https:"],
        fontSrc: ["'self'", "data:"],
        connectSrc: ["'self'"],
        frameSrc: ["'none'"],
        frameAncestors: ["'none'"],
        baseUri: ["'self'"],
        formAction: ["'self'"],
        upgradeInsecureRequests: []
    },
    reportOnly: false,
    reportUri: '/csp-report'
}));

app.post('/csp-report', express.json(), (req, res) => {
    const report = req.body['csp-report'];
    console.log('CSP violation:', report);
    res.status(204).send();
});

app.listen(3000);
"""

if __name__ == '__main__':
    # Example usage
    manager = CSPManager()
    print("CSP Header:", manager.generate_header())

Content-Security-Policy: default-src 'self'; report-uri /csp-report-endpoint;
Content-Security-Policy-Report-Only: default-src 'self';

# Analyze CSP violations
def analyze_csp_violations(reports):
    """Analyze CSP violation reports."""
    violations = {}
    
    for report in reports:
        directive = report.get('violated-directive')
        source = report.get('blocked-uri')
        
        if directive not in violations:
            violations[directive] = []
        violations[directive].append(source)
    
    # Generate recommendations
    recommendations = []
    for directive, sources in violations.items():
        if len(set(sources)) < 5:  # Few unique sources
            recommendations.append(f"Add to {directive}: {', '.join(set(sources))}")
    
    return recommendations