""" Path sanitization and security utilities Provides comprehensive path validation and sanitization to prevent security issues """ import logging import os import re from pathlib import Path, PurePath from typing import Optional, List, Tuple, Union try: from .error_handling import ValidationError, SecurityError except ImportError: from error_handling import ValidationError, SecurityError class PathSanitizer: """Comprehensive path sanitization and security validation""" def __init__(self): self.logger = logging.getLogger(__name__) # Dangerous path patterns self.dangerous_patterns = [ r'\.\./', # Directory traversal r'\.\.\.', # Multiple dots r'~/', # Home directory reference r'\$\{.*\}', # Environment variable expansion r'%[A-Za-z0-9_]+%', # Windows environment variables ] # Dangerous characters in file paths self.dangerous_chars = { '<': 'less_than', '>': 'greater_than', ':': 'colon', '"': 'quote', '|': 'pipe', '?': 'question', '*': 'asterisk', '\x00': 'null_byte', '\n': 'newline', '\r': 'carriage_return', '\t': 'tab' } # Reserved names (Windows) self.reserved_names = { 'CON', 'PRN', 'AUX', 'NUL', 'COM1', 'COM2', 'COM3', 'COM4', 'COM5', 'COM6', 'COM7', 'COM8', 'COM9', 'LPT1', 'LPT2', 'LPT3', 'LPT4', 'LPT5', 'LPT6', 'LPT7', 'LPT8', 'LPT9' } # Maximum path lengths self.max_path_length = 4096 # Unix/Linux limit self.max_filename_length = 255 # Most filesystems self.max_path_depth = 32 # Reasonable depth limit def sanitize_vault_relative_path( self, path: Union[str, Path], vault_root: Optional[Union[str, Path]] = None, allow_creation: bool = True ) -> str: """ Sanitize a path that should be relative to an Obsidian vault Args: path: Path to sanitize (should be relative to vault) vault_root: Optional vault root path for additional validation allow_creation: Whether to allow paths that don't exist yet Returns: Sanitized relative path Raises: ValidationError: If path is invalid or unsafe SecurityError: If path poses security risks """ if not path: raise ValidationError( message="Path cannot be empty", field_name="path", validation_rule="non_empty" ) # Convert to string and normalize path_str = str(path).strip() # Basic security checks self._check_dangerous_patterns(path_str) self._check_dangerous_characters(path_str) # Normalize path separators normalized_path = path_str.replace('\\', '/') # Remove leading/trailing slashes for relative paths normalized_path = normalized_path.strip('/') # Check for absolute path attempts if os.path.isabs(path_str) or path_str.startswith('/'): raise SecurityError( message="Absolute paths are not allowed - path must be relative to vault", security_issue="absolute_path_attempt", attempted_path=path_str ) # Check for directory traversal if '..' in normalized_path: raise SecurityError( message="Directory traversal is not allowed", security_issue="directory_traversal", attempted_path=path_str ) # Validate path components path_parts = normalized_path.split('/') self._validate_path_components(path_parts) # Check path length and depth self._validate_path_constraints(normalized_path, path_parts) # Additional validation if vault root is provided if vault_root: self._validate_against_vault_root(normalized_path, vault_root, allow_creation) return normalized_path def sanitize_absolute_path( self, path: Union[str, Path], allowed_roots: Optional[List[Union[str, Path]]] = None, must_exist: bool = True ) -> str: """ Sanitize an absolute path with security checks Args: path: Absolute path to sanitize allowed_roots: List of allowed root directories must_exist: Whether the path must exist Returns: Sanitized absolute path Raises: ValidationError: If path is invalid SecurityError: If path poses security risks """ if not path: raise ValidationError( message="Path cannot be empty", field_name="path", validation_rule="non_empty" ) # Convert to Path object for better handling try: path_obj = Path(path).resolve() except (OSError, ValueError) as e: raise ValidationError( message=f"Invalid path format: {e}", field_name="path", validation_rule="path_format" ) path_str = str(path_obj) # Basic security checks self._check_dangerous_patterns(path_str) self._check_dangerous_characters(path_str) # Check if path exists (if required) if must_exist and not path_obj.exists(): raise ValidationError( message=f"Path does not exist: {path_str}", field_name="path", validation_rule="path_exists" ) # Check against allowed roots if allowed_roots: self._validate_against_allowed_roots(path_obj, allowed_roots) # Check for symbolic link attacks self._check_symbolic_links(path_obj) # Validate path components path_parts = path_obj.parts self._validate_path_components(path_parts) return path_str def sanitize_filename( self, filename: str, allow_extensions: Optional[List[str]] = None, max_length: Optional[int] = None ) -> str: """ Sanitize a filename with security checks Args: filename: Filename to sanitize allow_extensions: List of allowed file extensions max_length: Maximum filename length Returns: Sanitized filename Raises: ValidationError: If filename is invalid SecurityError: If filename poses security risks """ if not filename: raise ValidationError( message="Filename cannot be empty", field_name="filename", validation_rule="non_empty" ) # Remove path separators (filename only) clean_filename = filename.replace('/', '').replace('\\', '') # Basic security checks self._check_dangerous_characters(clean_filename) # Check for reserved names name_without_ext = clean_filename.split('.')[0].upper() if name_without_ext in self.reserved_names: raise SecurityError( message=f"Filename uses reserved name: {name_without_ext}", security_issue="reserved_filename", attempted_path=filename ) # Check filename length max_len = max_length or self.max_filename_length if len(clean_filename) > max_len: raise ValidationError( message=f"Filename too long (max {max_len} characters)", field_name="filename", validation_rule="max_length" ) # Check file extension if restrictions apply if allow_extensions: file_ext = Path(clean_filename).suffix.lower() if file_ext not in allow_extensions: raise ValidationError( message=f"File extension not allowed. Allowed: {', '.join(allow_extensions)}", field_name="filename", validation_rule="extension_not_allowed" ) # Check for hidden files (starting with dot) if clean_filename.startswith('.') and clean_filename != '.obsidian': self.logger.warning(f"Hidden file detected: {clean_filename}") return clean_filename def validate_vault_structure(self, vault_path: Union[str, Path]) -> Tuple[bool, List[str]]: """ Validate that a directory is a proper Obsidian vault Args: vault_path: Path to validate as vault Returns: Tuple of (is_valid, list_of_issues) """ issues = [] vault_path_obj = Path(vault_path) # Check if path exists and is directory if not vault_path_obj.exists(): issues.append(f"Vault path does not exist: {vault_path}") return False, issues if not vault_path_obj.is_dir(): issues.append(f"Vault path is not a directory: {vault_path}") return False, issues # Check for .obsidian directory obsidian_dir = vault_path_obj / '.obsidian' if not obsidian_dir.exists(): issues.append("No .obsidian directory found - this may not be an Obsidian vault") # Check permissions if not os.access(vault_path_obj, os.R_OK): issues.append(f"Vault directory is not readable: {vault_path}") if not os.access(vault_path_obj, os.W_OK): issues.append(f"Vault directory is not writable: {vault_path}") # Check for suspicious files/directories try: for item in vault_path_obj.iterdir(): if item.name.startswith('..'): issues.append(f"Suspicious directory name found: {item.name}") # Check for executable files in vault (potential security risk) if item.is_file() and item.suffix.lower() in ['.exe', '.bat', '.sh', '.cmd']: issues.append(f"Executable file found in vault: {item.name}") except PermissionError: issues.append("Cannot read vault directory contents - permission denied") return len(issues) == 0, issues def create_safe_path( self, base_path: Union[str, Path], relative_path: str, create_dirs: bool = False ) -> Path: """ Safely create a path by joining base and relative paths Args: base_path: Base directory path relative_path: Relative path to join create_dirs: Whether to create intermediate directories Returns: Safe combined path Raises: SecurityError: If the resulting path would be unsafe """ # Sanitize the relative path first safe_relative = self.sanitize_vault_relative_path(relative_path, allow_creation=True) # Create the combined path base_path_obj = Path(base_path).resolve() combined_path = base_path_obj / safe_relative # Ensure the result is still within the base path try: combined_path.resolve().relative_to(base_path_obj.resolve()) except ValueError: raise SecurityError( message="Resulting path would be outside base directory", security_issue="path_escape", attempted_path=str(combined_path) ) # Create directories if requested if create_dirs and not combined_path.parent.exists(): try: combined_path.parent.mkdir(parents=True, exist_ok=True) self.logger.info(f"Created directory: {combined_path.parent}") except (OSError, PermissionError) as e: raise ValidationError( message=f"Cannot create directory: {e}", field_name="path", validation_rule="directory_creation" ) return combined_path def _check_dangerous_patterns(self, path: str) -> None: """Check for dangerous patterns in path""" for pattern in self.dangerous_patterns: if re.search(pattern, path, re.IGNORECASE): raise SecurityError( message=f"Dangerous pattern detected in path: {pattern}", security_issue="dangerous_pattern", attempted_path=path ) def _check_dangerous_characters(self, path: str) -> None: """Check for dangerous characters in path""" for char, char_name in self.dangerous_chars.items(): if char in path: raise SecurityError( message=f"Dangerous character '{char}' ({char_name}) found in path", security_issue="dangerous_character", attempted_path=path ) def _validate_path_components(self, path_parts: Union[List[str], Tuple[str, ...]]) -> None: """Validate individual path components""" for part in path_parts: if not part: # Empty component continue # Check for reserved names if part.upper() in self.reserved_names: raise SecurityError( message=f"Path component uses reserved name: {part}", security_issue="reserved_name", attempted_path=str(path_parts) ) # Check component length if len(part) > self.max_filename_length: raise ValidationError( message=f"Path component too long: {part} (max {self.max_filename_length})", field_name="path_component", validation_rule="max_length" ) # Check for control characters if any(ord(c) < 32 for c in part): raise SecurityError( message=f"Path component contains control characters: {part}", security_issue="control_characters", attempted_path=part ) def _validate_path_constraints(self, path: str, path_parts: List[str]) -> None: """Validate path length and depth constraints""" # Check total path length if len(path) > self.max_path_length: raise ValidationError( message=f"Path too long (max {self.max_path_length} characters)", field_name="path", validation_rule="max_path_length" ) # Check path depth if len(path_parts) > self.max_path_depth: raise ValidationError( message=f"Path too deep (max {self.max_path_depth} levels)", field_name="path", validation_rule="max_path_depth" ) def _validate_against_vault_root( self, relative_path: str, vault_root: Union[str, Path], allow_creation: bool ) -> None: """Validate path against vault root""" vault_root_obj = Path(vault_root) full_path = vault_root_obj / relative_path # Ensure path stays within vault try: full_path.resolve().relative_to(vault_root_obj.resolve()) except ValueError: raise SecurityError( message="Path would escape vault directory", security_issue="vault_escape", attempted_path=relative_path ) # Check if parent directory exists (for file creation) if not allow_creation and not full_path.parent.exists(): raise ValidationError( message=f"Parent directory does not exist: {full_path.parent}", field_name="path", validation_rule="parent_directory_exists" ) def _validate_against_allowed_roots( self, path_obj: Path, allowed_roots: List[Union[str, Path]] ) -> None: """Validate path is within allowed root directories""" path_resolved = path_obj.resolve() for allowed_root in allowed_roots: try: allowed_root_obj = Path(allowed_root).resolve() path_resolved.relative_to(allowed_root_obj) return # Path is within this allowed root except ValueError: continue # Try next allowed root # Path is not within any allowed root raise SecurityError( message=f"Path is not within any allowed root directory", security_issue="unauthorized_path", attempted_path=str(path_obj) ) def _check_symbolic_links(self, path_obj: Path) -> None: """Check for symbolic link attacks""" # Check if any part of the path is a symbolic link current_path = path_obj while current_path != current_path.parent: if current_path.is_symlink(): self.logger.warning(f"Symbolic link detected in path: {current_path}") # Don't reject, but log for security monitoring break current_path = current_path.parent class PathSecurityManager: """High-level path security management""" def __init__(self, vault_root: Optional[Union[str, Path]] = None): self.vault_root = Path(vault_root) if vault_root else None self.sanitizer = PathSanitizer() self.logger = logging.getLogger(__name__) def set_vault_root(self, vault_root: Union[str, Path]) -> None: """Set the vault root directory""" self.vault_root = Path(vault_root) # Validate vault structure is_valid, issues = self.sanitizer.validate_vault_structure(self.vault_root) if not is_valid: self.logger.warning(f"Vault validation issues: {'; '.join(issues)}") def get_safe_vault_path(self, relative_path: str, create_dirs: bool = False) -> Path: """ Get a safe path within the vault Args: relative_path: Relative path within vault create_dirs: Whether to create intermediate directories Returns: Safe absolute path within vault Raises: ValidationError: If vault root not set or path invalid """ if not self.vault_root: raise ValidationError( message="Vault root not configured", field_name="vault_root", validation_rule="not_configured" ) return self.sanitizer.create_safe_path( self.vault_root, relative_path, create_dirs=create_dirs ) def validate_file_operation( self, file_path: str, operation: str = "read", content_length: Optional[int] = None ) -> Tuple[bool, Optional[str]]: """ Validate a file operation for security Args: file_path: Path to file operation: Type of operation (read, write, append, delete) content_length: Length of content for write operations Returns: Tuple of (is_allowed, reason_if_not_allowed) """ try: # Sanitize the path safe_path = self.sanitizer.sanitize_vault_relative_path( file_path, self.vault_root, allow_creation=(operation in ['write', 'append']) ) # Additional checks based on operation if operation == 'write' and content_length: # Check for reasonable content size limits max_content_size = 10 * 1024 * 1024 # 10MB if content_length > max_content_size: return False, f"Content too large ({content_length} bytes, max {max_content_size})" # Check file extension for security file_ext = Path(safe_path).suffix.lower() dangerous_extensions = ['.exe', '.bat', '.sh', '.cmd', '.scr', '.vbs', '.js'] if file_ext in dangerous_extensions: return False, f"Dangerous file extension: {file_ext}" return True, None except (ValidationError, SecurityError) as e: return False, str(e) # Global instances path_sanitizer = PathSanitizer() path_security_manager = PathSecurityManager()