hermes-agent/validate_security.py

#!/usr/bin/env python3
"""Comprehensive security validation script.

Runs all security checks and reports status.
Usage: python validate_security.py
"""

import sys
import os
import subprocess
import ast
from pathlib import Path


class SecurityValidator:
    """Run comprehensive security validations."""
    
    def __init__(self):
        self.issues = []
        self.warnings = []
        self.checks_passed = 0
        self.checks_failed = 0
    
    def run_all(self):
        """Run all security checks."""
        print("=" * 80)
        print("🔒 SECURITY VALIDATION SUITE")
        print("=" * 80)
        
        self.check_command_injection()
        self.check_path_traversal()
        self.check_ssrf_protection()
        self.check_secret_leakage()
        self.check_interrupt_race_conditions()
        self.check_test_coverage()
        
        self.print_summary()
        return len(self.issues) == 0
    
    def check_command_injection(self):
        """Check for command injection vulnerabilities."""
        print("\n[1/6] Checking command injection protections...")
        
        # Check transcription_tools.py uses shlex.split
        content = Path("tools/transcription_tools.py").read_text()
        if "shlex.split" in content and "shell=False" in content:
            print("  ✅ transcription_tools.py: Uses safe list-based execution")
            self.checks_passed += 1
        else:
            print("  ❌ transcription_tools.py: May use unsafe shell execution")
            self.issues.append("Command injection in transcription_tools")
            self.checks_failed += 1
        
        # Check docker.py validates container IDs
        content = Path("tools/environments/docker.py").read_text()
        if "re.match" in content and "container" in content:
            print("  ✅ docker.py: Validates container ID format")
            self.checks_passed += 1
        else:
            print("  ⚠️  docker.py: Container ID validation not confirmed")
            self.warnings.append("Docker container ID validation")
    
    def check_path_traversal(self):
        """Check for path traversal protections."""
        print("\n[2/6] Checking path traversal protections...")
        
        content = Path("tools/file_operations.py").read_text()
        
        checks = [
            ("_validate_safe_path", "Path validation function"),
            ("_contains_path_traversal", "Traversal detection function"),
            ("../", "Unix traversal pattern"),
            ("..\\\\", "Windows traversal pattern"),
            ("\\\\x00", "Null byte detection"),
        ]
        
        for pattern, description in checks:
            if pattern in content:
                print(f"  ✅ {description}")
                self.checks_passed += 1
            else:
                print(f"  ❌ Missing: {description}")
                self.issues.append(f"Path traversal: {description}")
                self.checks_failed += 1
    
    def check_ssrf_protection(self):
        """Check for SSRF protections."""
        print("\n[3/6] Checking SSRF protections...")
        
        content = Path("tools/url_safety.py").read_text()
        
        checks = [
            ("_is_blocked_ip", "IP blocking function"),
            ("create_safe_socket", "Connection-level validation"),
            ("169.254", "Metadata service block"),
            ("is_private", "Private IP detection"),
        ]
        
        for pattern, description in checks:
            if pattern in content:
                print(f"  ✅ {description}")
                self.checks_passed += 1
            else:
                print(f"  ⚠️  {description} not found")
                self.warnings.append(f"SSRF: {description}")
    
    def check_secret_leakage(self):
        """Check for secret leakage protections."""
        print("\n[4/6] Checking secret leakage protections...")
        
        content = Path("tools/code_execution_tool.py").read_text()
        
        if "_ALLOWED_ENV_VARS" in content:
            print("  ✅ Uses whitelist for environment variables")
            self.checks_passed += 1
        elif "_SECRET_SUBSTRINGS" in content:
            print("  ⚠️  Uses blacklist (may be outdated version)")
            self.warnings.append("Blacklist instead of whitelist for secrets")
        else:
            print("  ❌ No secret filtering found")
            self.issues.append("Secret leakage protection")
            self.checks_failed += 1
        
        # Check for common secret patterns in allowed list
        dangerous_vars = ["API_KEY", "SECRET", "PASSWORD", "TOKEN"]
        found_dangerous = [v for v in dangerous_vars if v in content]
        
        if found_dangerous:
            print(f"  ⚠️  Found potential secret vars in code: {found_dangerous}")
    
    def check_interrupt_race_conditions(self):
        """Check for interrupt race condition fixes."""
        print("\n[5/6] Checking interrupt race condition protections...")
        
        content = Path("tools/interrupt.py").read_text()
        
        checks = [
            ("RLock", "Reentrant lock for thread safety"),
            ("_interrupt_lock", "Lock variable"),
            ("_interrupt_count", "Nesting count tracking"),
        ]
        
        for pattern, description in checks:
            if pattern in content:
                print(f"  ✅ {description}")
                self.checks_passed += 1
            else:
                print(f"  ❌ Missing: {description}")
                self.issues.append(f"Interrupt: {description}")
                self.checks_failed += 1
    
    def check_test_coverage(self):
        """Check security test coverage."""
        print("\n[6/6] Checking security test coverage...")
        
        test_files = [
            "tests/tools/test_interrupt.py",
            "tests/tools/test_path_traversal.py",
            "tests/tools/test_command_injection.py",
        ]
        
        for test_file in test_files:
            if Path(test_file).exists():
                print(f"  ✅ {test_file}")
                self.checks_passed += 1
            else:
                print(f"  ❌ Missing: {test_file}")
                self.issues.append(f"Missing test: {test_file}")
                self.checks_failed += 1
    
    def print_summary(self):
        """Print validation summary."""
        print("\n" + "=" * 80)
        print("VALIDATION SUMMARY")
        print("=" * 80)
        print(f"Checks Passed: {self.checks_passed}")
        print(f"Checks Failed: {self.checks_failed}")
        print(f"Warnings: {len(self.warnings)}")
        
        if self.issues:
            print("\n❌ CRITICAL ISSUES:")
            for issue in self.issues:
                print(f"  - {issue}")
        
        if self.warnings:
            print("\n⚠️  WARNINGS:")
            for warning in self.warnings:
                print(f"  - {warning}")
        
        if not self.issues:
            print("\n✅ ALL SECURITY CHECKS PASSED")
        
        print("=" * 80)


if __name__ == "__main__":
    validator = SecurityValidator()
    success = validator.run_all()
    sys.exit(0 if success else 1)