Files
philandClaude 25086fc01b Add comprehensive RSS scraper implementation with security and testing
- Modular architecture with separate modules for scraping, parsing, security, validation, and caching
- Comprehensive security measures including HTML sanitization, rate limiting, and input validation
- Robust error handling with custom exceptions and retry logic
- HTTP caching with ETags and Last-Modified headers for efficiency
- Pre-compiled regex patterns for improved performance
- Comprehensive test suite with 66 tests covering all major functionality
- Docker support for containerized deployment
- Configuration management with environment variable support
- Working parser that successfully extracts 32 articles from Warhammer Community

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-06 09:15:06 -06:00

113 lines
4.0 KiB
Python

"""URL and path validation utilities."""
import os
import urllib.parse
import logging
from typing import Optional
from .config import Config
from .exceptions import ValidationError, FileOperationError
from .security import validate_url_security, sanitize_filename
logger = logging.getLogger(__name__)
def validate_url(url: str) -> bool:
"""Validate URL against whitelist of allowed domains"""
try:
logger.debug(f"Validating URL: {url}")
# Enhanced security validation first
if not validate_url_security(url):
raise ValidationError(f"URL failed security validation: {url}")
parsed = urllib.parse.urlparse(url)
if not parsed.scheme or not parsed.netloc:
raise ValidationError("Invalid URL format")
# Check if domain is in allowed list
domain = parsed.netloc.lower()
allowed_domains = Config.get_allowed_domains()
if domain not in allowed_domains:
raise ValidationError(f"Domain {domain} not in allowed list: {allowed_domains}")
logger.debug(f"URL validation successful for domain: {domain}")
return True
except ValidationError:
raise
except Exception as e:
logger.error(f"URL validation failed for {url}: {e}")
raise ValidationError(f"URL validation failed: {e}")
def validate_output_path(path: str, base_dir: str) -> str:
"""Validate and sanitize output file path"""
logger.debug(f"Validating output path: {path} in base directory: {base_dir}")
try:
# Sanitize the filename component
dir_part, filename = os.path.split(path)
if filename:
sanitized_filename = sanitize_filename(filename)
path = os.path.join(dir_part, sanitized_filename)
logger.debug(f"Sanitized filename: {filename} -> {sanitized_filename}")
# Resolve to absolute path and check if it's safe
abs_path = os.path.abspath(path)
abs_base = os.path.abspath(base_dir)
# Ensure path is within allowed directory
if not abs_path.startswith(abs_base):
error_msg = f"Output path {abs_path} is outside allowed directory {abs_base}"
logger.error(error_msg)
raise ValidationError(error_msg)
# Additional security check for suspicious patterns - only check for directory traversal
# Note: We allow absolute paths since they're resolved safely above
if '..' in path:
error_msg = f"Directory traversal detected in path: {path}"
logger.error(error_msg)
raise ValidationError(error_msg)
# Ensure output directory exists
os.makedirs(abs_base, exist_ok=True)
logger.debug(f"Output path validated: {abs_path}")
return abs_path
except OSError as e:
raise FileOperationError(f"Failed to create or access directory {base_dir}: {e}")
except ValidationError:
raise
except Exception as e:
raise FileOperationError(f"Unexpected error validating path: {e}")
def validate_link(link: Optional[str], base_url: str) -> Optional[str]:
"""Validate and sanitize article links"""
if not link:
return None
try:
# Handle relative URLs
if link.startswith('/'):
parsed_base = urllib.parse.urlparse(base_url)
link = f"{parsed_base.scheme}://{parsed_base.netloc}{link}"
# Enhanced security validation
if not validate_url_security(link):
logger.warning(f"Link failed security validation: {link}")
return None
# Validate the resulting URL
parsed = urllib.parse.urlparse(link)
if not parsed.scheme or not parsed.netloc:
return None
# Ensure it's from allowed domain
domain = parsed.netloc.lower()
if domain not in Config.get_allowed_domains():
return None
return link
except Exception:
return None