Skip to content
Security
Hook

Hooks

What hs runs automatically, and when. A hook is a command Claude Code fires at a fixed moment, without you asking for it.

@frmorettofrmoretto@frmoretto
From plugin
hs
321 skill6 commands1 hook
Install
> /plugin marketplace add frmoretto/hardstop
> /plugin install hs@hardstop

Ships with hs. Installing the plugin gets these hooks.

What fires, and when

PreToolUse

  • MatchesBashpython ${CLAUDE_PLUGIN_ROOT}/hooks/pre_tool_use.py
  • MatchesPowerShellpython ${CLAUDE_PLUGIN_ROOT}/hooks/pre_tool_use.py
  • MatchesReadpython ${CLAUDE_PLUGIN_ROOT}/hooks/pre_read.py
Read hooks/hooks.json

Where it lives

  • hooks/pattern_loader.pyGitHub
    Read the script
    #!/usr/bin/env python3
"""
Pattern Loader for Hardstop v1.4.0
Loads security patterns from YAML files with caching support.
"""

import yaml
from pathlib import Path
from typing import Dict, List, Optional


class PatternLoader:
    """Loads and manages security patterns from YAML files."""

    def __init__(self, patterns_dir: Optional[Path] = None):
        """
        Initialize pattern loader.

        Args:
            patterns_dir: Directory containing pattern YAML files.
                         Defaults to ../patterns relative to this file.
        """
        if patterns_dir is None:
            patterns_dir = Path(__file__).parent.parent / "patterns"

        self.patterns_dir = Path(patterns_dir)
        self._cache: Dict[str, List[dict]] = {}

    def load_patterns(self, filename: str) -> List[dict]:
        """
        Load patterns from a YAML file.

        Args:
            filename: Name of the YAML file (e.g., 'dangerous_commands.yaml')

        Returns:
            List of pattern dictionaries, each containing:
                - id: Unique pattern identifier
                - regex: Regular expression pattern
                - message: Human-readable description
                - severity: Severity level (critical, high, medium, low, info)
                - category: Pattern category (credential, network, filesystem, etc.)
        """
        # Check cache first
        if filename in self._cache:
            return self._cache[filename]

        file_path = self.patterns_dir / filename

        # Handle missing files gracefully
        if not file_path.exists():
            return []

        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = yaml.safe_load(f)

            patterns = data.get('patterns', []) if data else []

            # Cache the results
            self._cache[filename] = patterns

            return patterns

        except Exception as e:
            # Log error but don't crash - return empty list
            print(f"Warning: Failed to load {filename}: {e}")
            return []

    def load_dangerous_commands(self) -> List[dict]:
        """Load dangerous command patterns."""
        return self.load_patterns('dangerous_commands.yaml')

    def load_dangerous_reads(self) -> List[dict]:
        """Load dangerous read patterns."""
        return self.load_patterns('dangerous_reads.yaml')

    def load_sensitive_reads(self) -> List[dict]:
        """Load sensitive read patterns."""
        return self.load_patterns('sensitive_reads.yaml')

    def load_safe_commands(self) -> List[dict]:
        """Load safe command patterns (currently empty by design)."""
        return self.load_patterns('safe_commands.yaml')

    def load_safe_reads(self) -> List[dict]:
        """Load safe read patterns (currently empty by design)."""
        return self.load_patterns('safe_reads.yaml')

    def get_all_patterns(self) -> Dict[str, List[dict]]:
        """
        Load all pattern files.

        Returns:
            Dictionary mapping category names to pattern lists:
                - dangerous_commands: 180 patterns
                - dangerous_reads: 71 patterns
                - sensitive_reads: 11 patterns
                - safe_commands: 0 patterns (empty by design)
                - safe_reads: 0 patterns (empty by design)
        """
        return {
            'dangerous_commands': self.load_dangerous_commands(),
            'dangerous_reads': self.load_dangerous_reads(),
            'sensitive_reads': self.load_sensitive_reads(),
            'safe_commands': self.load_safe_commands(),
            'safe_reads': self.load_safe_reads(),
        }

    def get_pattern_count(self) -> Dict[str, int]:
        """
        Get count of patterns in each category.

        Returns:
            Dictionary mapping category names to pattern counts.
        """
        all_patterns = self.get_all_patterns()
        return {category: len(patterns) for category, patterns in all_patterns.items()}

    def get_total_count(self) -> int:
        """
        Get total count of all loaded patterns.

        Returns:
            Total number of patterns across all categories.
        """
        counts = self.get_pattern_count()
        return sum(counts.values())

    def clear_cache(self):
        """Clear the pattern cache. Useful for testing or live reloading."""
        self._cache.clear()


