mirror of
https://github.com/anthropics/claude-code.git
synced 2026-02-19 04:27:33 -08:00
Compare commits
1 Commits
v2.1.42
...
claude/sla
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
965dbf27eb |
173
plugins/security-guidance/hooks/disk_space_utils.py
Normal file
173
plugins/security-guidance/hooks/disk_space_utils.py
Normal file
@@ -0,0 +1,173 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Disk space utilities for Claude Code hooks.
|
||||
|
||||
Provides helper functions to detect and handle disk space issues (ENOSPC errors)
|
||||
in a user-friendly manner.
|
||||
"""
|
||||
|
||||
import errno
|
||||
import os
|
||||
import sys
|
||||
from typing import Optional, Tuple
|
||||
|
||||
|
||||
# ENOSPC errno value (28 on Linux/Mac)
|
||||
ENOSPC_ERRNO = errno.ENOSPC
|
||||
|
||||
|
||||
def is_disk_space_error(exception: Exception) -> bool:
|
||||
"""Check if an exception is related to disk space issues.
|
||||
|
||||
Args:
|
||||
exception: The exception to check
|
||||
|
||||
Returns:
|
||||
True if the exception indicates a disk space issue
|
||||
"""
|
||||
# Check for OSError with ENOSPC errno
|
||||
if isinstance(exception, OSError):
|
||||
if hasattr(exception, 'errno') and exception.errno == ENOSPC_ERRNO:
|
||||
return True
|
||||
# Also check strerror for various disk space error messages
|
||||
if hasattr(exception, 'strerror') and exception.strerror:
|
||||
strerror_lower = exception.strerror.lower()
|
||||
disk_space_indicators = [
|
||||
'no space left on device',
|
||||
'disk quota exceeded',
|
||||
'not enough space',
|
||||
'insufficient disk space',
|
||||
]
|
||||
if any(indicator in strerror_lower for indicator in disk_space_indicators):
|
||||
return True
|
||||
|
||||
# Check error message string as fallback
|
||||
error_str = str(exception).lower()
|
||||
if 'enospc' in error_str or 'no space left' in error_str:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def get_disk_space_warning() -> str:
|
||||
"""Get a user-friendly warning message for disk space issues.
|
||||
|
||||
Returns:
|
||||
Warning message string
|
||||
"""
|
||||
return (
|
||||
"WARNING: Disk space issue detected. Your disk may be full or nearly full.\n"
|
||||
"This can cause Claude Code to become unresponsive or crash.\n"
|
||||
"\n"
|
||||
"Recommended actions:\n"
|
||||
" 1. Free up disk space by deleting unnecessary files\n"
|
||||
" 2. Check available space with: df -h\n"
|
||||
" 3. Clean up temporary files: sudo rm -rf /tmp/* (use with caution)\n"
|
||||
" 4. Empty trash/recycle bin\n"
|
||||
" 5. Consider removing old Docker images: docker system prune"
|
||||
)
|
||||
|
||||
|
||||
def check_available_disk_space(path: str = None, min_bytes: int = 10 * 1024 * 1024) -> Tuple[bool, Optional[str]]:
|
||||
"""Check if there's sufficient disk space available.
|
||||
|
||||
Args:
|
||||
path: Path to check (defaults to home directory)
|
||||
min_bytes: Minimum required bytes (default: 10MB)
|
||||
|
||||
Returns:
|
||||
Tuple of (has_space, warning_message)
|
||||
- has_space: True if sufficient space available
|
||||
- warning_message: Warning string if low on space, None otherwise
|
||||
"""
|
||||
if path is None:
|
||||
path = os.path.expanduser("~")
|
||||
|
||||
try:
|
||||
# Get disk usage statistics
|
||||
stat = os.statvfs(path)
|
||||
available_bytes = stat.f_frsize * stat.f_bavail
|
||||
|
||||
if available_bytes < min_bytes:
|
||||
available_mb = available_bytes / (1024 * 1024)
|
||||
required_mb = min_bytes / (1024 * 1024)
|
||||
return False, (
|
||||
f"Low disk space warning: Only {available_mb:.1f}MB available "
|
||||
f"(recommended minimum: {required_mb:.1f}MB)\n"
|
||||
f"{get_disk_space_warning()}"
|
||||
)
|
||||
|
||||
return True, None
|
||||
|
||||
except (OSError, AttributeError):
|
||||
# os.statvfs not available on all platforms (e.g., Windows)
|
||||
# Return True and let actual write operations fail if there's no space
|
||||
return True, None
|
||||
|
||||
|
||||
def safe_write_file(path: str, content: str, warn_on_disk_error: bool = True) -> Tuple[bool, Optional[str]]:
|
||||
"""Safely write content to a file with disk space error handling.
|
||||
|
||||
Args:
|
||||
path: Path to write to
|
||||
content: Content to write
|
||||
warn_on_disk_error: If True, print warning to stderr on disk space errors
|
||||
|
||||
Returns:
|
||||
Tuple of (success, error_message)
|
||||
- success: True if write succeeded
|
||||
- error_message: Error description if failed, None otherwise
|
||||
"""
|
||||
try:
|
||||
# Ensure directory exists
|
||||
dir_path = os.path.dirname(path)
|
||||
if dir_path:
|
||||
os.makedirs(dir_path, exist_ok=True)
|
||||
|
||||
with open(path, 'w') as f:
|
||||
f.write(content)
|
||||
|
||||
return True, None
|
||||
|
||||
except Exception as e:
|
||||
if is_disk_space_error(e):
|
||||
error_msg = f"Disk space error writing to {path}: {e}\n{get_disk_space_warning()}"
|
||||
if warn_on_disk_error:
|
||||
print(error_msg, file=sys.stderr)
|
||||
return False, error_msg
|
||||
else:
|
||||
return False, f"Error writing to {path}: {e}"
|
||||
|
||||
|
||||
def safe_append_file(path: str, content: str, warn_on_disk_error: bool = True) -> Tuple[bool, Optional[str]]:
|
||||
"""Safely append content to a file with disk space error handling.
|
||||
|
||||
Args:
|
||||
path: Path to append to
|
||||
content: Content to append
|
||||
warn_on_disk_error: If True, print warning to stderr on disk space errors
|
||||
|
||||
Returns:
|
||||
Tuple of (success, error_message)
|
||||
- success: True if append succeeded
|
||||
- error_message: Error description if failed, None otherwise
|
||||
"""
|
||||
try:
|
||||
# Ensure directory exists
|
||||
dir_path = os.path.dirname(path)
|
||||
if dir_path:
|
||||
os.makedirs(dir_path, exist_ok=True)
|
||||
|
||||
with open(path, 'a') as f:
|
||||
f.write(content)
|
||||
|
||||
return True, None
|
||||
|
||||
except Exception as e:
|
||||
if is_disk_space_error(e):
|
||||
error_msg = f"Disk space error appending to {path}: {e}\n{get_disk_space_warning()}"
|
||||
if warn_on_disk_error:
|
||||
print(error_msg, file=sys.stderr)
|
||||
return False, error_msg
|
||||
else:
|
||||
return False, f"Error appending to {path}: {e}"
|
||||
@@ -10,18 +10,40 @@ import random
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
# Import disk space utilities
|
||||
try:
|
||||
from disk_space_utils import (
|
||||
is_disk_space_error,
|
||||
get_disk_space_warning,
|
||||
check_available_disk_space,
|
||||
safe_write_file,
|
||||
safe_append_file,
|
||||
)
|
||||
DISK_UTILS_AVAILABLE = True
|
||||
except ImportError:
|
||||
# Fallback if disk_space_utils not available
|
||||
DISK_UTILS_AVAILABLE = False
|
||||
|
||||
# Debug log file
|
||||
DEBUG_LOG_FILE = "/tmp/security-warnings-log.txt"
|
||||
|
||||
# Track if we've already warned about disk space in this session
|
||||
_disk_space_warned = False
|
||||
|
||||
|
||||
def debug_log(message):
|
||||
"""Append debug message to log file with timestamp."""
|
||||
global _disk_space_warned
|
||||
try:
|
||||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
|
||||
with open(DEBUG_LOG_FILE, "a") as f:
|
||||
f.write(f"[{timestamp}] {message}\n")
|
||||
except Exception as e:
|
||||
# Silently ignore logging errors to avoid disrupting the hook
|
||||
# Check if this is a disk space error and warn the user
|
||||
if DISK_UTILS_AVAILABLE and is_disk_space_error(e) and not _disk_space_warned:
|
||||
_disk_space_warned = True
|
||||
print(f"[Security Hook] {get_disk_space_warning()}", file=sys.stderr)
|
||||
# Continue silently to avoid disrupting the hook
|
||||
pass
|
||||
|
||||
|
||||
@@ -158,26 +180,44 @@ def cleanup_old_state_files():
|
||||
|
||||
def load_state(session_id):
|
||||
"""Load the state of shown warnings from file."""
|
||||
global _disk_space_warned
|
||||
state_file = get_state_file(session_id)
|
||||
if os.path.exists(state_file):
|
||||
try:
|
||||
with open(state_file, "r") as f:
|
||||
return set(json.load(f))
|
||||
except (json.JSONDecodeError, IOError):
|
||||
except json.JSONDecodeError:
|
||||
debug_log(f"JSON decode error reading state file: {state_file}")
|
||||
return set()
|
||||
except Exception as e:
|
||||
# Check for disk-related errors (corrupted filesystem, etc.)
|
||||
if DISK_UTILS_AVAILABLE and is_disk_space_error(e):
|
||||
if not _disk_space_warned:
|
||||
_disk_space_warned = True
|
||||
print(f"[Security Hook] {get_disk_space_warning()}", file=sys.stderr)
|
||||
debug_log(f"Error loading state file: {e}")
|
||||
return set()
|
||||
return set()
|
||||
|
||||
|
||||
def save_state(session_id, shown_warnings):
|
||||
"""Save the state of shown warnings to file."""
|
||||
global _disk_space_warned
|
||||
state_file = get_state_file(session_id)
|
||||
try:
|
||||
os.makedirs(os.path.dirname(state_file), exist_ok=True)
|
||||
with open(state_file, "w") as f:
|
||||
json.dump(list(shown_warnings), f)
|
||||
except IOError as e:
|
||||
except Exception as e:
|
||||
# Check for disk space errors and provide user-friendly warning
|
||||
if DISK_UTILS_AVAILABLE and is_disk_space_error(e):
|
||||
if not _disk_space_warned:
|
||||
_disk_space_warned = True
|
||||
print(f"[Security Hook] {get_disk_space_warning()}", file=sys.stderr)
|
||||
debug_log(f"Disk space error saving state file: {e}")
|
||||
else:
|
||||
debug_log(f"Failed to save state file: {e}")
|
||||
pass # Fail silently if we can't save state
|
||||
# Fail silently to not disrupt operation
|
||||
|
||||
|
||||
def check_patterns(file_path, content):
|
||||
@@ -216,6 +256,8 @@ def extract_content_from_input(tool_name, tool_input):
|
||||
|
||||
def main():
|
||||
"""Main hook function."""
|
||||
global _disk_space_warned
|
||||
|
||||
# Check if security reminders are enabled
|
||||
security_reminder_enabled = os.environ.get("ENABLE_SECURITY_REMINDER", "1")
|
||||
|
||||
@@ -223,6 +265,13 @@ def main():
|
||||
if security_reminder_enabled == "0":
|
||||
sys.exit(0)
|
||||
|
||||
# Check for low disk space and warn user (only once per session)
|
||||
if DISK_UTILS_AVAILABLE and not _disk_space_warned:
|
||||
has_space, warning = check_available_disk_space()
|
||||
if not has_space:
|
||||
_disk_space_warned = True
|
||||
print(f"[Security Hook] {warning}", file=sys.stderr)
|
||||
|
||||
# Periodically clean up old state files (10% chance per run)
|
||||
if random.random() < 0.1:
|
||||
cleanup_old_state_files()
|
||||
|
||||
Reference in New Issue
Block a user