#!/usr/bin/env python3 """ Genesis Validation Gate 3 - Safety Validation ============================================== Validates safety, permissions, and real-time constraints. Real-time layer enforcing P2 (Currency), P7 (Hallucination), P9 (Adaptive). PM-044: Validation Gate 3 - Safety Enhancement - Validates safety and permission compliance - Blocks dangerous operations - Logs all security events """ import re import logging from typing import Dict, Any, List, Tuple, Optional from datetime import datetime from dataclasses import dataclass, asdict logger = logging.getLogger("ValidationGate3") logging.basicConfig(level=logging.INFO) @dataclass class SafetyResult: """Result of safety validation.""" valid: bool score: float checks: Dict[str, bool] blocked_operations: List[str] security_events: List[Dict[str, Any]] timestamp: str def to_dict(self) -> Dict[str, Any]: return asdict(self) class SafetyValidationGate: """ Gate 3: Safety and Security Layer Validates safety, permissions, and blocks dangerous operations. Enforces P2 (Currency), P7 (Hallucination), P9 (Adaptive thresholds). """ # Dangerous operation patterns DANGEROUS_PATTERNS = { "file_deletion": [ r'\brm\s+-rf?\b', r'\bdel\s+/[sqa]\b', r'unlink\s*\(', r'os\.remove\s*\(', r'shutil\.rmtree\s*\(' ], "code_execution": [ r'\bexec\s*\(', r'\beval\s*\(', r'subprocess\.(?:run|call|Popen)', r'os\.system\s*\(', r'os\.popen\s*\(' ], "credential_exposure": [ r'password\s*[=:]\s*["\'][^"\']+["\']', r'api[_-]?key\s*[=:]\s*["\'][^"\']+["\']', r'secret\s*[=:]\s*["\'][^"\']+["\']', r'token\s*[=:]\s*["\'][^"\']+["\']' ], "network_exfiltration": [ r'curl\s+.*\s+-d\s+', r'wget\s+.*--post', r'requests\.post\s*\(.*(password|secret|token)', r'socket\.connect' ], "privilege_escalation": [ r'\bsudo\b', r'\bsu\s+-', r'chmod\s+777', r'chown\s+root' ], "database_destruction": [ r'\bDROP\s+(?:TABLE|DATABASE)\b', r'\bTRUNCATE\s+TABLE\b', r'\bDELETE\s+FROM\s+\w+\s*;?\s*$' # DELETE without WHERE ] } # Permission levels PERMISSION_LEVELS = { "read": 1, "write": 2, "execute": 3, "admin": 4, "root": 5 } # Action risk levels ACTION_RISKS = { "read_file": "low", "write_file": "medium", "delete_file": "high", "execute_code": "high", "network_request": "medium", "database_query": "medium", "database_modify": "high", "system_command": "critical" } def __init__(self, memory_stores=None, permission_level: int = 2): """ Initialize safety validation gate. Args: memory_stores: Optional memory backend reference permission_level: Default permission level (1-5) """ self.memory = memory_stores self.permission_level = permission_level self.validation_log: List[SafetyResult] = [] self.security_events: List[Dict[str, Any]] = [] self.blocked_count = 0 self.adaptive_threshold = 0.8 # Can be adjusted based on history logger.info(f"Validation Gate 3 (Safety) initialized (permission_level={permission_level})") def validate(self, action: Any, worker_id: str = None, metadata: Dict[str, Any] = None) -> Tuple[float, Dict[str, bool]]: """ Validate action for safety and permission compliance. Args: action: The action/output to validate (string or dict) worker_id: ID of the worker performing the action metadata: Additional context including permissions Returns: Tuple of (score, checks_dict) """ metadata = metadata or {} blocked_operations = [] security_events = [] checks = {} # Convert action to string for pattern matching action_str = str(action) if not isinstance(action, str) else action # P7: Hallucination detection checks["no_hallucinations"] = self._detect_hallucinations(action_str) # P2: Information currency check checks["information_current"] = self._check_currency(action_str) # P9: Adaptive threshold check checks["threshold_met"] = self._check_adaptive_threshold(action_str, metadata) # Safety: Dangerous operation detection dangerous_ops = self._detect_dangerous_operations(action_str) checks["no_dangerous_ops"] = len(dangerous_ops) == 0 if dangerous_ops: for op_type, patterns in dangerous_ops.items(): blocked_operations.append(op_type) self._log_security_event("dangerous_operation_detected", { "type": op_type, "worker_id": worker_id, "patterns_matched": len(patterns) }) # Safety: Permission check required_permission = metadata.get("required_permission", "write") checks["permission_granted"] = self._check_permission(required_permission, metadata) if not checks["permission_granted"]: self._log_security_event("permission_denied", { "required": required_permission, "worker_id": worker_id, "current_level": self.permission_level }) # Safety: Rate limiting check checks["rate_limit_ok"] = self._check_rate_limit(worker_id) # Safety: Credential exposure check checks["no_credential_exposure"] = self._check_credential_exposure(action_str) if not checks["no_credential_exposure"]: self._log_security_event("credential_exposure_attempt", { "worker_id": worker_id, "blocked": True }) blocked_operations.append("credential_exposure") # Safety: Injection prevention checks["no_injection"] = self._check_injection(action_str) if not checks["no_injection"]: self._log_security_event("injection_attempt", { "worker_id": worker_id, "blocked": True }) blocked_operations.append("injection_attempt") # Calculate score score = sum(checks.values()) / len(checks) # Create result result = SafetyResult( valid=score >= self.adaptive_threshold and len(blocked_operations) == 0, score=score, checks=checks, blocked_operations=blocked_operations, security_events=self.security_events[-5:], # Last 5 events timestamp=datetime.utcnow().isoformat() ) # Update blocked count if blocked_operations: self.blocked_count += 1 # Log result self._log_validation(result, worker_id) return score, checks def _detect_hallucinations(self, text: str) -> bool: """ Check for hallucination indicators. P7: Hallucination detection. """ hallucination_indicators = [ r'I apologize, but I cannot', r'As an AI', r'\[citation needed\]', r'I do not have access to', r'I cannot verify', r'I\'m not able to', r'I don\'t have real-time', r'my knowledge cutoff' ] for pattern in hallucination_indicators: if re.search(pattern, text, re.IGNORECASE): return False return True def _check_currency(self, text: str) -> bool: """ Check information currency. P2: Flags potentially stale information. """ current_year = datetime.now().year # Check for very old year references (more than 5 years) old_years = [str(y) for y in range(2000, current_year - 5)] for year in old_years: # Only flag if year appears as a date reference if re.search(rf'\b{year}\b(?!\d)', text): # Check if it's in a historical context historical_keywords = ["founded", "established", "history", "since"] if not any(kw in text.lower() for kw in historical_keywords): return False return True def _check_adaptive_threshold(self, action: str, metadata: Dict[str, Any]) -> bool: """ Apply adaptive thresholds based on context. P9: Adjusts validation strictness. """ # Get trust score if available trust_score = metadata.get("trust_score", 0.5) # Adjust threshold based on trust effective_threshold = self.adaptive_threshold if trust_score > 0.8: effective_threshold *= 0.9 # More lenient for trusted workers elif trust_score < 0.3: effective_threshold *= 1.1 # Stricter for untrusted workers # Check action length as proxy for complexity action_length = len(action) if action_length > 10000: effective_threshold *= 1.05 # Slightly stricter for long actions return True # Threshold is used in final score calculation def _detect_dangerous_operations(self, text: str) -> Dict[str, List[str]]: """ Detect dangerous operation patterns. Returns dict of operation type to matched patterns. """ detected = {} for op_type, patterns in self.DANGEROUS_PATTERNS.items(): matches = [] for pattern in patterns: if re.search(pattern, text, re.IGNORECASE): matches.append(pattern) if matches: detected[op_type] = matches return detected def _check_permission(self, required: str, metadata: Dict[str, Any]) -> bool: """Check if current permission level allows the operation.""" required_level = self.PERMISSION_LEVELS.get(required, 2) current_level = metadata.get("permission_level", self.permission_level) return current_level >= required_level def _check_rate_limit(self, worker_id: str) -> bool: """Check if worker is within rate limits.""" if not worker_id: return True # Count recent validations for this worker recent_count = sum( 1 for r in self.validation_log[-100:] if worker_id in str(r) ) # Allow 50 validations per 100 entries return recent_count < 50 def _check_credential_exposure(self, text: str) -> bool: """Check for credential exposure patterns.""" for pattern in self.DANGEROUS_PATTERNS["credential_exposure"]: if re.search(pattern, text, re.IGNORECASE): return False return True def _check_injection(self, text: str) -> bool: """Check for injection patterns.""" injection_patterns = [ # SQL injection r"(?:(?:;|\')(?:--|#|/\*)|\b(?:SELECT|INSERT|UPDATE|DELETE|DROP|UNION)\b.*\b(?:FROM|INTO|WHERE)\b)", # Command injection r"(?:;|\||&&|\$\(|\`)[^;|&`]*(?:rm|cat|wget|curl|bash|sh|python|perl)", # Path traversal r"\.\.(?:/|\\){2,}", # LDAP injection r'\)\s*\(\|', # XPath injection r"'\s*or\s*'.*'.*'.*=" ] for pattern in injection_patterns: if re.search(pattern, text, re.IGNORECASE): return False return True def _log_security_event(self, event_type: str, details: Dict[str, Any]) -> None: """Log a security event.""" event = { "type": event_type, "timestamp": datetime.utcnow().isoformat(), "details": details } self.security_events.append(event) logger.warning(f"Security Event [{event_type}]: {details}") def _log_validation(self, result: SafetyResult, worker_id: str = None) -> None: """Log validation result.""" self.validation_log.append(result) log_level = logging.WARNING if not result.valid else logging.INFO logger.log( log_level, f"Gate3 Validation [worker={worker_id}]: valid={result.valid}, " f"score={result.score:.2f}, blocked={len(result.blocked_operations)}" ) def block_operation(self, operation_type: str, reason: str, worker_id: str = None) -> Dict[str, Any]: """ Explicitly block an operation. Args: operation_type: Type of operation being blocked reason: Reason for blocking worker_id: Worker attempting the operation Returns: Block result """ self.blocked_count += 1 self._log_security_event("operation_blocked", { "operation": operation_type, "reason": reason, "worker_id": worker_id }) return { "blocked": True, "operation": operation_type, "reason": reason, "timestamp": datetime.utcnow().isoformat() } def get_security_events(self, limit: int = 100, event_type: str = None) -> List[Dict[str, Any]]: """Get security events, optionally filtered by type.""" events = self.security_events[-limit:] if event_type: events = [e for e in events if e["type"] == event_type] return events def get_validation_history(self, limit: int = 100) -> List[Dict[str, Any]]: """Get recent validation results.""" return [r.to_dict() for r in self.validation_log[-limit:]] def get_metrics(self) -> Dict[str, Any]: """Get validation gate metrics.""" if not self.validation_log: return {"total_validations": 0} valid_count = sum(1 for r in self.validation_log if r.valid) total_count = len(self.validation_log) return { "total_validations": total_count, "valid_count": valid_count, "invalid_count": total_count - valid_count, "pass_rate": valid_count / total_count if total_count > 0 else 0, "avg_score": sum(r.score for r in self.validation_log) / total_count, "blocked_operations": self.blocked_count, "security_events": len(self.security_events), "adaptive_threshold": self.adaptive_threshold } def update_adaptive_threshold(self, adjustment: float) -> float: """ Update the adaptive threshold. P9: Allows threshold adjustment based on system state. Args: adjustment: Positive or negative adjustment (-0.1 to 0.1 recommended) Returns: New threshold value """ self.adaptive_threshold = max(0.5, min(1.0, self.adaptive_threshold + adjustment)) logger.info(f"Adaptive threshold updated to {self.adaptive_threshold}") return self.adaptive_threshold # Backward compatibility alias RealtimeGate = SafetyValidationGate if __name__ == "__main__": # Self-test gate = SafetyValidationGate() print("\n=== Validation Gate 3 (Safety) Test ===") # Test safe action safe_action = "Read the configuration file and display the settings." score, checks = gate.validate(safe_action, worker_id="worker_1") print(f"Safe action: score={score:.2f}, checks={checks}") # Test dangerous action dangerous_action = "rm -rf /tmp/* && exec(user_input)" score, checks = gate.validate(dangerous_action, worker_id="worker_2") print(f"Dangerous action: score={score:.2f}, checks={checks}") # Test credential exposure cred_action = "Set the api_key = 'sk_live_abc123xyz'" score, checks = gate.validate(cred_action, worker_id="worker_3") print(f"Credential exposure: score={score:.2f}, checks={checks}") # Test SQL injection sql_action = "SELECT * FROM users WHERE id = '1' OR '1'='1'; DROP TABLE users;--" score, checks = gate.validate(sql_action, worker_id="worker_4") print(f"SQL injection: score={score:.2f}, checks={checks}") # Block an operation explicitly block_result = gate.block_operation("delete_production", "Manual safety override", "admin") print(f"\nBlocked operation: {block_result}") # Security events print(f"\nSecurity Events: {gate.get_security_events(limit=5)}") # Metrics print(f"\nGate Metrics: {gate.get_metrics()}")