File Upload Security

import os
import magic
import hashlib
from PIL import Image
from werkzeug.utils import secure_filename
from uuid import uuid4

# Configuration
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'pdf'}
ALLOWED_MIMES = {
    'image/png': 'png',
    'image/jpeg': 'jpg', 
    'image/gif': 'gif',
    'application/pdf': 'pdf'
}
MAX_FILE_SIZE = 5 * 1024 * 1024  # 5MB
UPLOAD_DIR = '/var/uploads'  # Outside webroot!

class FileUploadError(Exception):
    pass

def validate_and_save_file(file_storage):
    """Comprehensive file upload validation"""
    
    # 1. Check if file exists
    if not file_storage or not file_storage.filename:
        raise FileUploadError("No file provided")
    
    # 2. Check file size (before reading entire file)
    file_storage.seek(0, os.SEEK_END)
    size = file_storage.tell()
    file_storage.seek(0)
    
    if size > MAX_FILE_SIZE:
        raise FileUploadError(f"File too large (max {MAX_FILE_SIZE // 1024 // 1024}MB)")
    
    if size == 0:
        raise FileUploadError("Empty file")
    
    # 3. Validate extension (from original filename)
    original_name = secure_filename(file_storage.filename)
    ext = original_name.rsplit('.', 1)[-1].lower() if '.' in original_name else ''
    
    if ext not in ALLOWED_EXTENSIONS:
        raise FileUploadError(f"Extension not allowed: {ext}")
    
    # 4. Validate MIME type using magic bytes (not Content-Type header!)
    file_content = file_storage.read()
    file_storage.seek(0)
    
    detected_mime = magic.from_buffer(file_content, mime=True)
    if detected_mime not in ALLOWED_MIMES:
        raise FileUploadError(f"File type not allowed: {detected_mime}")
    
    # 5. Ensure extension matches detected type
    expected_ext = ALLOWED_MIMES[detected_mime]
    if ext not in [expected_ext, 'jpg'] or (ext == 'jpg' and expected_ext != 'jpg'):
        # Handle jpg/jpeg case
        if not (ext in ['jpg', 'jpeg'] and expected_ext in ['jpg', 'jpeg']):
            raise FileUploadError("Extension doesn't match file type")
    
    # 6. Additional validation for images - re-process to strip metadata
    if detected_mime.startswith('image/'):
        try:
            img = Image.open(file_storage)
            img.verify()  # Verify it's a valid image
            file_storage.seek(0)
            
            # Re-open and save to strip EXIF/metadata
            img = Image.open(file_storage)
            clean_img = Image.new(img.mode, img.size)
            clean_img.putdata(list(img.getdata()))
        except Exception as e:
            raise FileUploadError(f"Invalid image file: {e}")
    
    # 7. Generate safe filename (never use original)
    safe_name = f"{uuid4().hex}.{expected_ext}"
    
    # 8. Save to upload directory
    save_path = os.path.join(UPLOAD_DIR, safe_name)
    
    # Double-check path traversal protection
    if not os.path.abspath(save_path).startswith(os.path.abspath(UPLOAD_DIR)):
        raise FileUploadError("Invalid save path")
    
    if detected_mime.startswith('image/'):
        clean_img.save(save_path)
    else:
        with open(save_path, 'wb') as f:
            f.write(file_content)
    
    return safe_name

# Nginx - Disable execution in upload directory
location /uploads {
    # Serve files as downloads, not executable
    add_header Content-Disposition "attachment";
    
    # Disable script execution
    location ~ \.php$ {
        deny all;
    }
    
    # Set safe Content-Type
    default_type application/octet-stream;
    
    # Disable includes
    location ~ \.s?html?$ {
        add_header Content-Type text/plain;
    }
}

# Better: Serve through application controller
location /files {
    internal;  # Only accessible via X-Accel-Redirect
    alias /var/uploads/;
}

from flask import send_file, abort
import os

SAFE_CONTENT_TYPES = {
    'png': 'image/png',
    'jpg': 'image/jpeg',
    'jpeg': 'image/jpeg', 
    'gif': 'image/gif',
    'pdf': 'application/pdf',
}

@app.route('/files/<file_id>')
def serve_file(file_id):
    # Look up file in database
    file_record = db.get_file(file_id)
    if not file_record:
        abort(404)
    
    # Check user has permission
    if not current_user.can_access(file_record):
        abort(403)
    
    # Get file path
    file_path = os.path.join(UPLOAD_DIR, file_record.stored_name)
    if not os.path.exists(file_path):
        abort(404)
    
    # Determine safe content type
    ext = file_record.stored_name.rsplit('.', 1)[-1].lower()
    content_type = SAFE_CONTENT_TYPES.get(ext, 'application/octet-stream')
    
    response = send_file(
        file_path,
        mimetype=content_type,
        as_attachment=True,  # Force download for non-image types
        download_name=file_record.original_name
    )
    
    # Security headers
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['Content-Security-Policy'] = "default-src 'none'"
    response.headers['X-Frame-Options'] = 'DENY'
    
    return response

