feat: extract sovereign work from hermes-agent fork into sidecar
Extracted 52 files from Timmy_Foundation/hermes-agent (gitea/main) into hermes-sovereign/ directory to restore clean upstream tracking. Layout: docs/ 19 files — deploy guides, performance reports, security docs, research security/ 5 files — audit workflows, PR checklists, validation scripts wizard-bootstrap/ 7 files — wizard environment, dependency checking, auditing notebooks/ 2 files — Jupyter health monitoring notebooks scripts/ 5 files — forge health, smoke tests, syntax guard, deploy validation ci/ 2 files — Gitea CI workflow definitions githooks/ 3 files — pre-commit hooks and config devkit/ 8 files — developer toolkit (Gitea client, health, notebook runner) README.md 1 file — directory overview Addresses: #337, #338
This commit is contained in:
Alexander Whitestone
199
hermes-sovereign/security/validate_security.py
Normal file
199
hermes-sovereign/security/validate_security.py
Normal file
@@ -0,0 +1,199 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user