gitea-admin/gh-christianlouis-dmarq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Address code review feedback: improve XSS prevention, structured errors, API key logging, CSP warnings

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-09 11:51:58 +00:00
parent 2e22941927
commit 8550799bf8
6 changed files with 77 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/api_v1/endpoints/reports.py
+3 -3
View File
@@ -5,7 +5,7 @@ import logging
from app.services.dmarc_parser import DMARCParser
from app.services.report_store import ReportStore
from app.utils.domain_validator import validate_domain
from app.utils.domain_validator import validate_domain, DomainValidationError
logger = logging.getLogger(__name__)
@@ -133,8 +133,8 @@ async def upload_report(file: UploadFile = File(...)):
)
# Validate domain format (not DNS resolution to avoid external calls)
is_valid, error_msg = validate_domain(domain)
if not is_valid and "could not be resolved" not in error_msg.lower():
is_valid, error_msg, error_code = validate_domain(domain, check_dns=False)
if not is_valid and error_code != DomainValidationError.DNS_RESOLUTION_FAILED:
# Allow domains that fail DNS resolution but have valid format
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
backend/app/core/security.py
+22 -1
View File
@@ -19,9 +19,30 @@ pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
security_bearer = HTTPBearer(auto_error=False)
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
# In-memory API keys storage (for MVP - should be moved to database in production)
# In-memory API keys storage
# ⚠️ WARNING: This is a simple in-memory implementation suitable for:
# - Development and testing environments
# - Single-instance deployments
# - MVP/prototype applications
#
# ⚠️ NOT SUITABLE FOR PRODUCTION when:
# - Running multiple application instances (keys not shared)
# - Requiring key persistence across restarts
# - Needing key rotation and management
#
# For production, implement:
# - Database-backed key storage (with encryption at rest)
# - Redis or similar distributed cache for shared key storage
# - Integration with external secret management (AWS Secrets Manager, HashiCorp Vault, etc.)
# - Proper key rotation policies
_api_keys = set()
logger.warning(
"Using in-memory API key storage. "
"Keys will be lost on restart. "
"Not suitable for production multi-instance deployments."
)
def generate_api_key() -> str:
"""
backend/app/main.py
+11 -3
View File
@@ -112,18 +112,26 @@ def create_app() -> FastAPI:
"""Initialize background tasks and security on application startup"""
global background_task
# Generate and log initial API key for admin access
# Generate and provide admin API key
api_key = generate_api_key()
add_api_key(api_key)
# Security: Log only last 8 characters for reference
logger.warning(
"=" * 80 + "\n"
"IMPORTANT: Admin API Key Generated\n"
f"API Key: {api_key}\n"
"Save this key securely - it will not be shown again.\n"
f"API Key (last 8 chars): ...{api_key[-8:]}\n"
"Full key stored securely in memory.\n"
"For production, retrieve the key through secure configuration management.\n"
"Use this key in the X-API-Key header for admin endpoints.\n"
"=" * 80
)
# In development, also log the full key for convenience
# This should be removed in production or controlled by environment variable
if os.getenv("ENVIRONMENT", "development") == "development":
logger.info(f"Development Mode - Full API Key: {api_key}")
# Check if IMAP credentials are configured
if all([settings.IMAP_SERVER, settings.IMAP_USERNAME, settings.IMAP_PASSWORD]):
logger.info("Starting IMAP polling background task")
backend/app/middleware/security.py
+6 -50
View File
@@ -51,10 +51,14 @@ class SecurityHeadersMiddleware(BaseHTTPMiddleware):
# Content Security Policy (CSP)
# Restricts sources of content that can be loaded
# TODO: Remove 'unsafe-inline' and 'unsafe-eval' and use nonces/hashes instead
csp_directives = [
"default-src 'self'",
"script-src 'self' 'unsafe-inline' 'unsafe-eval'", # Allow inline scripts for now
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com",
# Note: 'unsafe-inline' and 'unsafe-eval' weaken XSS protection
# These should be removed and replaced with nonces or CSP hashes
# See: https://developer.mozilla.org/en-US/docs/Web/HTTP/CSP
"script-src 'self' 'unsafe-inline' 'unsafe-eval'", # TODO: Use nonces
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com", # TODO: Use nonces
"font-src 'self' https://fonts.gstatic.com",
"img-src 'self' data: https:",
"connect-src 'self'",
@@ -112,51 +116,3 @@ class SecurityHeadersMiddleware(BaseHTTPMiddleware):
response.headers["Expires"] = "0"
return response
class RateLimitMiddleware(BaseHTTPMiddleware):
"""
Simple rate limiting middleware to prevent abuse.
For production, consider using a more robust solution like slowapi or Redis-based rate limiting.
"""
def __init__(self, app, requests_per_minute: int = 60):
"""
Initialize rate limiting middleware.
Args:
app: FastAPI application instance
requests_per_minute: Maximum requests allowed per minute per IP
"""
super().__init__(app)
self.requests_per_minute = requests_per_minute
self.request_counts = {} # Simple in-memory store (not production-ready)
logger.warning(
"Using in-memory rate limiting. "
"For production, use Redis or similar distributed storage."
)
async def dispatch(self, request: Request, call_next: Callable) -> Response:
"""
Check rate limit and process request.
Args:
request: Incoming HTTP request
call_next: Next middleware/handler in the chain
Returns:
HTTP response or 429 Too Many Requests if rate limit exceeded
"""
# Get client IP
client_ip = request.client.host if request.client else "unknown"
# For now, just log and pass through
# TODO: Implement actual rate limiting logic with time windows
# This is a placeholder for the actual implementation
response = await call_next(request)
# Add rate limit headers for transparency
response.headers["X-RateLimit-Limit"] = str(self.requests_per_minute)
return response
backend/app/tests/test_security.py
+5 -5
View File
@@ -70,7 +70,7 @@ class TestDomainValidation:
]
for domain in valid_domains:
is_valid, error = validate_domain(domain, check_dns=False)
is_valid, error, error_code = validate_domain(domain, check_dns=False)
assert is_valid, f"Domain {domain} should be valid: {error}"
def test_invalid_domain_format(self):
@@ -89,7 +89,7 @@ class TestDomainValidation:
]
for domain in invalid_domains:
is_valid, error = validate_domain(domain, check_dns=False)
is_valid, error, error_code = validate_domain(domain, check_dns=False)
assert not is_valid, f"Domain '{domain}' should be invalid"
assert error is not None
@@ -107,21 +107,21 @@ class TestDomainValidation:
]
for domain in malicious_domains:
is_valid, error = validate_domain(domain, check_dns=False)
is_valid, error, error_code = validate_domain(domain, check_dns=False)
assert not is_valid, f"Malicious domain '{domain}' should be rejected"
def test_domain_length_limits(self):
"""Test domain length validation."""
# Max label is 63 characters - this should be caught by label length check
long_label = "a" * 64 + ".example.com"
is_valid, error = validate_domain(long_label, check_dns=False)
is_valid, error, error_code = validate_domain(long_label, check_dns=False)
assert not is_valid
# Could be caught by format check or label length check
assert error is not None
# Max domain is 253 characters
long_domain = "a" * 254 # 254 chars, no dot
is_valid, error = validate_domain(long_domain, check_dns=False)
is_valid, error, error_code = validate_domain(long_domain, check_dns=False)
assert not is_valid
assert "too long" in error.lower() or "invalid" in error.lower()
backend/app/utils/domain_validator.py
+30 -16
View File
@@ -2,8 +2,19 @@ import re
import socket
from typing import Dict, Tuple, Union, Optional
# Error codes for structured error handling
class DomainValidationError:
"""Domain validation error codes"""
EMPTY = "empty"
TOO_LONG = "too_long"
INVALID_FORMAT = "invalid_format"
INVALID_CHARACTERS = "invalid_characters"
LABEL_TOO_LONG = "label_too_long"
INVALID_LABEL = "invalid_label"
DNS_RESOLUTION_FAILED = "dns_resolution_failed"
def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Optional[str]]:
def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Optional[str], Optional[str]]:
"""
Validates a domain name for format and optionally resolvability.
@@ -12,25 +23,26 @@ def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Opt
check_dns: Whether to perform DNS resolution check (default: True)
Returns:
Tuple containing (is_valid, error_message)
Tuple containing (is_valid, error_message, error_code)
- is_valid: Boolean indicating if domain is valid
- error_message: String with error message if not valid, None if valid
- error_code: Error code constant for programmatic handling, None if valid
"""
# Security: Check for empty or None domain
if not domain_name:
return False, "Domain name cannot be empty"
return False, "Domain name cannot be empty", DomainValidationError.EMPTY
# Security: Check maximum length (DNS standard is 253 characters)
if len(domain_name) > 253:
return False, "Domain name too long (max 253 characters)"
return False, "Domain name too long (max 253 characters)", DomainValidationError.TOO_LONG
# Security: Check for whitespace
if ' ' in domain_name or '\t' in domain_name or '\n' in domain_name:
return False, "Domain name cannot contain whitespace"
return False, "Domain name cannot contain whitespace", DomainValidationError.INVALID_CHARACTERS
# Security: Check for suspicious characters
if any(char in domain_name for char in ['<', '>', '"', "'", '\\', '|', ';', '&', '$', '`']):
return False, "Domain name contains invalid characters"
return False, "Domain name contains invalid characters", DomainValidationError.INVALID_CHARACTERS
# Check domain format with regex
# This regex allows domain names with alphanumeric characters, hyphens,
@@ -38,27 +50,27 @@ def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Opt
# Updated to be more strict and prevent potential attacks
domain_pattern = r'^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9]$'
if not re.match(domain_pattern, domain_name.lower()):
return False, "Invalid domain format"
return False, "Invalid domain format", DomainValidationError.INVALID_FORMAT
# Security: Check each label length (max 63 characters per label)
labels = domain_name.split('.')
for label in labels:
if len(label) > 63:
return False, f"Domain label too long: '{label}' (max 63 characters per label)"
return False, f"Domain label too long: '{label}' (max 63 characters per label)", DomainValidationError.LABEL_TOO_LONG
if label.startswith('-') or label.endswith('-'):
return False, f"Domain label cannot start or end with hyphen: '{label}'"
return False, f"Domain label cannot start or end with hyphen: '{label}'", DomainValidationError.INVALID_LABEL
# Check if domain exists by attempting to resolve DNS (optional)
if check_dns:
try:
socket.gethostbyname(domain_name)
return True, None
return True, None, None
except socket.gaierror:
# We could consider this valid if we don't require DNS resolution,
# but since DMARC requires valid DNS, we'll mark it as warning
return False, "Domain could not be resolved (DNS lookup failed)"
return False, "Domain could not be resolved (DNS lookup failed)", DomainValidationError.DNS_RESOLUTION_FAILED
return True, None
return True, None, None
def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]:
@@ -78,7 +90,7 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]:
# Validate domain name
if "name" in domain_data:
# Don't check DNS for domain config validation
is_valid, error_msg = validate_domain(domain_data["name"], check_dns=False)
is_valid, error_msg, error_code = validate_domain(domain_data["name"], check_dns=False)
if not is_valid:
errors["name"] = error_msg
else:
@@ -88,9 +100,11 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]:
if "description" in domain_data and domain_data["description"]:
if len(domain_data["description"]) > 500:
errors["description"] = "Description is too long (max 500 characters)"
# Security: Check for suspicious content in description
if any(char in domain_data["description"] for char in ['<script', '<iframe', 'javascript:']):
errors["description"] = "Description contains potentially unsafe content"
# Security: Use html.escape to prevent XSS
import html
escaped = html.escape(domain_data["description"])
if escaped != domain_data["description"]:
errors["description"] = "Description contains potentially unsafe HTML content"
# Return validation results
return {