"""Extension marketplace — Security Sentinel : chasse aux vulnérabilités locales. Outils : - ``sec_scan_code`` — patterns dangereux (injection, XSS, pickle, exec…) ; - ``sec_scan_secrets`` — secrets / clés / tokens à haute confiance ; - ``sec_supply_chain`` — revue des manifests de dépendances. """ from __future__ import annotations import json import re from pathlib import Path _SKIP = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build", ".tox"} _CODE = { ".py", ".js", ".ts", ".tsx", ".jsx", ".go", ".rs", ".java", ".php", ".rb", ".sh", ".bash", ".zsh", } _MANIFESTS = { "requirements.txt", "requirements-dev.txt", "pyproject.toml", "package.json", "package-lock.json", "pnpm-lock.yaml", "yarn.lock", "Pipfile", "poetry.lock", "go.mod", "Cargo.toml", } _DANGER = [ ("exec_eval", re.compile(r"\b(eval|exec|Function)\s*\("), "high", "Exécution dynamique de code"), ("pickle", re.compile(r"\bpickle\.(loads|load)\s*\("), "high", "Désérialisation pickle unsafe"), ("yaml_unsafe", re.compile(r"\byaml\.(load)\s*\("), "high", "yaml.load sans SafeLoader"), ("shell_true", re.compile(r"subprocess\.[a-zA-Z_]+\([^)]*shell\s*=\s*True"), "high", "subprocess shell=True"), ("os_system", re.compile(r"\bos\.system\s*\("), "high", "os.system"), ("sql_format", re.compile(r"(execute|executemany)\s*\(\s*(f['\"]|['\"].*%|['\"].*\.format)"), "high", "SQL potentiellement interpolé"), ("innerhtml", re.compile(r"\.innerHTML\s*="), "med", "innerHTML — risque XSS"), ("document_write", re.compile(r"document\.write\s*\("), "med", "document.write"), ("dangerously_html", re.compile(r"dangerouslySetInnerHTML"), "med", "React dangerouslySetInnerHTML"), ("verify_false", re.compile(r"verify\s*=\s*False|rejectUnauthorized\s*:\s*false"), "med", "TLS verify désactivé"), ("debug_true", re.compile(r"DEBUG\s*=\s*True|app\.run\([^)]*debug\s*=\s*True"), "low", "Mode debug exposé"), ] _SECRET = [ ("aws_key", re.compile(r"AKIA[0-9A-Z]{16}")), ("generic_api", re.compile(r"(?i)(api[_-]?key|secret|token|passwd|password)\s*[:=]\s*['\"][^'\"]{8,}['\"]")), ("private_key", re.compile(r"-----BEGIN (RSA |EC |OPENSSH )?PRIVATE KEY-----")), ("jwt", re.compile(r"eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}")), ("slack", re.compile(r"xox[baprs]-[0-9A-Za-z-]{10,}")), ("github_pat", re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}")), ] _RISKY_PYPI = { "pickle5": "souvent inutile / surface désérialisation", "crypto": "package trompeur — préférer cryptography", "acme-challenger": "typosquat fréquent historiquement", } def register(workspace: Path) -> list: """Enregistre les outils Security Sentinel.""" def _root(where: str) -> Path: p = Path(where).expanduser() root = p.resolve() if p.is_absolute() else (workspace / where).resolve() if not root.exists(): raise FileNotFoundError(f"introuvable: {root}") return root def _iter_files(root: Path, suffixes: set[str] | None, max_files: int) -> list[Path]: out: list[Path] = [] iterator = root.rglob("*") if root.is_dir() else iter([root]) for path in iterator: if any(part in _SKIP for part in path.parts): continue if not path.is_file(): continue if suffixes is not None and path.suffix.lower() not in suffixes: continue out.append(path) if len(out) >= max_files: break return out def sec_scan_code(where: str = ".", max_files: int = 250, max_hits: int = 80) -> str: """Scanne le code pour des patterns de vulnérabilité classiques. Args: where: Racine ou fichier. max_files: Plafond de fichiers. max_hits: Plafond de findings. """ try: root = _root(where) except Exception as exc: # noqa: BLE001 return json.dumps({"error": str(exc)}, ensure_ascii=False) hits: list[dict] = [] scanned = 0 for path in _iter_files(root, _CODE, max_files): scanned += 1 try: lines = path.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: continue rel = str(path.relative_to(workspace)) if path.is_relative_to(workspace) else str(path) for i, line in enumerate(lines, 1): for code, rx, sev, title in _DANGER: if rx.search(line): hits.append( { "severity": sev, "rule": code, "title": title, "path": rel, "line": i, "snippet": line.strip()[:180], } ) if len(hits) >= max_hits: break if len(hits) >= max_hits: break if len(hits) >= max_hits: break sev_rank = {"high": 0, "med": 1, "low": 2} hits.sort(key=lambda h: (sev_rank.get(h["severity"], 9), h["path"], h["line"])) return json.dumps( { "root": str(root), "files_scanned": scanned, "findings": hits, "summary": { "high": sum(1 for h in hits if h["severity"] == "high"), "med": sum(1 for h in hits if h["severity"] == "med"), "low": sum(1 for h in hits if h["severity"] == "low"), }, "disclaimer": "Heuristique locale — pas un substitut à CodeQL/Semgrep/audit humain.", }, ensure_ascii=False, indent=2, ) def sec_scan_secrets(where: str = ".", max_files: int = 300, max_hits: int = 60) -> str: """Détecte des secrets à haute confiance dans le workspace. Args: where: Racine. max_files: Plafond fichiers texte. max_hits: Plafond findings. """ try: root = _root(where) except Exception as exc: # noqa: BLE001 return json.dumps({"error": str(exc)}, ensure_ascii=False) textish = _CODE | {".env", ".ini", ".cfg", ".yml", ".yaml", ".toml", ".json", ".md", ".txt", ".pem", ".key"} hits: list[dict] = [] for path in _iter_files(root, None, max_files): if path.suffix.lower() not in textish and path.name not in {".env", ".env.local"}: # encore autoriser fichiers sans suffix suspects if not path.name.startswith(".env"): continue try: raw = path.read_text(encoding="utf-8", errors="replace") except OSError: continue # ignore binaries-ish if "\x00" in raw[:2048]: continue rel = str(path.relative_to(workspace)) if path.is_relative_to(workspace) else str(path) for i, line in enumerate(raw.splitlines(), 1): if "EXAMPLE" in line or "YOUR_" in line or "changeme" in line.lower(): continue for rule, rx in _SECRET: if rx.search(line): redacted = rx.sub("***", line.strip())[:160] hits.append({"rule": rule, "path": rel, "line": i, "snippet": redacted}) if len(hits) >= max_hits: break if len(hits) >= max_hits: break if len(hits) >= max_hits: break return json.dumps( { "root": str(root), "findings": hits, "count": len(hits), "advice": "Rotation immédiate si secret réel ; ajouter au .gitignore / secret manager.", }, ensure_ascii=False, indent=2, ) def sec_supply_chain(where: str = ".") -> str: """Passe en revue les manifests de dépendances (risques / typosquats connus). Args: where: Racine projet. """ try: root = _root(where) except Exception as exc: # noqa: BLE001 return json.dumps({"error": str(exc)}, ensure_ascii=False) found = [] warnings = [] for path in root.rglob("*"): if any(part in _SKIP for part in path.parts): continue if path.is_file() and path.name in _MANIFESTS: rel = str(path.relative_to(workspace)) if path.is_relative_to(workspace) else str(path) found.append(rel) try: text = path.read_text(encoding="utf-8", errors="replace").lower() except OSError: continue for pkg, why in _RISKY_PYPI.items(): if re.search(rf"\b{re.escape(pkg)}\b", text): warnings.append({"manifest": rel, "package": pkg, "reason": why}) if path.name == "package.json" and "\"*\"" in text: warnings.append({"manifest": rel, "package": "*", "reason": "version * — builds non reproductibles"}) if path.name == "requirements.txt": for line in text.splitlines(): line = line.strip() if line and not line.startswith("#") and "==" not in line and not line.startswith("-"): warnings.append( { "manifest": rel, "package": line.split("[")[0].split(">")[0].split("<")[0], "reason": "pas de pin == — supply-chain drift", } ) return json.dumps( { "root": str(root), "manifests": found[:50], "warnings": warnings[:80], "checklist": [ "Pin des versions (lockfile)", "Audit régulier (pip-audit / npm audit)", "Éviter install depuis URL non vérifiées", "Séparer deps runtime / dev", ], }, ensure_ascii=False, indent=2, ) return [sec_scan_code, sec_scan_secrets, sec_supply_chain]