Root cause
Path & link handling introduced
feat(0.8.0): MCP server + rule catalog + auto-discovery + enterprise gate
AI-assisted change: Claude
How AI contributed: Direct introductionPublished May 5, 2026 in
jo-jo98/ciguardAll 4 findings in this repository
discover_pipeline_files followed directory and file symlinks outside the requested scan root, allowing a planted link to expose pipeline-shaped files and subsequent contents from outside the root.
Root cause
feat(0.8.0): MCP server + rule catalog + auto-discovery + enterprise gate
AI-assisted change: Claude
Fix
fix(security): close 4 findings from Cycle 1 self-pentest (CYCLE-1-001..004)
AI-assisted fix: Claude Code · Jo.Moore
Case facts
@@ -601,6 +601,22 @@ def main() -> int: "all severity-based exit codes (returns 0 unless an error occurred).", ) + # ---- `mcp` subcommand (v0.8.0): launch the MCP stdio server. No CLI+ # output (would corrupt the stdio protocol). Requires the optional+ # `mcp` extra: `pip install 'ciguard[mcp]'`. Exits non-zero with a+ # clear message if the extra isn't installed.+ mcp_parser = subparsers.add_parser(+ "mcp",+ help="Launch the ciguard Model Context Protocol server (stdio "+ "transport). Exposes scan / scan_repo / explain_rule / "+ "diff_baseline / list_rules tools to MCP-compatible AI "+ "clients (Claude Desktop, Claude Code, Cursor). "+ "Requires `pip install 'ciguard[mcp]'`.",+ )+ # No flags currently — stdio is the only supported transport. Future+ # SSE / HTTP transports would add flags here.+ del mcp_parser # silence unused; argparse keeps a reference internally+ # ---- `baseline` subcommand: write a baseline from a fresh scan, no report. baseline_parser = subparsers.add_parser( "baseline",@@ -648,6 +664,40 @@ def main() -> int: return cmd_scan(args) elif args.command == "baseline": return cmd_baseline(args)+ elif args.command == "mcp":+ # Enterprise gate (v0.8.0). Sysadmins managing corporate fleets can+ # set CIGUARD_MCP_DISABLED=1 (via MDM / Group Policy / shell profile)+ # to prevent individual devs from running a local ciguard MCP server.+ # Common rationale: the org standardises on a centralised MCP+ # gateway that proxies, audits, and authorises tool traffic — local+ # MCP servers would bypass that control plane.+ import os+ disabled_raw = os.environ.get("CIGUARD_MCP_DISABLED", "").strip().lower()+ if disabled_raw in {"1", "true", "yes", "on"}:+ print(+ f"{_RED}Error:{_RESET} ciguard MCP server is disabled by "+ "policy (CIGUARD_MCP_DISABLED is set in this environment).",+ file=sys.stderr,+ )+ print(+ f" {_DIM}Contact your administrator if you need MCP access. "+ f"This typically means the org has standardised on a "+ f"centralised MCP gateway.{_RESET}",+ file=sys.stderr,+ )+ return 2+ try:+ from ciguard.mcp.server import run_stdio+ except ImportError as exc:+ print(+ f"{_RED}Error:{_RESET} MCP support requires the optional "+ "extra. Install with: pip install 'ciguard[mcp]'",+ file=sys.stderr,+ )+ print(f" {exc}", file=sys.stderr)+ return 1+ run_stdio()+ return 0 else: parser.print_help() return 0@@ -0,0 +1,153 @@+"""+Pipeline file auto-discovery (v0.8.0).++Walks a directory tree and returns every pipeline file ciguard can scan.+Used by the MCP server's `scan_repo` tool — and a future Slice 9+`ciguard scan-repo` CLI subcommand will use the same helper.++Discovery is deliberately conservative: filename + content sniff only,+no parsing here. The caller dispatches to the right parser per platform.++Conventions matched:+ - GitLab CI: `.gitlab-ci.yml` at any depth+ - GitHub Actions: `.github/workflows/*.yml` and `*.yaml`+ - Jenkins: `Jenkinsfile`, `Jenkinsfile.*`, `*.jenkinsfile`,+ `*.groovy` (with content sniff to filter out+ non-pipeline groovy)++Excludes by default: `.git/`, `node_modules/`, `venv/`, `.venv/`,+`__pycache__/`, `.tox/`, `dist/`, `build/`, `.pytest_cache/`. Pass+`exclude_dirs=...` to override.+"""+from __future__ import annotations++from dataclasses import dataclass+from pathlib import Path+from typing import Iterable, List, Optional, Set++import re++# We deliberately do NOT use `parser.jenkinsfile.looks_like_jenkinsfile`+# here because that helper accepts any `.groovy` suffix unconditionally+# (it's the right behaviour for the explicit `--input` CLI path, where+# the user has already pointed at a single file). For repo discovery+# we need to be stricter: a Gradle / Spring / Grails project full of+# `.groovy` files should not be flooded with "Jenkinsfile" hits. We+# require an actual `pipeline {` or `node('...') {` token to classify.+_PIPELINE_MARKER = re.compile(r"\b(pipeline\s*\{|node\s*\()")++_DEFAULT_EXCLUDES = {+ ".git",+ "node_modules",+ "venv",+ ".venv",+ "__pycache__",+ ".tox",+ "dist",+ "build",+ ".pytest_cache",+ ".mypy_cache",+ ".ruff_cache",+ ".idea",+ ".vscode",+}+++@dataclass(frozen=True)+class DiscoveredFile:+ """A pipeline file found by the discovery walk. `platform` is one of+ `gitlab-ci`, `github-actions`, `jenkins`."""++ path: Path+ platform: str+++def _is_excluded(part: str, exclude_dirs: Set[str]) -> bool:+ return part in exclude_dirs+++def discover_pipeline_files(+ root: Path,+ *,+ exclude_dirs: Optional[Iterable[str]] = None,+) -> List[DiscoveredFile]:+ """Walk `root` and return every recognised pipeline file. Symlinks+ are followed once but cycles are guarded by tracking visited resolved+ paths."""+ excludes = set(exclude_dirs) if exclude_dirs is not None else set(_DEFAULT_EXCLUDES)+ if not root.exists():+ return []++ found: List[DiscoveredFile] = []+ visited_dirs: Set[Path] = set()++ def walk(d: Path) -> None:+ try:+ resolved = d.resolve()+ except OSError:+ return+ if resolved in visited_dirs:@@ -0,0 +1,174 @@+"""+Rule catalog (v0.8.0).++ciguard's rule metadata (name / description / severity / category / remediation+/ compliance) lives embedded inside `Finding` emissions across the four+analyzer modules (`rules.py`, `gha_rules.py`, `jenkins_rules.py`, `sca_rules.py`).+There has never been a separate registry.++For the MCP server's `list_rules` and `explain_rule` tools — and for any future+docs / VS Code extension / web UI that needs to enumerate rules — we need+canonical rule metadata. Rather than refactor four analyzer modules to extract+a static registry (high blast radius, would touch every rule emission), this+module **harvests the catalog at startup** by running the analyzer against+the labelled bad fixtures and capturing the first emission seen for each+`rule_id`.++What this gives us:+- Zero analyzer changes — the analyzer remains the source of truth for what+ fields a finding has, and this module reads from that source.+- Coverage across every rule that fires on the labelled bad fixtures (which+ is every rule we have regression tests for).++What this does NOT cover:+- Rules that don't fire on any bad fixture — most SCA-* rules need network+ data to fire concretely. To handle those we run an OFFLINE SCA pass+ (cache-only) and harvest whatever does fire. Rules still missing after+ that are listed by ID but with placeholder metadata (clearly marked).++This catalog is built lazily on first access (cache property) and shared+across the process. Cost is one fixture scan (~50ms) per process.+"""+from __future__ import annotations++import os+from dataclasses import dataclass, field+from pathlib import Path+from typing import Dict, List, Optional++from .analyzer.engine import AnalysisEngine+from .models.pipeline import Category, Finding, Severity+++@dataclass+class RuleSpec:+ """Canonical metadata for a single ciguard rule.++ Harvested from the first `Finding` emitted with this `rule_id` during+ the fixture-scan startup pass. `platforms` lists which platforms (out+ of `gitlab-ci`, `github-actions`, `jenkins`, `cross-platform`) the rule+ can fire on — derived from which fixture surfaced it."""++ rule_id: str+ name: str+ description: str+ severity: str+ category: str+ remediation: str+ compliance: Dict[str, List[str]] = field(default_factory=dict)+ platforms: List[str] = field(default_factory=list)+ sample_evidence: Optional[str] = None+ sample_location: Optional[str] = None++ @classmethod+ def from_finding(cls, finding: Finding, platform: str) -> "RuleSpec":+ return cls(+ rule_id=finding.rule_id,+ name=finding.name,+ description=finding.description,+ severity=finding.severity.value+ if isinstance(finding.severity, Severity) else str(finding.severity),+ category=finding.category.value+ if isinstance(finding.category, Category) else str(finding.category),+ remediation=finding.remediation,+ compliance={+ "iso_27001": list(finding.compliance.iso_27001 or []),+ "soc2": list(finding.compliance.soc2 or []),+ "nist": list(finding.compliance.nist or []),+ },+ platforms=[platform],+ sample_evidence=finding.evidence,+ sample_location=finding.location,+ )+++_FIXTURES_DIR = Path(__file__).parent.parent.parent / "tests" / "fixtures"+++def _scan_fixture(path: Path, platform_label: str, *, enable_sca: bool = False) -> List[Finding]:+ """Run the analyzer on a fixture, return every Finding it emitted.@@ -0,0 +1,55 @@+"""HTTP security headers middleware (added v0.8.2).++Sets the standard set of defence-in-depth headers per the OWASP Secure+Headers Project. Per-path CSP carve-out for `/api/docs` because FastAPI's+auto-generated Swagger UI loads scripts/styles from the jsdelivr CDN.++The Web UI is intended for local-only deployment, but adding these+headers does no harm in that context and provides correct posture if a+user ever fronts ciguard with nginx + LetsEncrypt for a tiny private+team. Future PRD Slice 9 (GitHub App) and Slice 12+ (hosted dashboard)+will carry the same middleware forward.+"""+from __future__ import annotations++from starlette.middleware.base import BaseHTTPMiddleware+from starlette.requests import Request+from starlette.responses import Response+++class SecurityHeadersMiddleware(BaseHTTPMiddleware):+ async def dispatch(self, request: Request, call_next) -> Response:+ response = await call_next(request)+ response.headers["X-Content-Type-Options"] = "nosniff"+ response.headers["X-Frame-Options"] = "DENY"+ response.headers["Referrer-Policy"] = "no-referrer"+ response.headers["Permissions-Policy"] = "interest-cohort=()"+ response.headers["Cross-Origin-Opener-Policy"] = "same-origin"+ response.headers["Cross-Origin-Resource-Policy"] = "same-origin"+ # COEP intentionally NOT set to require-corp: Swagger UI's+ # cross-origin assets would break, and we have no SharedArrayBuffer+ # use that would benefit from it.++ # Per-path CSP. The Swagger UI at /api/docs needs jsdelivr.+ if request.url.path.startswith("/api/docs") or request.url.path.startswith("/api/redoc"):+ response.headers["Content-Security-Policy"] = (+ "default-src 'self'; "+ "script-src 'self' https://cdn.jsdelivr.net 'unsafe-inline'; "+ "style-src 'self' https://cdn.jsdelivr.net 'unsafe-inline'; "+ "img-src 'self' data: https://fastapi.tiangolo.com; "+ "font-src 'self' data:; "+ "connect-src 'self'"+ )+ else:+ response.headers["Content-Security-Policy"] = (+ "default-src 'self'; "+ "script-src 'self'; "+ "style-src 'self' 'unsafe-inline'; "+ "img-src 'self' data:; "+ "font-src 'self' data:; "+ "connect-src 'self'"+ )+ return response+++__all__ = ["SecurityHeadersMiddleware"]@@ -70,18 +70,35 @@ def discover_pipeline_files( root: Path, *, exclude_dirs: Optional[Iterable[str]] = None,+ follow_symlinks: bool = False, ) -> List[DiscoveredFile]:- """Walk `root` and return every recognised pipeline file. Symlinks- are followed once but cycles are guarded by tracking visited resolved- paths."""+ """Walk `root` and return every recognised pipeline file.++ `follow_symlinks` defaults to False: symlinked directories AND symlinked+ files are skipped. This prevents path-escape attacks where an attacker+ plants a symlink in a directory the user (or AI agent) scans, causing+ discovery to read pipeline-shaped files from outside `root`. Callers+ with a legitimate reason to follow can opt in explicitly.++ Defence-in-depth: even with `follow_symlinks=True`, results are+ filtered to those whose `.resolve()` lies under `root.resolve()`.+ """ excludes = set(exclude_dirs) if exclude_dirs is not None else set(_DEFAULT_EXCLUDES) if not root.exists(): return [] + try:+ root_resolved = root.resolve()+ except OSError:+ return []+ found: List[DiscoveredFile] = [] visited_dirs: Set[Path] = set() def walk(d: Path) -> None:+ # Skip symlinked directories unless explicitly opted in.+ if not follow_symlinks and d.is_symlink():+ return try: resolved = d.resolve() except OSError:@@ -103,12 +120,29 @@ def walk(d: Path) -> None: continue if not entry.is_file(): continue+ # Skip symlinked files too (a symlink can target a regular file+ # outside root that iterdir() reports as a file).+ if not follow_symlinks and entry.is_symlink():+ continue platform = _classify(entry) if platform is not None: found.append(DiscoveredFile(path=entry, platform=platform)) walk(root)++ # Belt-and-braces: filter results to those that actually live under the+ # resolved root. Catches any symlink-following surface the walker missed+ # (e.g. opt-in `follow_symlinks=True` callers).+ def _under_root(p: Path) -> bool:+ try:+ p.resolve().relative_to(root_resolved)+ return True+ except (OSError, ValueError):+ return False++ found = [df for df in found if _under_root(df.path)]+ # Stable sort for deterministic output: by path string. found.sort(key=lambda df: str(df.path)) return found@@ -45,6 +45,12 @@ DEFAULT_TTL_SECONDS = 24 * 60 * 60 DEFAULT_TIMEOUT_SECONDS = 8 +# Cap on the response body we'll accept. OSV.dev advisory responses are+# typically <100 KB; 5 MB is generous headroom. Prevents memory exhaustion+# from a hostile / compromised endpoint or successful TLS MITM returning a+# multi-GB response.+MAX_RESPONSE_BYTES = 5 * 1024 * 1024+ # OSV.dev ecosystem string for GitHub Actions advisories. Case-sensitive. ECOSYSTEM_GITHUB_ACTIONS = "GitHub Actions" @@ -180,7 +186,12 @@ def _fetch( with urllib.request.urlopen(req, timeout=self.timeout_seconds) as resp: # nosec B310 if resp.status != 200: return None- payload = json.loads(resp.read().decode("utf-8"))+ # Read with size cap (+1) so we can detect overflow. A hostile+ # server returning gigabytes would otherwise OOM-kill ciguard.+ body = resp.read(MAX_RESPONSE_BYTES + 1)+ if len(body) > MAX_RESPONSE_BYTES:+ return None+ payload = json.loads(body.decode("utf-8")) if not isinstance(payload, dict): return None vulns = payload.get("vulns")AI-assisted change ea73b24f88a822c48e5703aa611dd10c46cb624845ba2051714003028975a350 · Fix ffa6892c125ae7c9ba0196e0d97ce9552fb50d0ae1ad91e69c6025c1e571caa8