""" Report Validator for AegisLM Reporting System Production-grade report integrity verification with comprehensive validation checks. """ import json import hashlib from datetime import datetime from typing import Dict, Any, Optional, List from pathlib import Path import sys # Add parent directory to path for imports current_dir = Path(__file__).parent backend_dir = current_dir.parent if str(backend_dir) not in sys.path: sys.path.insert(0, str(backend_dir)) from schemas.report_schema import ( ReportValidationRequest, ReportValidationResult, ReportFormat, FullReport, SummaryReport, CSVReportData, ReportMetadata ) class ReportValidator: """ Production-grade report validator for integrity verification. Validates report authenticity, integrity, and compliance. """ def __init__(self, reports_dir: Optional[str] = None): """ Initialize report validator. Args: reports_dir: Directory where reports are stored """ if reports_dir is None: # Default to backend/reports relative to this file current_file = Path(__file__) self.reports_dir = current_file.parent.parent / "reports" else: self.reports_dir = Path(reports_dir) def verify_report(self, request: ReportValidationRequest) -> ReportValidationResult: """ Verify report integrity and authenticity. Args: request: Report validation request Returns: Validation result Raises: ValueError: If validation request is invalid """ try: validation_timestamp = datetime.utcnow() # Determine validation source if request.file_path: return self._verify_file(request, validation_timestamp) elif request.file_content: return self._verify_content(request, validation_timestamp) elif request.report_id: return self._verify_report_id(request, validation_timestamp) else: raise ValueError("No report source provided in validation request") except Exception as e: return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, issues=[f"Validation failed: {str(e)}"], recommendations=["Check report source and format"] ) def verify_file_integrity(self, file_path: str) -> bool: """ Quick file integrity check. Args: file_path: Path to report file Returns: True if file is valid, False otherwise """ try: request = ReportValidationRequest(file_path=file_path) result = self.verify_report(request) return result.is_valid except Exception: return False def verify_batch_reports(self, file_paths: List[str]) -> List[ReportValidationResult]: """ Verify multiple reports in batch. Args: file_paths: List of report file paths Returns: List of validation results """ results = [] for file_path in file_paths: try: request = ReportValidationRequest(file_path=file_path) result = self.verify_report(request) results.append(result) except Exception as e: # Create failed result for this file results.append(ReportValidationResult( request=ReportValidationRequest(file_path=file_path), is_valid=False, validation_timestamp=datetime.utcnow(), issues=[f"Validation error: {str(e)}"], recommendations=["Check file format and integrity"] )) return results def _verify_file(self, request: ReportValidationRequest, validation_timestamp: datetime) -> ReportValidationResult: """Verify report from file path.""" file_path = Path(request.file_path) if not file_path.exists(): return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, format_valid=False, issues=[f"File not found: {request.file_path}"], recommendations=["Check file path and ensure file exists"] ) try: # Read file content with open(file_path, 'r', encoding='utf-8') as f: if file_path.suffix.lower() == '.json': content = f.read() elif file_path.suffix.lower() == '.csv': content = f.read() else: return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, format_valid=False, issues=[f"Unsupported file format: {file_path.suffix}"], recommendations=["Use JSON or CSV format"] ) # Update request with content request.file_content = content # Verify content return self._verify_content(request, validation_timestamp) except Exception as e: return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, format_valid=False, issues=[f"Failed to read file: {str(e)}"], recommendations=["Check file permissions and format"] ) def _verify_content(self, request: ReportValidationRequest, validation_timestamp: datetime) -> ReportValidationResult: """Verify report from content.""" try: content = request.file_content if not content: return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, format_valid=False, issues=["Empty file content"], recommendations=["Check file content"] ) # Parse content based on format if content.strip().startswith('{'): # JSON format return self._verify_json_content(request, validation_timestamp) else: # CSV format return self._verify_csv_content(request, validation_timestamp) except Exception as e: return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, format_valid=False, issues=[f"Content parsing error: {str(e)}"], recommendations=["Check file format and encoding"] ) def _verify_json_content(self, request: ReportValidationRequest, validation_timestamp: datetime) -> ReportValidationResult: """Verify JSON report content.""" issues = [] warnings = [] recommendations = [] try: # Parse JSON data = json.loads(request.file_content) # Check for export metadata if 'export_metadata' not in data: issues.append("Missing export metadata") recommendations.append("Ensure report includes export metadata") format_valid = False else: format_valid = True metadata = data['export_metadata'] # Validate metadata if 'exported_at' not in metadata: issues.append("Missing export timestamp") if 'run_id' not in metadata: issues.append("Missing run_id in metadata") if 'file_checksum' in metadata: # Verify file checksum expected_checksum = metadata['file_checksum'] actual_checksum = self._calculate_content_checksum(request.file_content) if expected_checksum != actual_checksum: issues.append("File checksum mismatch") recommendations.append("Report may have been tampered with") integrity_valid = False else: integrity_valid = True else: integrity_valid = True warnings.append("No file checksum for integrity verification") # Check expiration if requested not_expired = True if request.check_expiration and 'expires_at' in metadata: expires_at = datetime.fromisoformat(metadata['expires_at']) if expires_at < validation_timestamp: issues.append("Report has expired") not_expired = False recommendations.append("Generate a fresh report") # Validate report data if 'report_data' not in data: issues.append("Missing report data") recommendations.append("Ensure report contains data") else: report_data = data['report_data'] # Validate report structure structure_valid = self._validate_report_structure(report_data, issues, warnings) # Extract and validate hashes config_hash_match, result_checksum_match = self._validate_hashes(report_data, issues) # Extract report data for return report_data_for_result = self._extract_report_data(report_data) # Determine overall validity is_valid = (format_valid and integrity_valid and not_expired and len(issues) == 0 and structure_valid) return ReportValidationResult( request=request, is_valid=is_valid, validation_timestamp=validation_timestamp, format_valid=format_valid, integrity_valid=integrity_valid, signature_valid=None, # Not implemented yet not_expired=not_expired, config_hash_match=config_hash_match, result_checksum_match=result_checksum_match, report_data=report_data_for_result if is_valid else None, issues=issues, warnings=warnings, recommendations=recommendations ) except json.JSONDecodeError as e: return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, format_valid=False, issues=[f"Invalid JSON format: {str(e)}"], recommendations=["Check JSON syntax and structure"] ) def _verify_csv_content(self, request: ReportValidationRequest, validation_timestamp: datetime) -> ReportValidationResult: """Verify CSV report content.""" issues = [] warnings = [] recommendations = [] try: lines = request.file_content.strip().split('\n') if len(lines) < 2: issues.append("CSV file has no data rows") recommendations.append("Ensure CSV has header and data") return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, format_valid=False, issues=issues, recommendations=recommendations ) # Parse CSV header header = lines[0].split(',') expected_columns = ['run_id', 'model_name', 'dataset_name', 'attack_types', 'created_at', 'status', 'total_attacks', 'robustness_score', 'risk_score', 'config_hash', 'result_checksum', 'audit_status'] missing_columns = [col for col in expected_columns if col not in header] if missing_columns: issues.append(f"Missing CSV columns: {', '.join(missing_columns)}") recommendations.append("Ensure CSV has all required columns") format_valid = False else: format_valid = True # Validate data rows data_rows = [] for i, line in enumerate(lines[1:], 1): if not line.strip(): continue values = line.split(',') if len(values) != len(header): issues.append(f"Row {i} has {len(values)} values, expected {len(header)}") continue row_data = dict(zip(header, values)) data_rows.append(row_data) # Validate hashes in data rows config_hash_match = True result_checksum_match = True for i, row in enumerate(data_rows): run_id = row.get('run_id', '') config_hash = row.get('config_hash', '') result_checksum = row.get('result_checksum', '') if not config_hash or not result_checksum: issues.append(f"Row {i+1} missing hash information") config_hash_match = False result_checksum_match = False elif len(config_hash) != 64 or len(result_checksum) != 64: issues.append(f"Row {i+1} has invalid hash format") config_hash_match = False result_checksum_match = False # Check for metadata file metadata_path = None if request.file_path: file_path = Path(request.file_path) metadata_path = file_path.with_suffix('.json') if metadata_path.exists(): try: with open(metadata_path, 'r', encoding='utf-8') as f: metadata = json.load(f) # Verify metadata checksum if 'file_checksum' in metadata.get('export_metadata', {}): expected_checksum = metadata['export_metadata']['file_checksum'] actual_checksum = self._calculate_content_checksum(request.file_content) if expected_checksum != actual_checksum: issues.append("CSV file checksum mismatch") integrity_valid = False else: integrity_valid = True else: integrity_valid = True warnings.append("No checksum in metadata file") except Exception: integrity_valid = False issues.append("Failed to read metadata file") else: integrity_valid = True warnings.append("No metadata file found") else: integrity_valid = True warnings.append("No file path provided for metadata check") # Determine overall validity is_valid = format_valid and integrity_valid and len(issues) == 0 return ReportValidationResult( request=request, is_valid=is_valid, validation_timestamp=validation_timestamp, format_valid=format_valid, integrity_valid=integrity_valid, signature_valid=None, not_expired=True, # CSV reports don't expire config_hash_match=config_hash_match, result_checksum_match=result_checksum_match, report_data={'csv_rows': data_rows} if is_valid else None, issues=issues, warnings=warnings, recommendations=recommendations ) except Exception as e: return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, format_valid=False, issues=[f"CSV parsing error: {str(e)}"], recommendations=["Check CSV format and encoding"] ) def _verify_report_id(self, request: ReportValidationRequest, validation_timestamp: datetime) -> ReportValidationResult: """Verify report by report ID.""" # This would typically involve a database lookup # For now, return a placeholder result return ReportValidationResult( request=request, is_valid=False, validation_timestamp=validation_timestamp, issues=["Report ID verification not implemented"], recommendations=["Use file path or content for verification"] ) def _validate_report_structure(self, report_data: Dict[str, Any], issues: List[str], warnings: List[str]) -> bool: """Validate report structure.""" structure_valid = True # Check for required sections if 'experiment' not in report_data: issues.append("Missing experiment section") structure_valid = False if 'audit' not in report_data: issues.append("Missing audit section") structure_valid = False # Validate experiment section if 'experiment' in report_data: experiment = report_data['experiment'] required_fields = ['run_id', 'model_name', 'attack_types', 'status', 'created_at'] for field in required_fields: if field not in experiment: issues.append(f"Missing required field in experiment: {field}") structure_valid = False # Validate audit section if 'audit' in report_data: audit = report_data['audit'] required_fields = ['run_id', 'config_hash', 'result_checksum', 'audit_status'] for field in required_fields: if field not in audit: issues.append(f"Missing required field in audit: {field}") structure_valid = False return structure_valid def _validate_hashes(self, report_data: Dict[str, Any], issues: List[str]) -> tuple[bool, bool]: """Validate configuration and result hashes.""" config_hash_match = True result_checksum_match = True try: # Extract experiment and audit data experiment = report_data.get('experiment', {}) audit = report_data.get('audit', {}) # Get stored hashes stored_config_hash = audit.get('config_hash', '') stored_result_checksum = audit.get('result_checksum', '') # Calculate current hashes config_snapshot = report_data.get('config_snapshot', {}) full_result = report_data.get('full_result', {}) if config_snapshot and stored_config_hash: current_config_hash = self._calculate_config_hash(config_snapshot) if current_config_hash != stored_config_hash: issues.append("Configuration hash mismatch") config_hash_match = False if full_result and stored_result_checksum: current_result_checksum = self._calculate_result_checksum(full_result) if current_result_checksum != stored_result_checksum: issues.append("Result checksum mismatch") result_checksum_match = False except Exception as e: issues.append(f"Hash validation error: {str(e)}") config_hash_match = False result_checksum_match = False return config_hash_match, result_checksum_match def _extract_report_data(self, report_data: Dict[str, Any]) -> Dict[str, Any]: """Extract relevant report data for return.""" extracted = { 'experiment': report_data.get('experiment', {}), 'audit': report_data.get('audit', {}), 'export_metadata': report_data.get('export_metadata', {}) } # Include optional data if 'analysis' in report_data: extracted['analysis'] = report_data['analysis'] if 'config_snapshot' in report_data: extracted['config_snapshot'] = report_data['config_snapshot'] if 'full_result' in report_data: extracted['full_result'] = report_data['full_result'] return extracted def _calculate_content_checksum(self, content: str) -> str: """Calculate SHA-256 checksum of content.""" hash_sha256 = hashlib.sha256() hash_sha256.update(content.encode('utf-8')) return hash_sha256.hexdigest() def _calculate_config_hash(self, config: Dict[str, Any]) -> str: """Calculate configuration hash.""" # Normalize configuration normalized_config = json.dumps(config, sort_keys=True, separators=(',', ':')) # Calculate hash hash_sha256 = hashlib.sha256() hash_sha256.update(normalized_config.encode('utf-8')) return hash_sha256.hexdigest() def _calculate_result_checksum(self, result: Dict[str, Any]) -> str: """Calculate result checksum.""" # Normalize result normalized_result = json.dumps(result, sort_keys=True, separators=(',', ':'), default=str) # Calculate hash hash_sha256 = hashlib.sha256() hash_sha256.update(normalized_result.encode('utf-8')) return hash_sha256.hexdigest() def validate_report_format(self, file_path: str) -> Dict[str, Any]: """ Validate report format and return detailed information. Args: file_path: Path to report file Returns: Validation information """ try: request = ReportValidationRequest(file_path=file_path) result = self.verify_report(request) return { 'is_valid': result.is_valid, 'format': 'json' if file_path.endswith('.json') else 'csv', 'validation_timestamp': result.validation_timestamp.isoformat(), 'issues': result.issues, 'warnings': result.warnings, 'recommendations': result.recommendations, 'integrity_verified': result.integrity_valid, 'hashes_match': { 'config': result.config_hash_match, 'result': result.result_checksum_match } } except Exception as e: return { 'is_valid': False, 'format': 'unknown', 'validation_timestamp': datetime.utcnow().isoformat(), 'issues': [f"Validation error: {str(e)}"], 'warnings': [], 'recommendations': ["Check file format and permissions"], 'integrity_verified': False, 'hashes_match': {'config': False, 'result': False} } def get_validation_summary(self, results: List[ReportValidationResult]) -> Dict[str, Any]: """ Get summary of validation results. Args: results: List of validation results Returns: Validation summary """ total_reports = len(results) valid_reports = sum(1 for r in results if r.is_valid) invalid_reports = total_reports - valid_reports # Count issues all_issues = [] for result in results: all_issues.extend(result.issues) # Count warnings all_warnings = [] for result in results: all_warnings.extend(result.warnings) # Common issues issue_counts = {} for issue in all_issues: issue_counts[issue] = issue_counts.get(issue, 0) + 1 # Sort by frequency common_issues = sorted(issue_counts.items(), key=lambda x: x[1], reverse=True)[:5] return { 'total_reports': total_reports, 'valid_reports': valid_reports, 'invalid_reports': invalid_reports, 'validation_rate': valid_reports / total_reports if total_reports > 0 else 0, 'total_issues': len(all_issues), 'total_warnings': len(all_warnings), 'common_issues': common_issues, 'validation_timestamp': datetime.utcnow().isoformat() } # Global instance for easy access _report_validator = None def get_report_validator() -> ReportValidator: """ Get the global report validator instance. Returns: Global report validator instance """ global _report_validator if _report_validator is None: _report_validator = ReportValidator() return _report_validator # Convenience functions for direct usage def verify_report(file_path: str, verify_integrity: bool = True, check_expiration: bool = True) -> ReportValidationResult: """ Verify report integrity. Args: file_path: Path to report file verify_integrity: Whether to verify integrity check_expiration: Whether to check expiration Returns: Validation result """ validator = get_report_validator() request = ReportValidationRequest( file_path=file_path, verify_integrity=verify_integrity, check_expiration=check_expiration ) return validator.verify_report(request) def verify_report_integrity(file_path: str) -> bool: """ Quick integrity verification. Args: file_path: Path to report file Returns: True if integrity verified, False otherwise """ validator = get_report_validator() return validator.verify_file_integrity(file_path)