# Singleton instance for convenience
_default_loader: Optional[PatternLoader] = None


def get_loader() -> PatternLoader:
    """Get the default pattern loader instance."""
    global _default_loader
    if _default_loader is None:
        _default_loader = PatternLoader()
    return _default_loader


# Convenience functions for direct access
def load_dangerous_commands() -> List[dict]:
    """Load dangerous command patterns."""
    return get_loader().load_dangerous_commands()


def load_dangerous_reads() -> List[dict]:
    """Load dangerous read patterns."""
    return get_loader().load_dangerous_reads()


def load_sensitive_reads() -> List[dict]:
    """Load sensitive read patterns."""
    return get_loader().load_sensitive_reads()


def load_safe_commands() -> List[dict]:
    """Load safe command patterns."""
    return get_loader().load_safe_commands()


def load_safe_reads() -> List[dict]:
    """Load safe read patterns."""
    return get_loader().load_safe_reads()


def get_pattern_count() -> Dict[str, int]:
    """Get count of patterns in each category."""
    return get_loader().get_pattern_count()


def get_total_count() -> int:
    """Get total count of all loaded patterns."""
    return get_loader().get_total_count()
  • hooks/pre_read.pyRunsGitHub
    Read the script
    #!/usr/bin/env python3
"""
Hardstop Plugin — PreToolUse Hook (Read)

Blocks reading of sensitive credential files to prevent secrets exposure.

Exit codes:
  0 = Success (uses JSON output for allow/deny decision)

Blocking uses permissionDecision: "deny" in JSON output instead of exit code 2.
This ensures consistent behavior between CLI and VS Code extension.

Design principle: Fail-closed. If safety check fails, block the read.
"""

import sys
import json
import re
import os
from pathlib import Path
from datetime import datetime
from typing import Tuple, Optional, List, Dict

# Import pattern loader for YAML-based patterns
try:
    from pattern_loader import load_dangerous_reads, load_sensitive_reads
    PATTERN_LOADER_AVAILABLE = True
except ImportError:
    PATTERN_LOADER_AVAILABLE = False

# Import session tracker for risk scoring (v1.4.0+)
try:
    from session_tracker import get_tracker
    SESSION_TRACKER_AVAILABLE = True
except ImportError:
    SESSION_TRACKER_AVAILABLE = False

# === CONFIGURATION ===

STATE_DIR = Path.home() / ".hardstop"
SKIP_FILE = STATE_DIR / "skip_next"
LOG_FILE = STATE_DIR / "audit.log"
DEBUG_FILE = STATE_DIR / "hook_debug.log"

# Fail-closed: if True, errors during safety check block the read
FAIL_CLOSED = True

# === DEBUG LOGGING ===

try:
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    with open(DEBUG_FILE, "a") as f:
        f.write(f"[{datetime.now().isoformat()}] Read hook invoked\n")
except:
    pass

# === DANGEROUS READ PATTERNS ===
# These paths contain secrets that should never be read by AI

# Pattern registries: store full pattern metadata for risk scoring
_READ_PATTERN_REGISTRY: Dict[str, Dict] = {}
_SENSITIVE_READ_REGISTRY: Dict[str, Dict] = {}

# Load patterns from YAML files (v1.4.0+) or use fallback
def _load_dangerous_read_patterns() -> List[Tuple[str, str]]:
    """
    Load dangerous read patterns from YAML files.
    Returns list of (regex, pattern_id) tuples.
    Builds _READ_PATTERN_REGISTRY with full metadata.
    Falls back to empty list if pattern loader unavailable.
    """
    global _READ_PATTERN_REGISTRY

    if not PATTERN_LOADER_AVAILABLE:
        print("Warning: pattern_loader not available, using empty dangerous read patterns list", file=sys.stderr)
        return []

    try:
        yaml_patterns = load_dangerous_reads()

        # Build registry with full pattern data
        _READ_PATTERN_REGISTRY.clear()
        for pattern in yaml_patterns:
            pattern_id = pattern.get('id', 'UNKNOWN')
            _READ_PATTERN_REGISTRY[pattern_id] = pattern

        # Return matching list: (regex, pattern_id)
        return [(p['regex'], p.get('id', 'UNKNOWN')) for p in yaml_patterns
                if 'regex' in p]

    except Exception as e:
        print(f"Warning: Failed to load dangerous read patterns: {e}", file=sys.stderr)
        return []


