583 lines
21 KiB
Python
583 lines
21 KiB
Python
|
"""
|
||
|
|
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()
|