import clamd
import hashlib

def scan_file_for_malware(file_path: str) -> bool:
    """Scan file with ClamAV"""
    try:
        cd = clamd.ClamdUnixSocket()
        result = cd.scan(file_path)
        
        if result is None:
            return True  # No threats found
        
        # Check scan result
        status = result.get(file_path)
        if status and status[0] == 'FOUND':
            # Log the detection
            logger.warning(f"Malware detected in {file_path}: {status[1]}")
            # Delete the file
            os.remove(file_path)
            return False
        
        return True
    except clamd.ConnectionError:
        logger.error("ClamAV not available - rejecting upload for safety")
        return False

# Alternative: VirusTotal API for cloud scanning
import requests

def scan_with_virustotal(file_path: str, api_key: str) -> bool:
    """Scan file with VirusTotal"""
    # Calculate file hash
    with open(file_path, 'rb') as f:
        file_hash = hashlib.sha256(f.read()).hexdigest()
    
    # Check if already scanned
    response = requests.get(
        f'https://www.virustotal.com/api/v3/files/{file_hash}',
        headers={'x-apikey': api_key}
    )
    
    if response.status_code == 200:
        data = response.json()
        stats = data['data']['attributes']['last_analysis_stats']
        if stats['malicious'] > 0 or stats['suspicious'] > 0:
            return False
    
    # If not found, upload for scanning (implement as needed)
    return True

const multer = require('multer');
const path = require('path');
const crypto = require('crypto');
const fileType = require('file-type');

// Configure multer
const storage = multer.diskStorage({
  destination: '/var/uploads',  // Outside webroot
  filename: (req, file, cb) => {
    // Generate random filename
    const ext = path.extname(file.originalname).toLowerCase();
    const name = crypto.randomBytes(16).toString('hex') + ext;
    cb(null, name);
  }
});

const fileFilter = (req, file, cb) => {
  const allowedMimes = ['image/jpeg', 'image/png', 'image/gif'];
  if (allowedMimes.includes(file.mimetype)) {
    cb(null, true);
  } else {
    cb(new Error('Invalid file type'), false);
  }
};

const upload = multer({
  storage,
  fileFilter,
  limits: {
    fileSize: 5 * 1024 * 1024  // 5MB
  }
});

// Validate magic bytes after upload
const validateMagicBytes = async (filePath) => {
  const type = await fileType.fromFile(filePath);
  const allowedTypes = ['image/jpeg', 'image/png', 'image/gif'];
  return type && allowedTypes.includes(type.mime);
};

app.post('/upload', upload.single('file'), async (req, res) => {
  if (!req.file) {
    return res.status(400).json({ error: 'No file uploaded' });
  }
  
  // Validate magic bytes
  const isValid = await validateMagicBytes(req.file.path);
  if (!isValid) {
    fs.unlinkSync(req.file.path);  // Delete invalid file
    return res.status(400).json({ error: 'Invalid file content' });
  }
  
  res.json({ filename: req.file.filename });
});

public class FileUploadService
{
    private readonly string[] _allowedExtensions = { ".jpg", ".jpeg", ".png", ".gif", ".pdf" };
    private readonly Dictionary<string, byte[]> _fileSignatures = new()
    {
        { ".jpg", new byte[] { 0xFF, 0xD8, 0xFF } },
        { ".jpeg", new byte[] { 0xFF, 0xD8, 0xFF } },
        { ".png", new byte[] { 0x89, 0x50, 0x4E, 0x47 } },
        { ".gif", new byte[] { 0x47, 0x49, 0x46, 0x38 } },
        { ".pdf", new byte[] { 0x25, 0x50, 0x44, 0x46 } }
    };
    
    public async Task<string> UploadFileAsync(IFormFile file)
    {
        // Check extension
        var ext = Path.GetExtension(file.FileName).ToLowerInvariant();
        if (!_allowedExtensions.Contains(ext))
            throw new InvalidOperationException("Invalid file extension");
        
        // Check size
        if (file.Length > 5 * 1024 * 1024)
            throw new InvalidOperationException("File too large");
        
        // Validate magic bytes
        using var reader = new BinaryReader(file.OpenReadStream());
        var headerBytes = reader.ReadBytes(_fileSignatures[ext].Length);
        
        if (!headerBytes.SequenceEqual(_fileSignatures[ext]))
            throw new InvalidOperationException("File signature mismatch");
        
        // Generate safe filename
        var safeFileName = $"{Guid.NewGuid()}{ext}";
        var uploadPath = Path.Combine(_uploadDirectory, safeFileName);
        
        // Save file
        using var stream = new FileStream(uploadPath, FileMode.Create);
        await file.CopyToAsync(stream);
        
        return safeFileName;
    }
}