scanner/update_rules.py

"""
VibeSec — Semgrep Rules Dynamic Sync Engine
Checks, validates, and atomically updates local community rulesets from the official Semgrep registry.
Supports both JSON and YAML registry formats.
"""
from __future__ import annotations
import os
import sys
import json
import hashlib
import tempfile
import logging
from pathlib import Path
import requests

# Set up logging matching vibesec pattern
logger = logging.getLogger("vibesec.rules_sync")
if not logger.handlers:
    import sys
    sh = logging.StreamHandler(sys.stdout)
    sh.setFormatter(logging.Formatter('[Rules Sync] %(asctime)s - %(levelname)s - %(message)s'))
    logger.addHandler(sh)
    logger.setLevel(logging.INFO)

RULES_DIR = Path(__file__).parent.parent / "rules"

# Maps Semgrep registry rule pack names to our local target files
RULESET_MAPPINGS = {
    "p/owasp-top-ten": "community-owasp-top-ten.json",
    "p/javascript": "community-javascript.json",
    "p/nodejs": "community-nodejs.json",
    "p/python": "community-python.json",
    "p/react": "community-react.json",
}

def get_semgrep_executable() -> str:
    """Finds the semgrep executable in the active python virtual environment or PATH."""
    semgrep_bin = "semgrep"
    venv_bin_dir = os.path.dirname(sys.executable)
    for ext in ["", ".exe"]:
        candidate = os.path.join(venv_bin_dir, f"semgrep{ext}")
        if os.path.exists(candidate):
            return candidate
    return semgrep_bin

def compute_sha256(file_path: Path) -> str:
    """Computes SHA-256 hash of a file."""
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as f:
        while chunk := f.read(8192):
            sha256.update(chunk)
    return sha256.hexdigest()

def validate_ruleset_with_semgrep(rules_file_path: Path) -> bool:
    """Uses basic syntax parsing (JSON/YAML) to ensure the ruleset is syntactically sound without timeouts."""
    if not rules_file_path.exists():
        return False
    ext = rules_file_path.suffix.lower()
    try:
        content = rules_file_path.read_text(encoding="utf-8")
        if ext == ".json":
            import json
            json.loads(content)
        elif ext in {".yaml", ".yml"}:
            import yaml
            yaml.safe_load(content)
        return True
    except Exception as e:
        logger.error(f"Syntax validation failed for {rules_file_path.name}: {e}")
        return False

def sync_single_ruleset(registry_name: str, local_filename: str) -> bool:
    """
    Downloads, validates, and atomically updates a single ruleset file.
    Returns True if an update was written, False if skipped or failed.
    """
    local_path = RULES_DIR / local_filename
    temp_file_path = None
    
    url = f"https://semgrep.dev/c/{registry_name}"
    logger.info(f"Checking {registry_name} -> {local_filename}...")
    
    try:
        # 1. Download
        headers = {"User-Agent": "VibeSec-Rules-Sync/2.0 (Security Scanner Sync)"}
        response = requests.get(url, headers=headers, timeout=30)
        if response.status_code != 200:
            logger.error(f"  [ERROR] Failed to download {registry_name}: HTTP {response.status_code}")
            return False
            
        # 2. Parse and Validate JSON or YAML structure
        data = None
        try:
            data = response.json()
        except ValueError:
            # Fallback to YAML parsing
            try:
                import yaml
                data = yaml.safe_load(response.text)
            except Exception as yaml_err:
                logger.error(f"  [ERROR] Registry returned invalid JSON and YAML formats for {registry_name}: {yaml_err}")
                return False
            
        if not isinstance(data, dict) or "rules" not in data or not isinstance(data["rules"], list):
            logger.error(f"  [ERROR] Invalid ruleset structure in registry response for {registry_name}.")
            return False
            
        rule_count = len(data["rules"])
        if rule_count == 0:
            logger.error(f"  [ERROR] Downloaded ruleset for {registry_name} is empty (0 rules). Aborting.")
            return False
            
        # 3. Create a temporary file to run Semgrep validation
        os.makedirs(RULES_DIR, exist_ok=True)
        # Use tempfile inside the target rules directory to guarantee they're on the same drive (needed for atomic os.replace)
        with tempfile.NamedTemporaryFile(dir=str(RULES_DIR), suffix=".json", delete=False, mode="w", encoding="utf-8") as tf:
            json.dump(data, tf, indent=2)
            temp_file_path = Path(tf.name)
            
        # 4. Perform dynamic validation using Semgrep CLI compiler validation
        if not validate_ruleset_with_semgrep(temp_file_path):
            logger.error(f"  [ERROR] Semgrep compilation validation failed for downloaded {registry_name}. Rejecting update.")
            try:
                os.unlink(temp_file_path)
            except Exception:
                pass
            return False
            
        # 5. Compare with current local ruleset via SHA-256 hash
        new_hash = compute_sha256(temp_file_path)
        
        if local_path.exists():
            old_hash = compute_sha256(local_path)
            if new_hash == old_hash:
                logger.info(f"  [OK] Up-to-date. (Hash match: {new_hash[:8]})")
                try:
                    os.unlink(temp_file_path)
                except Exception:
                    pass
                return False
                
            logger.info(f"  [UPDATE] Update detected! Hash changed from {old_hash[:8]} to {new_hash[:8]} ({rule_count} rules).")
        else:
            logger.info(f"  [UPDATE] Fresh installation of ruleset ({rule_count} rules).")
            
        # 6. Atomic swap
        os.replace(temp_file_path, local_path)
        logger.info(f"  [SUCCESS] Atomically updated {local_filename} ({(local_path.stat().st_size/1024):.1f} KB)")
        return True
        
    except Exception as e:
        logger.error(f"  [ERROR] Error syncing {registry_name}: {e}")
        if temp_file_path and temp_file_path.exists():
            try:
                os.unlink(temp_file_path)
            except Exception:
                pass
        return False

def sync_all_community_rulesets() -> dict[str, str]:
    """Syncs all community rulesets and returns a summary dict."""
    logger.info("=========================================")
    logger.info("  VibeSec Community Rulesets Sync Engine  ")
    logger.info("=========================================")
    
    summary = {}
    updated_count = 0
    failed_count = 0
    
    for registry_name, local_filename in RULESET_MAPPINGS.items():
        success = sync_single_ruleset(registry_name, local_filename)
        if success:
            summary[registry_name] = "Updated"
            updated_count += 1
        else:
            # We check if local_path exists to classify if it's up to date or failed
            local_path = RULES_DIR / local_filename
            if local_path.exists():
                summary[registry_name] = "Up-to-date"
            else:
                summary[registry_name] = "Failed"
                failed_count += 1
                
    logger.info("\n=========================================")
    logger.info(f" Sync Complete: {updated_count} updated, {failed_count} failed, {len(RULESET_MAPPINGS)-updated_count-failed_count} already current.")
    logger.info("=========================================\n")
    
    return summary

if __name__ == "__main__":
    sync_all_community_rulesets()