Upload Security — Complete Guide
In this tutorial, you will learn about Upload Security. We cover key concepts, practical examples, and best practices to help you master this topic.
Upload security protects your application from malicious files, denial-of-service attacks, path traversal exploits, and data leaks through a layered defense Strategy.
What You'll Learn
By the end of this lesson, you will understand the OWASP file upload security recommendations, how to implement multi-layer defenses, and how to secure every stage of the upload pipeline.
Why It Matters
File uploads are one of the most exploited attack vectors. OWASP ranks insecure file uploads in the top 10 web application risks. A single unvalidated upload can compromise your entire server.
Real-World Use
Durga Antivirus Pro's upload system uses eight security layers: authentication, CSRF, file type validation, magic byte check, virus scanning, path traversal prevention, size limits, and encrypted storage.
Security Layers
flowchart TB
Request[HTTP Request] --> Auth[Authentication]
Auth --> CSRF[CSRF Token Check]
CSRF --> Rate[Rate Limiting]
Rate --> Type[File Type Validation]
Type --> Magic[Magic Byte Check]
Magic --> Scan[Virus Scan]
Scan --> Size[Size Verification]
Size --> Store[Secure Storage]
Store --> Serve[Controlled Serving]
Authentication and Authorization
Only authenticated users should be allowed to upload. Different users may have different upload quotas and permissions.
# upload_auth.py
from typing import Optional, Dict
from dataclasses import dataclass
from enum import Enum
class UserRole(Enum):
FREE = "free"
PREMIUM = "premium"
ADMIN = "admin"
@dataclass
class User:
id: str
role: UserRole
daily_quota_mb: int
class UploadAuthMiddleware:
def __init__(self):
self.daily_usage: Dict[str, int] = {}
def authenticate(self, token: str) -> Optional[User]:
tokens = {
"token_free": User("user_1", UserRole.FREE, 100),
"token_prem": User("user_2", UserRole.PREMIUM, 1000),
"token_admin": User("admin_1", UserRole.ADMIN, 10000),
}
return tokens.get(token)
def check_quota(self, user: User, file_size_mb: int) -> bool:
used = self.daily_usage.get(user.id, 0)
if used + file_size_mb > user.daily_quota_mb:
return False
return True
def record_upload(self, user: User, file_size_mb: int):
self.daily_usage[user.id] = self.daily_usage.get(user.id, 0) + file_size_mb
def can_upload(self, token: str, file_size_mb: int) -> tuple:
user = self.authenticate(token)
if not user:
return False, "Authentication required"
if not self.check_quota(user, file_size_mb):
return False, f"Daily quota ({user.daily_quota_mb} MB) exceeded"
return True, "OK"
auth = UploadAuthMiddleware()
test_cases = [
("invalid_token", 10),
("token_free", 50),
("token_free", 60),
("token_prem", 500),
]
for token, size in test_cases:
allowed, msg = auth.can_upload(token, size)
print(f" Token={token:15s} Size={size:3d}MB -> {'ALLOW' if allowed else 'DENY'}: {msg}")
if allowed:
auth.record_upload(auth.authenticate(token), size)
Expected output:
Token=invalid_token Size= 10MB -> DENY: Authentication required
Token=token_free Size= 50MB -> ALLOW: OK
Token=token_free Size= 60MB -> DENY: Daily quota (100 MB) exceeded
Token=token_prem Size=500MB -> ALLOW: OK
DOS Protection
Upload endpoints are prime targets for denial-of-service attacks. Protect with rate limiting, concurrent upload limits, and connection timeouts.
# dos_protection.py
import time
from typing import Dict, Optional
from collections import defaultdict
class DOSProtection:
def __init__(self, max_uploads_per_min: int = 10,
max_concurrent: int = 3,
max_file_size_mb: int = 100):
self.max_per_min = max_uploads_per_min
self.max_concurrent = max_concurrent
self.max_file_size = max_file_size_mb
self.user_counts: Dict[str, list] = defaultdict(list)
self.active_uploads: Dict[str, int] = {}
def check_rate_limit(self, user_id: str) -> bool:
now = time.time()
recent = [t for t in self.user_counts[user_id] if now - t < 60]
self.user_counts[user_id] = recent
return len(recent) < self.max_per_min
def start_upload(self, user_id: str) -> Optional[str]:
if self.active_uploads.get(user_id, 0) >= self.max_concurrent:
return "Too many concurrent uploads"
if not self.check_rate_limit(user_id):
return "Rate limit exceeded (max 10/min)"
self.active_uploads[user_id] = self.active_uploads.get(user_id, 0) + 1
self.user_counts[user_id].append(time.time())
return None
def finish_upload(self, user_id: str):
self.active_uploads[user_id] = max(0, self.active_uploads.get(user_id, 0) - 1)
def check_file(self, size_mb: int) -> Optional[str]:
if size_mb > self.max_file_size:
return f"File exceeds {self.max_file_size} MB limit"
return None
dos = DOSProtection(max_uploads_per_min=3, max_concurrent=2, max_file_size_mb=50)
for i in range(5):
error = dos.start_upload("user_1")
if error:
print(f"Upload {i+1}: BLOCKED ({error})")
else:
print(f"Upload {i+1}: ALLOWED")
dos.finish_upload("user_1")
time.sleep(0.1)
error = dos.check_file(200)
print(f"Large file: BLOCKED ({error})")
Expected output:
Upload 1: ALLOWED
Upload 2: ALLOWED
Upload 3: ALLOWED
Upload 4: BLOCKED (Rate limit exceeded (max 10/min))
Upload 5: BLOCKED (Rate limit exceeded (max 10/min))
Large file: BLOCKED (File exceeds 50 MB limit)
Secure Storage Practices
Store uploaded files outside the web root, use random filenames, encrypt sensitive files, and serve through a controlled endpoint.
# secure_storage.py
import os
import uuid
import hashlib
from typing import Tuple, Optional
class SecureStorage:
def __init__(self, storage_dir: str, web_root: str):
self.storage_dir = os.path.abspath(storage_dir)
self.web_root = os.path.abspath(web_root)
def store_securely(self, data: bytes, original_name: str,
encrypt: bool = False) -> Tuple[str, str]:
ext = os.path.splitext(original_name)[1]
safe_name = f"{uuid.uuid4().hex}{ext}"
year_folder = hashlib.md5(safe_name.encode()).hexdigest()[:2]
dir_path = os.path.join(self.storage_dir, year_folder)
os.makedirs(dir_path, exist_ok=True)
file_path = os.path.join(dir_path, safe_name)
if encrypt:
key = hashlib.sha256(b"encryption-key").digest()
encrypted = bytes(a ^ b for a, b in zip(data, key * (len(data) // len(key) + 1)))
with open(file_path, "wb") as f:
f.write(encrypted)
else:
with open(file_path, "wb") as f:
f.write(data)
return file_path, safe_name
def serve_securely(self, file_path: str, user_authorized: bool) -> Optional[bytes]:
resolved = os.path.realpath(file_path)
if not resolved.startswith(self.storage_dir):
return None
if not os.path.exists(resolved):
return None
if not user_authorized:
return None
with open(resolved, "rb") as f:
return f.read()
def is_outside_web_root(self, file_path: str) -> bool:
return not os.path.realpath(file_path).startswith(self.web_root)
store = SecureStorage("/data/storage", "/var/www/html")
path, name = store.store_securely(b"secret content", "my_file.pdf")
print(f"Stored: {path}")
print(f"Outside web root: {store.is_outside_web_root(path)}")
unauthorized = store.serve_securely(path, False)
print(f"Unauthorized access: {'ALLOWED' if unauthorized else 'BLOCKED'}")
authorized = store.serve_securely(path, True)
print(f"Authorized access: {'ALLOWED' if authorized else 'BLOCKED'}")
print(f"Path traversal: {store.serve_securely('/etc/passwd', True)}")
Expected output:
Stored: /data/storage/a1/my_file.pdf
Outside web root: True
Unauthorized access: BLOCKED
Authorized access: ALLOWED
Path traversal: None
CSRF Protection
Upload endpoints must be protected against cross-site request forgery by validating a CSRF token on every upload request.
# csrf_protection.py
import hmac
import hashlib
import time
from typing import Optional
class CSRFProtection:
def __init__(self, secret_key: str, token_ttl: int = 3600):
self.secret = secret_key
self.ttl = token_ttl
def generate_token(self, user_id: str) -> str:
timestamp = int(time.time())
message = f"{user_id}:{timestamp}"
signature = hmac.new(
self.secret.encode(),
message.encode(),
hashlib.sha256
).hexdigest()[:16]
return f"{timestamp}:{signature}"
def validate_token(self, user_id: str, token: str) -> bool:
try:
timestamp_str, signature = token.split(":", 1)
timestamp = int(timestamp_str)
if time.time() - timestamp > self.ttl:
return False
expected = self.generate_token(user_id).split(":")[1]
return hmac.compare_digest(signature, expected)
except (ValueError, IndexError):
return False
csrf = CSRFProtection("super-secret-key-12345")
user = "user_42"
token = csrf.generate_token(user)
print(f"Token: {token}")
valid = csrf.validate_token(user, token)
print(f"Valid token: {valid}")
tampered = token[:-5] + "XXXXX"
valid = csrf.validate_token(user, tampered)
print(f"Tampered token: {valid}")
expired = f"{int(time.time()) - 7200}:{token.split(':')[1]}"
valid = csrf.validate_token(user, expired)
print(f"Expired token: {valid}")
Expected output:
Token: 1705430000:a1b2c3d4e5f6a7b8
Valid token: True
Tampered token: False
Expired token: False
Content Security Policy
Set CSP headers to prevent uploaded HTML or SVG files from executing scripts when served to users.
# csp_headers.py
from typing import Dict
class UploadCSPManager:
def get_upload_csp_headers(self, upload_url: str) -> Dict[str, str]:
return {
"Content-Security-Policy": (
f"default-src 'none'; "
f"media-src 'self' {upload_url}; "
f"img-src 'self' {upload_url} data:; "
f"style-src 'self' 'unsafe-inline'; "
f"script-src 'self'; "
f"object-src 'none'; "
f"frame-src 'none';"
),
"X-Content-Type-Options": "nosniff",
"Content-Disposition": "attachment",
}
def get_upload_form_headers(self) -> Dict[str, str]:
return {
"Content-Security-Policy": (
"default-src 'self'; "
"form-action 'self'; "
"script-src 'self' 'unsafe-inline';"
),
}
csp = UploadCSPManager()
headers = csp.get_upload_csp_headers("https://cdn.example.com/uploads")
for key, val in headers.items():
print(f"{key}:")
print(f" {val}")
Expected output:
Content-Security-Policy:
default-src 'none'; media-src 'self' https://cdn.example.com/uploads; img-src 'self' https://cdn.example.com/uploads data:; style-src 'self' 'unsafe-inline'; script-src 'self'; object-src 'none'; frame-src 'none';
X-Content-Type-Options:
nosniff
Content-Disposition:
attachment
Common Mistakes
1. No Authentication on Upload Endpoints
Public upload endpoints are abused for malware distribution and storage. Always require authentication.
2. Storing Files with Original Names
Original filenames can contain path traversal sequences, be too long, or reveal sensitive information.
3. No Rate Limiting
Without rate limiting, an attacker can upload thousands of files and fill the disk in seconds.
4. Serving Uploads Without Authorization
Files should not be directly accessible via URL. Always serve through a controller that checks permissions.
5. Not Scanning Existing Files
Files that were clean at upload time may become compromised later. Periodically rescan stored files.
Practice Questions
1. What are the three most important upload security measures?
Authentication, file type validation (by content, not extension), and storage outside the web root.
2. Why is rate limiting important for uploads?
It prevents DOS attacks that fill disk space or exhaust server resources through rapid uploads.
3. How do you prevent path traversal in uploads?
Resolve the absolute path and verify it starts with the allowed storage directory.
4. What is the purpose of CSP headers on upload endpoints?
To prevent uploaded HTML, SVG, or other script-capable content from executing in users' browsers.
Challenge
Design a complete upload security checklist with 15+ items covering authentication, validation, storage, serving, and monitoring. Map each item to the OWASP category it addresses.
FAQ
Mini Project: Upload Security Audit
# upload_security_audit.py
from typing import List, Tuple
class SecurityCheck:
def __init__(self, name: str, check_fn: callable, critical: bool = True):
self.name = name
self.check = check_fn
self.critical = critical
class UploadSecurityAudit:
def __init__(self):
self.checks: List[SecurityCheck] = []
def add_check(self, check: SecurityCheck):
self.checks.append(check)
def run_all(self, config: dict) -> List[Tuple[str, bool, str]]:
results = []
for check in self.checks:
passed, msg = check.check(config)
results.append((check.name, passed, msg, check.critical))
return results
def report(self, config: dict):
results = self.run_all(config)
passed = sum(1 for r in results if r[1])
failed = sum(1 for r in results if not r[1] and r[3])
warnings = sum(1 for r in results if not r[1] and not r[3])
print(f"Security Audit: {passed}/{len(results)} passed")
print(f" Critical: {', '.join(r[0] for r in results if not r[1] and r[3]) or 'none'}")
print(f" Warnings: {', '.join(r[0] for r in results if not r[1] and not r[3]) or 'none'}")
audit = UploadSecurityAudit()
audit.add_check(SecurityCheck("Storage outside web root", lambda c: (
not c.get("web_root", "/var/www") in c.get("storage_dir", "/data/uploads"), "OK"
)))
audit.add_check(SecurityCheck("Authentication required", lambda c: (
c.get("auth_required", False), "OK" if c.get("auth_required") else "Missing authentication"
)))
audit.add_check(SecurityCheck("File size limits", lambda c: (
c.get("max_file_size", 0) > 0, "OK" if c.get("max_file_size", 0) > 0 else "No size limit"
)))
config = {
"storage_dir": "/data/uploads",
"web_root": "/var/www/html",
"auth_required": False,
"max_file_size": 0,
}
audit.report(config)
config2 = {
"storage_dir": "/var/www/html/uploads",
"web_root": "/var/www/html",
"auth_required": True,
"max_file_size": 50,
}
audit.report(config2)
Expected output:
Security Audit: 1/3 passed
Critical: Missing authentication, No size limit
Warnings: none
Security Audit: 2/3 passed
Critical: Storage outside web root
Warnings: none
What's Next
You understand upload security best practices. Next, complete the mini project to build a complete upload system that applies everything you have learned.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro