""" Asset Validation and Integrity Checking for Trixy Application This module provides comprehensive validation and integrity checking for all types of assets in the Trixy asset management system. It ensures asset quality, format compliance, and data integrity. Features: - Format-specific validation for all supported asset types - File integrity checking using checksums and signatures - Content validation (e.g., JSON syntax, image corruption) - Size and constraint validation - Security scanning for malicious content - Custom validation rule support - Batch validation for multiple assets - Validation result reporting and logging Asset Validation Types: - Audio files: Format validity, sample rate, channel count, duration - Image files: Format validity, dimensions, color depth, corruption detection - Text files: Encoding validation, syntax checking, content analysis - Config files: Schema validation, format compliance, required fields - Binary files: Signature verification, integrity checks - Model files: Format validation, metadata verification Usage: from trixy_core.assets.asset_validator import AssetValidator validator = AssetValidator() result = validator.validate_asset("/path/to/asset.wav", AssetType.AUDIO_WAV) if result.is_valid: print("Asset is valid") else: print(f"Validation errors: {result.errors}") """ import os import hashlib import json import struct import mimetypes from pathlib import Path from typing import Dict, Any, Optional, List, Set, Union, Tuple, Callable from dataclasses import dataclass, field from enum import Enum import re import time # Try to import optional dependencies try: import yaml HAS_YAML = True except ImportError: HAS_YAML = False try: from PIL import Image HAS_PIL = True except ImportError: HAS_PIL = False # Import asset types (with fallback if not available) try: from .asset_types import AssetType, AssetConstraints, get_asset_constraints, ASSET_TYPE_CONSTRAINTS except ImportError: # Define minimal types for standalone operation class AssetType: AUDIO_WAV = "audio/wav" IMAGE_PNG = "image/png" TEXT_JSON = "text/json" UNKNOWN = "unknown" ASSET_TYPE_CONSTRAINTS = {} def get_asset_constraints(asset_type): return None class ValidationSeverity(Enum): """Severity levels for validation issues.""" INFO = "info" WARNING = "warning" ERROR = "error" CRITICAL = "critical" class ValidationCategory(Enum): """Categories of validation checks.""" FORMAT = "format" INTEGRITY = "integrity" CONTENT = "content" SECURITY = "security" PERFORMANCE = "performance" METADATA = "metadata" @dataclass class ValidationIssue: """Represents a single validation issue.""" category: ValidationCategory severity: ValidationSeverity message: str code: Optional[str] = None location: Optional[str] = None # Line number, byte offset, etc. suggestion: Optional[str] = None # How to fix the issue def __str__(self) -> str: parts = [f"[{self.severity.value.upper()}]"] if self.category: parts.append(f"[{self.category.value}]") parts.append(self.message) if self.location: parts.append(f"at {self.location}") return " ".join(parts) @dataclass class AssetValidationResult: """Results of asset validation.""" asset_path: str asset_type: Optional[str] = None is_valid: bool = True validation_time: float = 0.0 file_size: int = 0 checksum: Optional[str] = None # Issues found during validation issues: List[ValidationIssue] = field(default_factory=list) # Extracted metadata metadata: Dict[str, Any] = field(default_factory=dict) # Performance metrics load_time: Optional[float] = None parse_time: Optional[float] = None @property def errors(self) -> List[ValidationIssue]: """Get all error-level issues.""" return [issue for issue in self.issues if issue.severity in (ValidationSeverity.ERROR, ValidationSeverity.CRITICAL)] @property def warnings(self) -> List[ValidationIssue]: """Get all warning-level issues.""" return [issue for issue in self.issues if issue.severity == ValidationSeverity.WARNING] @property def has_errors(self) -> bool: """Check if validation has any errors.""" return len(self.errors) > 0 @property def has_warnings(self) -> bool: """Check if validation has any warnings.""" return len(self.warnings) > 0 def add_issue(self, category: ValidationCategory, severity: ValidationSeverity, message: str, code: str = None, location: str = None, suggestion: str = None): """Add a validation issue.""" issue = ValidationIssue( category=category, severity=severity, message=message, code=code, location=location, suggestion=suggestion ) self.issues.append(issue) # Update is_valid flag if severity in (ValidationSeverity.ERROR, ValidationSeverity.CRITICAL): self.is_valid = False def to_dict(self) -> Dict[str, Any]: """Convert result to dictionary.""" return { 'asset_path': self.asset_path, 'asset_type': self.asset_type, 'is_valid': self.is_valid, 'validation_time': self.validation_time, 'file_size': self.file_size, 'checksum': self.checksum, 'issues': [ { 'category': issue.category.value, 'severity': issue.severity.value, 'message': issue.message, 'code': issue.code, 'location': issue.location, 'suggestion': issue.suggestion } for issue in self.issues ], 'metadata': self.metadata, 'load_time': self.load_time, 'parse_time': self.parse_time, 'error_count': len(self.errors), 'warning_count': len(self.warnings) } def pprint(message: str) -> None: """Asset validator logging function.""" print(f"[ASSET_VALIDATOR] {message}") class AssetValidator: """ Comprehensive asset validation system. Provides format-specific validation, integrity checking, and content analysis for all supported asset types in the Trixy system. """ def __init__(self, strict_mode: bool = False, enable_security_checks: bool = True): """ Initialize the asset validator. Args: strict_mode: Whether to treat warnings as errors enable_security_checks: Whether to enable security scanning """ self.strict_mode = strict_mode self.enable_security_checks = enable_security_checks # Register built-in validators self._validators: Dict[str, Callable] = {} self._register_built_in_validators() # Security patterns for basic scanning self._security_patterns = self._compile_security_patterns() pprint(f"AssetValidator initialized (strict_mode: {strict_mode}, security_checks: {enable_security_checks})") def _register_built_in_validators(self): """Register built-in format validators.""" self._validators.update({ 'audio/wav': self._validate_wav_audio, 'audio/mp3': self._validate_mp3_audio, 'audio/ogg': self._validate_ogg_audio, 'image/png': self._validate_png_image, 'image/jpeg': self._validate_jpeg_image, 'image/gif': self._validate_gif_image, 'text/json': self._validate_json_text, 'text/yaml': self._validate_yaml_text, 'text/plain': self._validate_plain_text, 'config/json': self._validate_json_config, 'config/yaml': self._validate_yaml_config, 'model/pytorch': self._validate_pytorch_model, 'script/python': self._validate_python_script, }) def _compile_security_patterns(self) -> Dict[str, List[re.Pattern]]: """Compile security scan patterns.""" return { 'script_injection': [ re.compile(r']*>.*?', re.IGNORECASE | re.DOTALL), re.compile(r'javascript:', re.IGNORECASE), re.compile(r'on\w+\s*=', re.IGNORECASE), ], 'file_inclusion': [ re.compile(r'\.\./.*\.\./'), re.compile(r'file://'), re.compile(r'\\\\.*\\\\'), ], 'command_injection': [ re.compile(r'[;&|`$]'), re.compile(r'eval\s*\('), re.compile(r'exec\s*\('), ] } def validate_asset(self, file_path: str, asset_type: str = None) -> AssetValidationResult: """ Validate an asset file comprehensively. Args: file_path: Path to the asset file asset_type: Asset type (will be detected if None) Returns: AssetValidationResult with validation details """ start_time = time.time() result = AssetValidationResult( asset_path=file_path, asset_type=asset_type ) try: # Basic file validation if not self._validate_file_exists(file_path, result): result.validation_time = time.time() - start_time return result # Get file information path = Path(file_path) result.file_size = path.stat().st_size # Calculate checksum result.checksum = self._calculate_checksum(file_path) # Detect asset type if not provided if not asset_type: asset_type = self._detect_asset_type(file_path) result.asset_type = asset_type # Validate constraints self._validate_constraints(file_path, asset_type, result) # Format-specific validation if asset_type in self._validators: self._validators[asset_type](file_path, result) else: self._validate_generic_file(file_path, result) # Security checks if self.enable_security_checks: self._perform_security_checks(file_path, asset_type, result) # Performance analysis self._analyze_performance_characteristics(file_path, asset_type, result) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.CRITICAL, f"Validation failed with error: {str(e)}", code="VALIDATION_ERROR" ) result.validation_time = time.time() - start_time # Apply strict mode if self.strict_mode and result.has_warnings: result.is_valid = False return result def _validate_file_exists(self, file_path: str, result: AssetValidationResult) -> bool: """Validate that file exists and is accessible.""" path = Path(file_path) if not path.exists(): result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.CRITICAL, "File does not exist", code="FILE_NOT_FOUND" ) return False if not path.is_file(): result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, "Path is not a regular file", code="NOT_A_FILE" ) return False try: # Test file accessibility with open(file_path, 'rb') as f: f.read(1) # Try to read at least one byte except PermissionError: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, "File is not readable (permission denied)", code="PERMISSION_DENIED" ) return False except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"File access error: {str(e)}", code="FILE_ACCESS_ERROR" ) return False return True def _calculate_checksum(self, file_path: str) -> str: """Calculate SHA256 checksum of the file.""" hash_sha256 = hashlib.sha256() with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b""): hash_sha256.update(chunk) return hash_sha256.hexdigest() def _detect_asset_type(self, file_path: str) -> str: """Detect asset type from file path and content.""" # This would use the asset_types module if available # For now, use basic detection path = Path(file_path) extension = path.suffix.lower() extension_map = { '.wav': 'audio/wav', '.mp3': 'audio/mp3', '.png': 'image/png', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.json': 'text/json', '.yaml': 'text/yaml', '.yml': 'text/yaml', '.txt': 'text/plain' } return extension_map.get(extension, 'unknown') def _validate_constraints(self, file_path: str, asset_type: str, result: AssetValidationResult): """Validate asset against its type constraints.""" constraints = get_asset_constraints(asset_type) if get_asset_constraints else None if not constraints: return file_size = result.file_size # Check size limits if constraints.max_size_bytes and file_size > constraints.max_size_bytes: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.ERROR, f"File size ({file_size} bytes) exceeds maximum ({constraints.max_size_bytes} bytes)", code="FILE_TOO_LARGE", suggestion=f"Reduce file size to under {constraints.max_size_bytes} bytes" ) if constraints.min_size_bytes and file_size < constraints.min_size_bytes: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.ERROR, f"File size ({file_size} bytes) is below minimum ({constraints.min_size_bytes} bytes)", code="FILE_TOO_SMALL" ) # Check file extension if constraints.allowed_extensions: path = Path(file_path) extension = path.suffix.lower() if extension not in constraints.allowed_extensions: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.WARNING, f"File extension '{extension}' not in allowed list: {list(constraints.allowed_extensions)}", code="UNEXPECTED_EXTENSION" ) def _validate_wav_audio(self, file_path: str, result: AssetValidationResult): """Validate WAV audio file.""" try: with open(file_path, 'rb') as f: header = f.read(44) # WAV header is typically 44 bytes if len(header) < 44: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, "WAV file header too short", code="WAV_HEADER_TOO_SHORT" ) return # Check RIFF signature if not header.startswith(b'RIFF'): result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, "Invalid WAV file (missing RIFF signature)", code="WAV_INVALID_SIGNATURE" ) return # Check WAVE format if b'WAVE' not in header[:16]: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, "Invalid WAV file (missing WAVE format)", code="WAV_INVALID_FORMAT" ) return # Extract audio parameters try: # Audio format (2 bytes at offset 20) audio_format = struct.unpack(' 8: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.WARNING, f"Unusual channel count: {channels}", code="WAV_UNUSUAL_CHANNELS" ) if sample_rate < 8000 or sample_rate > 192000: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.WARNING, f"Unusual sample rate: {sample_rate} Hz", code="WAV_UNUSUAL_SAMPLE_RATE" ) if bits_per_sample not in [8, 16, 24, 32]: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.WARNING, f"Unusual bit depth: {bits_per_sample} bits", code="WAV_UNUSUAL_BIT_DEPTH" ) except struct.error as e: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, f"Error parsing WAV header: {str(e)}", code="WAV_HEADER_PARSE_ERROR" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating WAV file: {str(e)}", code="WAV_VALIDATION_ERROR" ) def _validate_mp3_audio(self, file_path: str, result: AssetValidationResult): """Validate MP3 audio file.""" try: with open(file_path, 'rb') as f: header = f.read(10) # Read first 10 bytes # Check for ID3 tag or MPEG sync if header.startswith(b'ID3'): result.metadata['has_id3_tag'] = True elif header.startswith((b'\xFF\xFB', b'\xFF\xFA')): result.metadata['has_mpeg_sync'] = True else: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.WARNING, "MP3 file missing expected signature", code="MP3_MISSING_SIGNATURE" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating MP3 file: {str(e)}", code="MP3_VALIDATION_ERROR" ) def _validate_ogg_audio(self, file_path: str, result: AssetValidationResult): """Validate OGG audio file.""" try: with open(file_path, 'rb') as f: header = f.read(4) if not header.startswith(b'OggS'): result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, "Invalid OGG file (missing OggS signature)", code="OGG_INVALID_SIGNATURE" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating OGG file: {str(e)}", code="OGG_VALIDATION_ERROR" ) def _validate_png_image(self, file_path: str, result: AssetValidationResult): """Validate PNG image file.""" try: with open(file_path, 'rb') as f: header = f.read(24) # PNG signature + IHDR chunk start # Check PNG signature if not header.startswith(b'\x89PNG\r\n\x1a\n'): result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, "Invalid PNG file signature", code="PNG_INVALID_SIGNATURE" ) return # Extract image dimensions if len(header) >= 24: width = struct.unpack('>I', header[16:20])[0] height = struct.unpack('>I', header[20:24])[0] result.metadata.update({ 'width': width, 'height': height, 'aspect_ratio': width / height if height > 0 else 0 }) # Validate dimensions if width == 0 or height == 0: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.ERROR, f"Invalid image dimensions: {width}x{height}", code="PNG_INVALID_DIMENSIONS" ) if width > 16384 or height > 16384: result.add_issue( ValidationCategory.PERFORMANCE, ValidationSeverity.WARNING, f"Very large image dimensions: {width}x{height}", code="PNG_LARGE_DIMENSIONS", suggestion="Consider reducing image size for better performance" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating PNG file: {str(e)}", code="PNG_VALIDATION_ERROR" ) def _validate_jpeg_image(self, file_path: str, result: AssetValidationResult): """Validate JPEG image file.""" try: with open(file_path, 'rb') as f: header = f.read(10) if not header.startswith(b'\xFF\xD8\xFF'): result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, "Invalid JPEG file signature", code="JPEG_INVALID_SIGNATURE" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating JPEG file: {str(e)}", code="JPEG_VALIDATION_ERROR" ) def _validate_gif_image(self, file_path: str, result: AssetValidationResult): """Validate GIF image file.""" try: with open(file_path, 'rb') as f: header = f.read(10) if not header.startswith((b'GIF87a', b'GIF89a')): result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, "Invalid GIF file signature", code="GIF_INVALID_SIGNATURE" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating GIF file: {str(e)}", code="GIF_VALIDATION_ERROR" ) def _validate_json_text(self, file_path: str, result: AssetValidationResult): """Validate JSON text file.""" try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() try: data = json.loads(content) result.metadata.update({ 'json_valid': True, 'json_type': type(data).__name__, 'character_count': len(content) }) if isinstance(data, dict): result.metadata['key_count'] = len(data) elif isinstance(data, list): result.metadata['item_count'] = len(data) except json.JSONDecodeError as e: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.ERROR, f"Invalid JSON syntax: {str(e)}", code="JSON_SYNTAX_ERROR", location=f"line {e.lineno}, column {e.colno}" ) except UnicodeDecodeError as e: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, f"Text encoding error: {str(e)}", code="TEXT_ENCODING_ERROR" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating JSON file: {str(e)}", code="JSON_VALIDATION_ERROR" ) def _validate_yaml_text(self, file_path: str, result: AssetValidationResult): """Validate YAML text file.""" if not HAS_YAML: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.WARNING, "YAML validation skipped (PyYAML not available)", code="YAML_VALIDATOR_UNAVAILABLE" ) return try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() try: data = yaml.safe_load(content) result.metadata.update({ 'yaml_valid': True, 'yaml_type': type(data).__name__, 'character_count': len(content) }) except yaml.YAMLError as e: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.ERROR, f"Invalid YAML syntax: {str(e)}", code="YAML_SYNTAX_ERROR" ) except UnicodeDecodeError as e: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, f"Text encoding error: {str(e)}", code="TEXT_ENCODING_ERROR" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating YAML file: {str(e)}", code="YAML_VALIDATION_ERROR" ) def _validate_plain_text(self, file_path: str, result: AssetValidationResult): """Validate plain text file.""" try: encodings = ['utf-8', 'utf-16', 'latin-1', 'cp1252'] content = None encoding_used = None for encoding in encodings: try: with open(file_path, 'r', encoding=encoding) as f: content = f.read() encoding_used = encoding break except UnicodeDecodeError: continue if content is not None: result.metadata.update({ 'encoding': encoding_used, 'character_count': len(content), 'line_count': content.count('\n') + 1, 'word_count': len(content.split()), 'has_unicode': any(ord(char) > 127 for char in content) }) else: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, "Could not decode text file with any standard encoding", code="TEXT_ENCODING_UNKNOWN" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating text file: {str(e)}", code="TEXT_VALIDATION_ERROR" ) def _validate_json_config(self, file_path: str, result: AssetValidationResult): """Validate JSON configuration file.""" # Start with basic JSON validation self._validate_json_text(file_path, result) if result.is_valid: # Additional config-specific checks could be added here # For example, schema validation, required fields, etc. pass def _validate_yaml_config(self, file_path: str, result: AssetValidationResult): """Validate YAML configuration file.""" # Start with basic YAML validation self._validate_yaml_text(file_path, result) if result.is_valid: # Additional config-specific checks could be added here pass def _validate_pytorch_model(self, file_path: str, result: AssetValidationResult): """Validate PyTorch model file.""" try: # Basic file signature check for pickle/torch files with open(file_path, 'rb') as f: header = f.read(8) # PyTorch files are typically pickle files if not header.startswith(b'\x80'): # Pickle protocol marker result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.WARNING, "File may not be a valid PyTorch model (unexpected format)", code="PYTORCH_UNEXPECTED_FORMAT" ) # Try to load with torch if available try: import torch # Don't actually load the model, just check if it's loadable # This is a placeholder - actual implementation would be more careful result.metadata['pytorch_available'] = True except ImportError: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.WARNING, "PyTorch not available for model validation", code="PYTORCH_UNAVAILABLE" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating PyTorch model: {str(e)}", code="PYTORCH_VALIDATION_ERROR" ) def _validate_python_script(self, file_path: str, result: AssetValidationResult): """Validate Python script file.""" try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # Check Python syntax try: compile(content, file_path, 'exec') result.metadata['python_syntax_valid'] = True except SyntaxError as e: result.add_issue( ValidationCategory.CONTENT, ValidationSeverity.ERROR, f"Python syntax error: {str(e)}", code="PYTHON_SYNTAX_ERROR", location=f"line {e.lineno}" ) # Basic metrics lines = content.split('\n') result.metadata.update({ 'line_count': len(lines), 'character_count': len(content), 'has_shebang': content.startswith('#!') }) except UnicodeDecodeError as e: result.add_issue( ValidationCategory.FORMAT, ValidationSeverity.ERROR, f"Python script encoding error: {str(e)}", code="PYTHON_ENCODING_ERROR" ) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error validating Python script: {str(e)}", code="PYTHON_VALIDATION_ERROR" ) def _validate_generic_file(self, file_path: str, result: AssetValidationResult): """Generic validation for unknown file types.""" try: # Basic file information path = Path(file_path) result.metadata.update({ 'file_extension': path.suffix, 'mime_type': mimetypes.guess_type(str(path))[0] }) # Check if file is readable with open(file_path, 'rb') as f: first_bytes = f.read(1024) # Read first 1KB result.metadata.update({ 'is_binary': b'\x00' in first_bytes, 'first_bytes_length': len(first_bytes) }) except Exception as e: result.add_issue( ValidationCategory.INTEGRITY, ValidationSeverity.ERROR, f"Error during generic validation: {str(e)}", code="GENERIC_VALIDATION_ERROR" ) def _perform_security_checks(self, file_path: str, asset_type: str, result: AssetValidationResult): """Perform basic security checks on the asset.""" if not self.enable_security_checks: return try: # Only scan text-based files for security issues if asset_type.startswith(('text/', 'config/', 'script/')): self._scan_text_security(file_path, result) # Check file permissions (if available) self._check_file_permissions(file_path, result) except Exception as e: result.add_issue( ValidationCategory.SECURITY, ValidationSeverity.WARNING, f"Security scan error: {str(e)}", code="SECURITY_SCAN_ERROR" ) def _scan_text_security(self, file_path: str, result: AssetValidationResult): """Scan text files for security issues.""" try: with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: content = f.read(10240) # Read first 10KB for scanning for category, patterns in self._security_patterns.items(): for pattern in patterns: matches = pattern.findall(content) if matches: result.add_issue( ValidationCategory.SECURITY, ValidationSeverity.WARNING, f"Potential {category.replace('_', ' ')} detected: {len(matches)} matches", code=f"SECURITY_{category.upper()}", suggestion="Review content for security implications" ) except Exception: # Ignore errors in security scanning pass def _check_file_permissions(self, file_path: str, result: AssetValidationResult): """Check file permissions for security issues.""" try: path = Path(file_path) stat = path.stat() # Check for world-writable files if stat.st_mode & 0o002: result.add_issue( ValidationCategory.SECURITY, ValidationSeverity.WARNING, "File is world-writable", code="SECURITY_WORLD_WRITABLE", suggestion="Consider restricting file permissions" ) result.metadata['file_permissions'] = oct(stat.st_mode)[-3:] except Exception: # Ignore permission check errors pass def _analyze_performance_characteristics(self, file_path: str, asset_type: str, result: AssetValidationResult): """Analyze performance characteristics of the asset.""" try: file_size = result.file_size # General size-based performance warnings if file_size > 50 * 1024 * 1024: # 50MB result.add_issue( ValidationCategory.PERFORMANCE, ValidationSeverity.WARNING, f"Large file size ({file_size / 1024 / 1024:.1f} MB) may impact loading performance", code="PERFORMANCE_LARGE_FILE", suggestion="Consider optimizing or compressing the asset" ) # Type-specific performance analysis if asset_type.startswith('image/'): self._analyze_image_performance(file_path, result) elif asset_type.startswith('audio/'): self._analyze_audio_performance(file_path, result) except Exception: # Ignore performance analysis errors pass def _analyze_image_performance(self, file_path: str, result: AssetValidationResult): """Analyze image performance characteristics.""" try: if 'width' in result.metadata and 'height' in result.metadata: width = result.metadata['width'] height = result.metadata['height'] pixel_count = width * height if pixel_count > 16 * 1024 * 1024: # 16 megapixels result.add_issue( ValidationCategory.PERFORMANCE, ValidationSeverity.WARNING, f"High resolution image ({width}x{height}) may impact performance", code="PERFORMANCE_HIGH_RESOLUTION", suggestion="Consider resizing for better performance" ) except Exception: pass def _analyze_audio_performance(self, file_path: str, result: AssetValidationResult): """Analyze audio performance characteristics.""" try: if 'sample_rate' in result.metadata and 'channels' in result.metadata: sample_rate = result.metadata['sample_rate'] channels = result.metadata['channels'] if sample_rate > 48000: result.add_issue( ValidationCategory.PERFORMANCE, ValidationSeverity.INFO, f"High sample rate ({sample_rate} Hz) may be overkill for voice assistant", code="PERFORMANCE_HIGH_SAMPLE_RATE", suggestion="Consider using 44.1kHz or 48kHz for better performance" ) if channels > 2: result.add_issue( ValidationCategory.PERFORMANCE, ValidationSeverity.INFO, f"Multi-channel audio ({channels} channels) may not be needed", code="PERFORMANCE_MULTI_CHANNEL", suggestion="Consider using mono or stereo audio" ) except Exception: pass class AssetIntegrityChecker: """ Specialized integrity checker for assets. Provides deeper integrity checking including checksum verification, format consistency, and corruption detection. """ def __init__(self): """Initialize the integrity checker.""" self._checksum_cache: Dict[str, Tuple[str, float]] = {} # path -> (checksum, mtime) pprint("AssetIntegrityChecker initialized") def check_integrity(self, file_path: str, expected_checksum: str = None) -> bool: """ Check file integrity using checksums. Args: file_path: Path to the file to check expected_checksum: Expected SHA256 checksum (optional) Returns: True if integrity check passes, False otherwise """ try: current_checksum = self._get_checksum(file_path) if expected_checksum: return current_checksum == expected_checksum else: # If no expected checksum, just verify we can calculate one return current_checksum is not None except Exception: return False def _get_checksum(self, file_path: str) -> str: """Get cached or calculate checksum for a file.""" try: path = Path(file_path) mtime = path.stat().st_mtime # Check cache if file_path in self._checksum_cache: cached_checksum, cached_mtime = self._checksum_cache[file_path] if cached_mtime == mtime: return cached_checksum # Calculate new checksum hash_sha256 = hashlib.sha256() with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b""): hash_sha256.update(chunk) checksum = hash_sha256.hexdigest() # Update cache self._checksum_cache[file_path] = (checksum, mtime) return checksum except Exception: return None def clear_cache(self): """Clear the checksum cache.""" self._checksum_cache.clear() # Convenience functions def quick_validate(file_path: str, asset_type: str = None, strict: bool = False) -> AssetValidationResult: """ Quick validation of an asset file. Args: file_path: Path to the asset file asset_type: Asset type (will be detected if None) strict: Whether to use strict mode Returns: AssetValidationResult """ validator = AssetValidator(strict_mode=strict) return validator.validate_asset(file_path, asset_type) def batch_validate(file_paths: List[str], asset_types: List[str] = None) -> List[AssetValidationResult]: """ Validate multiple assets in batch. Args: file_paths: List of file paths to validate asset_types: List of asset types (optional) Returns: List of AssetValidationResult objects """ validator = AssetValidator() results = [] for i, file_path in enumerate(file_paths): asset_type = asset_types[i] if asset_types and i < len(asset_types) else None result = validator.validate_asset(file_path, asset_type) results.append(result) return results # Module exports __all__ = [ 'AssetValidator', 'AssetIntegrityChecker', 'AssetValidationResult', 'ValidationIssue', 'ValidationSeverity', 'ValidationCategory', 'quick_validate', 'batch_validate', 'pprint' ]