DANGEROUS_READ_PATTERNS = _load_dangerous_read_patterns()


def _load_sensitive_read_patterns() -> List[Tuple[str, str]]:
    """
    Load sensitive read patterns from YAML files.
    Returns list of (regex, pattern_id) tuples.
    Builds _SENSITIVE_READ_REGISTRY with full metadata.
    Falls back to empty list if pattern loader unavailable.
    """
    global _SENSITIVE_READ_REGISTRY

    if not PATTERN_LOADER_AVAILABLE:
        print("Warning: pattern_loader not available, using empty sensitive read patterns list", file=sys.stderr)
        return []

    try:
        yaml_patterns = load_sensitive_reads()

        # Build registry with full pattern data
        _SENSITIVE_READ_REGISTRY.clear()
        for pattern in yaml_patterns:
            pattern_id = pattern.get('id', 'UNKNOWN')
            _SENSITIVE_READ_REGISTRY[pattern_id] = pattern

        # Return matching list: (regex, pattern_id)
        return [(p['regex'], p.get('id', 'UNKNOWN')) for p in yaml_patterns
                if 'regex' in p]

    except Exception as e:
        print(f"Warning: Failed to load sensitive read patterns: {e}", file=sys.stderr)
        return []


SENSITIVE_READ_PATTERNS = _load_sensitive_read_patterns()

# Note: Legacy hardcoded patterns removed in v1.4.0 (now in patterns/sensitive_reads.yaml)

# === SAFE READ PATTERNS ===
# Explicit allowlist for common safe reads

SAFE_READ_PATTERNS = [
    # Documentation
    r"README\.md$",
    r"README\.rst$",
    r"README\.txt$",
    r"README$",
    r"CHANGELOG\.md$",
    r"CHANGELOG$",
    r"HISTORY\.md$",
    r"LICENSE$",
    r"LICENSE\.md$",
    r"LICENSE\.txt$",
    r"CONTRIBUTING\.md$",
    r"CODE_OF_CONDUCT\.md$",
    r"\.md$",
    r"\.rst$",
    r"\.txt$",

    # Source code
    r"\.py$",
    r"\.pyi$",
    r"\.js$",
    r"\.mjs$",
    r"\.cjs$",
    r"\.ts$",
    r"\.tsx$",
    r"\.jsx$",
    r"\.go$",
    r"\.rs$",
    r"\.java$",
    r"\.kt$",
    r"\.scala$",
    r"\.c$",
    r"\.cpp$",
    r"\.cc$",
    r"\.h$",
    r"\.hpp$",
    r"\.cs$",
    r"\.rb$",
    r"\.php$",
    r"\.swift$",
    r"\.m$",
    r"\.mm$",
    r"\.lua$",
    r"\.pl$",
    r"\.sh$",
    r"\.bash$",
    r"\.zsh$",
    r"\.fish$",
    r"\.ps1$",
    r"\.bat$",
    r"\.cmd$",
    r"\.sql$",
    r"\.graphql$",
    r"\.gql$",

    # Config (Non-Sensitive)
    r"package\.json$",
    r"package-lock\.json$",
    r"yarn\.lock$",
    r"pnpm-lock\.yaml$",
    r"tsconfig\.json$",
    r"jsconfig\.json$",
    r"pyproject\.toml$",
    r"setup\.py$",
    r"setup\.cfg$",
    r"Cargo\.toml$",
    r"Cargo\.lock$",
    r"go\.mod$",
    r"go\.sum$",
    r"requirements\.txt$",
    r"Pipfile$",
    r"Pipfile\.lock$",
    r"Gemfile$",
    r"Gemfile\.lock$",
    r"composer\.json$",
    r"composer\.lock$",
    r"Makefile$",
    r"CMakeLists\.txt$",
    r"\.gitignore$",
    r"\.dockerignore$",
    r"Dockerfile$",
    r"docker-compose\.yml$",
    r"docker-compose\.yaml$",

    # Example/Template Files (safe versions of .env)
    r"\.env\.example$",
    r"\.env\.template$",
    r"\.env\.sample$",
    r"\.env\.dist$",
    r"example\.",
    r"sample\.",
    r"template\.",

    # Web assets
  • hooks/pre_tool_use.pyRunsGitHub
    Read the script
    #!/usr/bin/env python3
"""
Hardstop Plugin — PreToolUse Hook (Bash)

Two-layer protection:
  Layer 1: Pattern matching (instant)
  Layer 2: Claude CLI analysis (within subscription)

Exit codes:
  0 = Success (uses JSON output for allow/deny decision)

Blocking uses permissionDecision: "deny" in JSON output instead of exit code 2.
This ensures consistent behavior between CLI and VS Code extension.

Design principle: Fail-closed. If safety check fails, block the command.
"""

import sys
import json
import re
import subprocess
import os
import shlex
import tempfile
from pathlib import Path
from datetime import datetime
from typing import Tuple, Optional, List, Dict

# Import pattern loader for YAML-based patterns
try:
    from pattern_loader import load_dangerous_commands
    PATTERN_LOADER_AVAILABLE = True
except ImportError:
    PATTERN_LOADER_AVAILABLE = False

# Import session tracker for risk scoring (v1.4.0+)
try:
    from session_tracker import get_tracker
    SESSION_TRACKER_AVAILABLE = True
except ImportError:
    SESSION_TRACKER_AVAILABLE = False

# DEBUG: Write to file to confirm hook is being invoked
DEBUG_FILE = Path.home() / ".hardstop" / "hook_debug.log"
try:
    DEBUG_FILE.parent.mkdir(parents=True, exist_ok=True)
    with open(DEBUG_FILE, "a") as f:
        f.write(f"[{datetime.now().isoformat()}] Hook invoked\n")
except:
    pass

# === CONFIGURATION ===

STATE_DIR = Path.home() / ".hardstop"
STATE_FILE = STATE_DIR / "state.json"
SKIP_FILE = STATE_DIR / "skip_next"
LOG_FILE = STATE_DIR / "audit.log"

# Fail-closed: if True, errors during safety check block the command
FAIL_CLOSED = True

# === PATTERNS ===

# Pattern registry: stores full pattern metadata for risk scoring
_PATTERN_REGISTRY: Dict[str, Dict] = {}

# Load patterns from YAML files (v1.4.0+) or use fallback
def _load_dangerous_patterns() -> List[Tuple[str, str]]:
    """
    Load dangerous command patterns from YAML files.
    Returns list of (regex, pattern_id) tuples.
    Builds _PATTERN_REGISTRY with full metadata.
    Falls back to empty list if pattern loader unavailable.
    """
    global _PATTERN_REGISTRY

    if not PATTERN_LOADER_AVAILABLE:
        print("Warning: pattern_loader not available, using empty dangerous patterns list", file=sys.stderr)
        return []

    try:
        yaml_patterns = load_dangerous_commands()

        # Build registry with full pattern data
        _PATTERN_REGISTRY.clear()
        for pattern in yaml_patterns:
            pattern_id = pattern.get('id', 'UNKNOWN')
            _PATTERN_REGISTRY[pattern_id] = pattern

        # Return matching list: (regex, pattern_id)
        # Only include patterns that have both regex and id
        return [(p['regex'], p.get('id', 'UNKNOWN')) for p in yaml_patterns
                if 'regex' in p]

    except Exception as e:
        print(f"Warning: Failed to load dangerous patterns: {e}", file=sys.stderr)
        return []


# Load patterns at module import time
DANGEROUS_PATTERNS = _load_dangerous_patterns()

# Note: Legacy hardcoded patterns removed in v1.4.0 (now in patterns/dangerous_commands.yaml)

# Derive config dir from installed location: hooks/ -> hs/ -> plugins/ -> config dir
_CLAUDE_DIR = str(Path(__file__).absolute().parent.parent.parent.parent)
_CLAUDE_DIR_RE = re.escape(_CLAUDE_DIR)

SAFE_PATTERNS = [
    # Hardstop's own operations (must be able to manage itself)
    rf"^python\s+.*{_CLAUDE_DIR_RE}[/\\]plugins[/\\]hs[/\\].*\.py(?:\s+.*)?$",
    r"^python\s+.*\.hardstop.*$",
    r"^cat\s+.*\.hardstop[/\\].*$",
    rf"^cat\s+.*{_CLAUDE_DIR_RE}[/\\]plugins[/\\]hs[/\\].*$",
    r"^rm\s+(-f\s+)?.*\.hardstop[/\\](skip_next|hook_debug\.log)$",
    rf"^grep\s+.*{_CLAUDE_DIR_RE}[/\\]plugins[/\\]hs[/\\].*$",

    # Read-only operations
    r"^ls(?:\s+.*)?$",
    # cd with path - blocks command substitution $() and backticks
    # Allows: cd, cd /path, cd "path", cd 'path', cd ~/dir, cd ..
    # Blocks: cd $(cmd), cd `cmd`, cd ${var}$(cmd)
    r"^cd(?:\s+(?:\"[^`$()]*\"|'[^']*'|[^\s`$()]+))?$",
    # cat: allow reading files, but NOT credential paths (those are caught by DANGEROUS first)
    r"^cat\s+(?!.*(\.ssh/id_|\.aws/credentials|\.kube/config|\.docker/config\.json|\.npmrc|\.netrc|\.gnupg/|\.git-credentials|/etc/shadow|\.env$|\.env\s)).+$",
    r"^head\s+.+$",
    r"^tail\s+.+$",
    r"^less\s+.+$",
    r"^more\s+.+$",
    r"^pwd\s*$",
    r"^which\s+.+$",
    r"^type\s+.+$",
    r"^file\s+.+$",
    r"^wc\s+.+$",
    r"^grep\s+.+$",
    r"^find\s+.*\s-name\s+.*$",  # find with -name (read-only)
    r"^echo(?:\s+.*)?$",
    r"^date\s*$",
    r"^whoami\s*$",
    r"^hostname\s*$",
    r"^uname(?:\s+.*)?$",
    r"^env\s*$",
    r"^printenv(?:\s+.*)?$",
    
    # Git read operations
    r"^git\s+(status|log|diff|show|remote|describe|shortlog|whatchanged|rev-parse|rev-list|cat-file|ls-tree)(?:\s+.*)?$",
    r"^git\s+ls-[^\s]+(?:\s+.*)?$",

    # Git standard workflow (recoverable via reflog)
    # Excludes: reset (--hard loses uncommitted work), clean (deletes untracked), rebase --exec (runs shell)
    r"^git\s+(add|commit|push|pull|fetch|clone|stash|checkout|switch|restore|merge|cherry-pick|branch|tag|init|config|am|apply|bisect|blame|bundle|format-patch|gc|mv|notes|reflog|revert|rm|submodule|worktree)(?:[\s\S]+)?$",
    r"^git\s+rebase(?!\s+.*--exec)(?:[\s\S]+)?$",  # rebase allowed, but not with --exec
    
    # Regeneratable cleanup
    r"^rm\s+(-[^\s]*\s+)*node_modules/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*__pycache__/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*\.venv/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*venv/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*\.pytest_cache/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*dist/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*build/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*\.next/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*\.nuxt/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*coverage/?\s*$",
    r"^rm\s+(-[^\s]*\s+)*(/tmp/|\$TMPDIR)\s*$",
    
    # Package managers (read/lock operations)
    r"^npm\s+(list|ls|outdated|audit|view)(?:\s+.*)?$",
    r"^pip\s+(list|show|freeze)
  • hooks/risk_scoring.pyGitHub
    Read the script
    #!/usr/bin/env python3
"""
Risk scoring system for command execution safety.

Tracks cumulative risk score per session based on blocked commands.
Each blocked command contributes to a session risk score based on its severity.

Part of Hardstop v1.4.0 - Phase 2.3: Risk Scoring System
"""

from typing import Dict, Tuple

# Severity weights: points added to risk score when command is blocked
SEVERITY_WEIGHTS = {
    "critical": 25,  # Fork bomb, rm -rf /, credential theft
    "high": 15,      # Reverse shell, sudo abuse, network exfiltration
    "medium": 10,    # Config changes, overly permissive permissions
    "low": 5,        # Suspicious but likely benign
    "info": 1,       # Logged but not concerning
}

# Risk thresholds: ranges that define overall session risk level
RISK_THRESHOLDS = {
    "low": (0, 24),           # 0-24: minimal risk
    "moderate": (25, 49),     # 25-49: some concerning patterns
    "high": (50, 74),         # 50-74: multiple dangerous attempts
    "critical": (75, float('inf')),  # 75+: sustained attack pattern
}


def calculate_risk_level(score: int) -> str:
    """
    Calculate risk level from numeric score.

    Args:
        score: Cumulative risk score (sum of severity weights)

    Returns:
        Risk level: "low", "moderate", "high", or "critical"

    Examples:
        >>> calculate_risk_level(10)
        'low'
        >>> calculate_risk_level(30)
        'moderate'
        >>> calculate_risk_level(60)
        'high'
        >>> calculate_risk_level(100)
        'critical'
    """
    for level, (min_score, max_score) in RISK_THRESHOLDS.items():
        if min_score <= score <= max_score:
            return level
    return "unknown"


def get_severity_weight(severity: str) -> int:
    """
    Get numeric weight for a severity level.

    Args:
        severity: Severity level string

    Returns:
        Weight (points to add to risk score)

    Examples:
        >>> get_severity_weight("critical")
        25
        >>> get_severity_weight("high")
        15
        >>> get_severity_weight("invalid")
        0
    """
    return SEVERITY_WEIGHTS.get(severity.lower(), 0)


def get_risk_color(level: str) -> str:
    """
    Get ANSI color code for risk level.

    Args:
        level: Risk level string

    Returns:
        ANSI color code

    Examples:
        >>> get_risk_color("critical")
        '\\033[91m'
        >>> get_risk_color("low")
        '\\033[92m'
    """
    colors = {
        "low": "\033[92m",      # Green
        "moderate": "\033[93m",  # Yellow
        "high": "\033[91m",      # Red
        "critical": "\033[95m",  # Magenta
    }
    return colors.get(level.lower(), "\033[0m")


def format_risk_display(score: int, level: str) -> str:
    """
    Format risk score and level for display.

    Args:
        score: Numeric risk score
        level: Risk level string

    Returns:
        Formatted string with color

    Examples:
        >>> format_risk_display(35, "moderate")
        '\\033[93mMODERATE\\033[0m (35 points)'
    """
    color = get_risk_color(level)
    reset = "\033[0m"
    return f"{color}{level.upper()}{reset} ({score} points)"


def get_risk_description(level: str) -> str:
    """
    Get human-readable description of risk level.

    Args:
        level: Risk level string

    Returns:
        Description of what this risk level means
    """
    descriptions = {
        "low": "Minimal risk detected. Normal usage patterns.",
        "moderate": "Some concerning patterns detected. Review blocked commands.",
        "high": "Multiple dangerous attempts detected. Potential security threat.",
        "critical": "Sustained attack pattern detected. Immediate review recommended.",
    }
    return descriptions.get(level.lower(), "Unknown risk level.")


# Export all public functions
__all__ = [
    'SEVERITY_WEIGHTS',
    'RISK_THRESHOLDS',
    'calculate_risk_level',
    'get_severity_weight',
    'get_risk_color',
    'format_risk_display',
    'get_risk_description',
]
  • hooks/session_tracker.pyGitHub
    Read the script
    #!/usr/bin/env python3
"""
Session-based risk tracking for Hardstop.

Tracks blocked commands and cumulative risk score per session.
Persists data to ~/.hardstop/session.json

Part of Hardstop v1.4.0 - Phase 2.3: Risk Scoring System
"""

import json
import os
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Optional
from risk_scoring import SEVERITY_WEIGHTS, calculate_risk_level, get_severity_weight

HARDSTOP_DIR = Path.home() / ".hardstop"
SESSION_FILE = HARDSTOP_DIR / "session.json"


class SessionTracker:
    """Track session risk and blocked commands."""

    def __init__(self):
        """Initialize session tracker."""
        self.session_id = self._get_session_id()
        self.data = self._load_session()

    def _get_session_id(self) -> str:
        """
        Get or create session ID.

        Uses HARDSTOP_SESSION_ID environment variable if set,
        otherwise creates a new session ID from current timestamp.

        Returns:
            Session ID string
        """
        session_id = os.environ.get('HARDSTOP_SESSION_ID')
        if not session_id:
            session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
            os.environ['HARDSTOP_SESSION_ID'] = session_id
        return session_id

    def _load_session(self) -> Dict:
        """
        Load session data from disk.

        If session file exists and session ID matches, loads existing data.
        Otherwise creates a new session.

        Returns:
            Session data dictionary
        """
        if SESSION_FILE.exists():
            try:
                with open(SESSION_FILE) as f:
                    data = json.load(f)
                    # Reset if session ID changed
                    if data.get('session_id') != self.session_id:
                        return self._create_new_session()
                    return data
            except (json.JSONDecodeError, IOError) as e:
                # If file is corrupted, create new session
                print(f"Warning: Failed to load session data: {e}", flush=True)
                return self._create_new_session()
        return self._create_new_session()

    def _create_new_session(self) -> Dict:
        """
        Create new session data structure.

        Returns:
            New session dictionary with initial values
        """
        return {
            'session_id': self.session_id,
            'started_at': datetime.now().isoformat(),
            'risk_score': 0,
            'blocked_commands': [],
        }

    def _save_session(self):
        """
        Persist session data to disk.

        Creates ~/.hardstop directory if it doesn't exist.
        """
        try:
            HARDSTOP_DIR.mkdir(parents=True, exist_ok=True)
            with open(SESSION_FILE, 'w') as f:
                json.dump(self.data, f, indent=2)
        except IOError as e:
            print(f"Warning: Failed to save session data: {e}", flush=True)

    def record_block(self, command: str, pattern_data: Dict):
        """
        Record a blocked command and update risk score.

        Args:
            command: The command that was blocked
            pattern_data: Pattern information including severity, message, etc.
        """
        severity = pattern_data.get('severity', 'medium')
        weight = get_severity_weight(severity)

        block_record = {
            'timestamp': datetime.now().isoformat(),
            'command': command[:200],  # Truncate long commands
            'severity': severity,
            'weight': weight,
            'message': pattern_data.get('message', ''),
            'pattern_id': pattern_data.get('id', ''),
            'mitre_attack': pattern_data.get('mitre_attack'),
            'category': pattern_data.get('category'),
        }

        self.data['blocked_commands'].append(block_record)
        self.data['risk_score'] += weight
        self.data['last_blocked_at'] = datetime.now().isoformat()
        self._save_session()

    def get_risk_score(self) -> int:
        """
        Get current session risk score.

        Returns:
            Current risk score (cumulative weight of all blocked commands)
        """
        return self.data.get('risk_score', 0)

    def get_risk_level(self) -> str:
        """
        Get current risk level based on score.

        Returns:
            Risk level: "low", "moderate", "high", or "critical"
        """
        return calculate_risk_level(self.get_risk_score())

    def get_blocked_commands(self) -> List[Dict]:
        """
        Get list of blocked commands this session.

        Returns:
            List of blocked command records
        """
        return self.data.get('blocked_commands', [])

    def get_blocked_count(self) -> int:
        """
        Get count of blocked commands.

        Returns:
            Number of commands blocked this session
        """
        return len(self.data.get('blocked_commands', []))

    def get_session_info(self) -> Dict:
        """
        Get complete session information.

        Returns:
            Dictionary with session_id, started_at, risk_score, risk_level, blocked_count
        """
        return {
            'session_id': self.session_id,
            'started_at': self.data.get('started_at'),
            'last_blocked_at': self.data.get('last_blocked_at'),
            'risk_score': self.get_risk_score(),
            'risk_level': self.get_risk_level(),
            'blocked_count': self.get_blocked_count(),
        }

    def get_severity_breakdown(self) -> Dict[str, int]:
        """
        Get breakdown of blocked commands by severity.

        Returns:
            Dictionary mapping severity levels to counts
        """
        breakdown = {'critical': 0, 'high': 0, 'medium': 0, 'low': 0, 'info': 0}
        for cmd in self.get_blocked_commands():
            severity = cmd.get('severity', 'medium')
            if severity in breakdown:
                breakdown[severity] += 1

Read the script before you install anything that runs on your machine. This is the one part of a plugin that acts without being asked.

Ships withhs

👉 ⭐ Star on GitHub if Hardstop keeps you safe! Pre-execution safety validation for AI coding agents.

Get the whole plugin
Stats
PythonJavaScriptNodeBashPowerShell
+7 tags
DockerKubernetesAWSLinuxGitGitHubAnthropic
32
Stars
2
Forks
Maintained
Maintenance
Python
Language
5mo ago
Last commit
8mo ago
Created

Repo: frmoretto/hardstop

In this plugin
6 commands1 skill
See everything inside