#!/usr/bin/env python3 """dvsentry: a command-line client of the DVSentry /v1 API for CI systems. Code Owner: Aslam Jabri. One file, Python 3.9+, standard library only, so any CI runner can fetch it from https:///downloads/dvsentry.py and run it. It is the counterpart of the GitHub App and GitLab integrations for everything else (Jenkins, CircleCI, Azure Pipelines, Bitbucket Pipelines, Buildkite, ...). Configuration comes from the environment, never from arguments, so the token never appears in a process list or a CI log: DVSENTRY_URL https://dvsentry.example.com (http only for localhost) DVSENTRY_TOKEN a token, ideally a service account's with only the scopes the job needs DVSENTRY_TOKEN_FILE or a file holding it Typical CI step (scopes: scans:trigger, scans:read, gates:read): python3 dvsentry.py scan --repository-id "$REPO_ID" --scanner smart --ref "$GIT_COMMIT" (--scanner full, or name scanners: --scanner semgrep-local --scanner gitleaks) Exit codes: 0 the policy gate passed, 1 it failed, 2 the scan did not finish (error, cancelled, still running at --timeout), 3 usage or configuration error, 4 the API refused the request or could not be reached. Local scan, nothing uploaded (no URL or token needed): python3 dvsentry.py local . # scan a working copy python3 dvsentry.py local --staged . # only what is staged for commit (secrets) python3 dvsentry.py install-hook # run that before every commit `local` runs the scanners already installed on this machine (gitleaks, trivy, and semgrep with rules you point it at), offline, and prints what they found. It makes no network call itself, sends nothing to DVSentry, never prints a secret's value, and exits 1 when a finding reaches --fail-on (default: high). """ from __future__ import annotations import argparse import ipaddress import json import os import shutil import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request import uuid __version__ = "1.1.0" API_PREFIX = "/v1" TERMINAL_REPORT_STATUSES = {"completed", "partial", "failed", "cancelled", "timed_out"} EXIT_PASS, EXIT_FAIL, EXIT_INCOMPLETE, EXIT_USAGE, EXIT_API = 0, 1, 2, 3, 4 RETRY_STATUSES = {429, 502, 503, 504} class CliError(Exception): def __init__(self, message: str, exit_code: int = EXIT_API): super().__init__(message) self.exit_code = exit_code class _NoRedirect(urllib.request.HTTPRedirectHandler): """Never follow a redirect: it would carry the token to wherever it points.""" def redirect_request(self, req, fp, code, msg, headers, newurl): raise CliError(f"The server redirected to {newurl}; set DVSENTRY_URL to the final address.") def _urllib_transport(method: str, url: str, headers: dict, body: bytes | None, timeout: float): """Send one request; returns (status, headers, body bytes).""" request = urllib.request.Request(url, data=body, headers=headers, method=method) opener = urllib.request.build_opener(_NoRedirect()) try: with opener.open(request, timeout=timeout) as response: return response.status, dict(response.headers.items()), response.read() except urllib.error.HTTPError as error: return error.code, dict(error.headers.items()), error.read() except (urllib.error.URLError, OSError) as error: raise CliError(f"Could not reach {urllib.parse.urlsplit(url).netloc}: {getattr(error, 'reason', error)}") def _is_loopback(host: str) -> bool: if host == "localhost": return True try: return ipaddress.ip_address(host).is_loopback except ValueError: return False class Client: def __init__(self, base_url: str, token: str, transport=_urllib_transport, timeout: float = 30.0, sleep=time.sleep): parts = urllib.parse.urlsplit(base_url.strip()) if parts.scheme not in {"https", "http"} or not parts.hostname: raise CliError("DVSENTRY_URL must be an https:// URL, e.g. https://dvsentry.example.com", EXIT_USAGE) if parts.scheme == "http" and not _is_loopback(parts.hostname): raise CliError("DVSENTRY_URL must use https:// (plain http is accepted only for localhost)", EXIT_USAGE) if parts.username or parts.password or parts.query or parts.fragment: raise CliError("DVSENTRY_URL must not carry credentials, a query or a fragment", EXIT_USAGE) if not token: raise CliError("Set DVSENTRY_TOKEN (or DVSENTRY_TOKEN_FILE) to an API token", EXIT_USAGE) self.base = f"{parts.scheme}://{parts.netloc}{parts.path.rstrip('/')}" self.token = token self.transport = transport self.timeout = timeout self.sleep = sleep def request(self, method: str, path: str, *, query: dict | None = None, payload: dict | None = None, headers: dict | None = None, retry: bool | None = None): url = self.base + API_PREFIX + path if query: url += "?" + urllib.parse.urlencode({key: value for key, value in query.items() if value is not None}, doseq=True) sent = {"Authorization": f"Bearer {self.token}", "Accept": "application/json", "User-Agent": f"dvsentry-cli/{__version__}", **(headers or {})} body = None if payload is not None: body = json.dumps(payload).encode() sent["Content-Type"] = "application/json" # Reads are always safe to repeat; a write only when it carries an Idempotency-Key. attempts = 4 if (method == "GET" if retry is None else retry) else 1 for attempt in range(attempts): status, response_headers, content = self.transport(method, url, sent, body, self.timeout) if status in RETRY_STATUSES and attempt + 1 < attempts: self.sleep(_retry_after(response_headers, attempt)) continue break if status >= 400: raise CliError(f"{method} {API_PREFIX}{path} answered {status}: {_detail(content)}") if not content: return None try: return json.loads(content) except ValueError: return content.decode("utf-8", "replace") def _retry_after(headers: dict, attempt: int) -> float: for key, value in headers.items(): if key.lower() == "retry-after": try: return min(max(float(value), 1.0), 60.0) except ValueError: break return float(2 ** attempt) def _detail(content: bytes) -> str: try: detail = json.loads(content).get("detail") except (ValueError, AttributeError): detail = None if isinstance(detail, str): return detail if detail is not None: return json.dumps(detail)[:500] return content.decode("utf-8", "replace")[:200] or "no detail" def _token_from_environment(environ) -> str: token = (environ.get("DVSENTRY_TOKEN") or "").strip() token_file = environ.get("DVSENTRY_TOKEN_FILE") if not token and token_file: try: with open(token_file, encoding="utf-8") as handle: token = handle.read().strip() except OSError as error: raise CliError(f"Cannot read DVSENTRY_TOKEN_FILE: {error.strerror}", EXIT_USAGE) return token def _print_json(value, out) -> None: out.write(json.dumps(value, indent=2, sort_keys=True) + "\n") def command_whoami(client: Client, args, out) -> int: me = client.request("GET", "/auth/me") if args.json: _print_json(me, out) else: # dvsentry-pii: ignore[file-written] whoami shows the signed-in member their own address on their own terminal out.write(f"{me.get('username')} ({me.get('email')})\n") return EXIT_PASS def command_repositories(client: Client, args, out) -> int: page = client.request("GET", "/repositories", query={"limit": args.limit, "offset": args.offset}) if args.json: _print_json(page, out) return EXIT_PASS for repository in page.get("items", []): out.write(f"{repository.get('id')} {repository.get('name')} {repository.get('url') or ''}\n") out.write(f"{len(page.get('items', []))} of {page.get('total', 0)} repositories\n") return EXIT_PASS SCAN_PROFILES = ("smart", "full") def _wait(client: Client, report_id: str, timeout: float, interval: float, out, clock=time.monotonic) -> dict: deadline = clock() + timeout last = None while True: report = client.request("GET", f"/security-reports/{report_id}") state = (report.get("status"), report.get("progress")) if state != last: out.write(f"report {report_id}: {state[0]} ({state[1]}%)\n") last = state if report.get("status") in TERMINAL_REPORT_STATUSES or clock() >= deadline: return report client.sleep(interval) def _gate(client: Client, report_id: str, args, out) -> int: result = client.request("GET", f"/security-reports/{report_id}/ci-result") if getattr(args, "json", False): _print_json(result, out) else: out.write(f"{result.get('summary')}\n") for violation in (result.get("policy") or {}).get("violations", []): out.write(f" - {violation.get('policy_name')}: {violation.get('reason')}\n") code = result.get("exit_code") return code if code in (EXIT_PASS, EXIT_FAIL, EXIT_INCOMPLETE) else EXIT_INCOMPLETE def command_scan(client: Client, args, out) -> int: payload = {"repository_id": args.repository_id, "environment": args.environment, "scan_strategy": args.strategy, "name": args.name, "timeout_seconds": args.scan_timeout} profiles = [name for name in args.scanner if name.lower() in SCAN_PROFILES] if profiles and (len(profiles) > 1 or len(args.scanner) > 1): raise CliError("--scanner smart or --scanner full stands alone; or name scanners instead", EXIT_USAGE) if profiles: payload["scan_profile"] = profiles[0].lower() # the server picks the scanners for the repository's stack else: payload["scanners"] = args.scanner if args.ref: payload["git_ref"] = args.ref if args.path: payload["scan_paths"] = args.path # A key per invocation unless the caller pins one (e.g. the CI build id), so # a retried request never queues the same scan twice. key = args.idempotency_key or str(uuid.uuid4()) report = client.request("POST", "/security-reports", payload=payload, headers={"Idempotency-Key": key}, retry=True) report_id = report["id"] out.write(f"queued report {report_id}\n") if args.no_wait: if args.json: _print_json(report, out) return EXIT_PASS report = _wait(client, report_id, args.timeout, args.poll_interval, out) if report.get("status") not in TERMINAL_REPORT_STATUSES: out.write(f"report {report_id} is still {report.get('status')} after {args.timeout:g}s\n") return EXIT_INCOMPLETE return _gate(client, report_id, args, out) def command_wait(client: Client, args, out) -> int: report = _wait(client, args.report_id, args.timeout, args.poll_interval, out) if report.get("status") not in TERMINAL_REPORT_STATUSES: return EXIT_INCOMPLETE return _gate(client, args.report_id, args, out) def command_gate(client: Client, args, out) -> int: return _gate(client, args.report_id, args, out) def command_findings(client: Client, args, out) -> int: page = client.request("GET", f"/security-reports/{args.report_id}/findings", query={"severity": args.severity, "limit": args.limit, "offset": args.offset}) if args.json: _print_json(page, out) return EXIT_PASS for finding in page.get("items", []): where = finding.get("location") or finding.get("target") or "" out.write(f"{finding.get('severity', ''):<8} {finding.get('title')} {where}\n") out.write(f"{len(page.get('items', []))} of {page.get('total', 0)} findings\n") return EXIT_PASS def command_cancel(client: Client, args, out) -> int: report = client.request("POST", f"/security-reports/{args.report_id}/cancel") out.write(f"report {args.report_id}: {report.get('status')}\n") return EXIT_PASS # -- local scan: offline, nothing leaves this machine ------------------------- SEVERITY_ORDER = {"info": 0, "low": 1, "medium": 2, "high": 3, "critical": 4} LOCAL_SCANNERS = ("gitleaks", "trivy", "semgrep") # Never handed to a scanner: a scanner has no business with DVSentry's token. _PRIVATE_ENVIRONMENT = ("DVSENTRY_TOKEN", "DVSENTRY_TOKEN_FILE", "DVSENTRY_URL") def _scanner_environment(environ) -> dict: environment = {key: value for key, value in environ.items() if key not in _PRIVATE_ENVIRONMENT} # Semgrep sends usage metrics unless told not to; local means local. environment["SEMGREP_SEND_METRICS"] = "off" environment["TRIVY_SKIP_DB_UPDATE"] = "true" return environment def _severity(value, default="medium") -> str: text = str(value or "").strip().lower() aliases = {"error": "high", "warning": "medium", "note": "low", "unknown": default, "moderate": "medium"} text = aliases.get(text, text) return text if text in SEVERITY_ORDER else default def _local_commands(name: str, path: str, report: str, args) -> list | None: """The fixed command line for one scanner, or None when it cannot run offline as asked.""" if name == "gitleaks": mode = ["protect", "--staged"] if args.staged else ["detect", "--no-git"] return ["gitleaks", *mode, "--source", path, "--report-format", "json", "--report-path", report, "--redact", "--exit-code", "0", "--no-banner"] if args.staged: return None # only the secret scanner knows what is staged if name == "trivy": return ["trivy", "fs", "--quiet", "--skip-db-update", "--offline-scan", "--scanners", "vuln,misconfig", "--format", "json", "--output", report, path] if name == "semgrep": if not args.semgrep_config: return None # registry rules would be downloaded; only local rules are used return ["semgrep", "scan", "--metrics", "off", "--disable-version-check", "--quiet", "--json", "--output", report, "--config", args.semgrep_config, path] return None def _parse_gitleaks(data) -> list[dict]: return [{"scanner": "gitleaks", "severity": "high", "rule": str(item.get("RuleID") or "secret"), "title": str(item.get("Description") or "Secret"), "location": f"{item.get('File')}:{item.get('StartLine')}"} for item in (data or []) if isinstance(item, dict)] def _parse_trivy(data) -> list[dict]: findings = [] for result in (data or {}).get("Results") or []: target = result.get("Target") for item in result.get("Vulnerabilities") or []: findings.append({"scanner": "trivy", "severity": _severity(item.get("Severity")), "rule": str(item.get("VulnerabilityID")), "title": f"{item.get('PkgName')} {item.get('InstalledVersion')}: " f"{item.get('Title') or item.get('VulnerabilityID')}", "location": target}) for item in result.get("Misconfigurations") or []: if item.get("Status") == "PASS": continue findings.append({"scanner": "trivy", "severity": _severity(item.get("Severity")), "rule": str(item.get("ID") or item.get("AVDID")), "title": str(item.get("Title")), "location": target}) return findings def _parse_semgrep(data) -> list[dict]: return [{"scanner": "semgrep", "severity": _severity((item.get("extra") or {}).get("severity")), "rule": str(item.get("check_id")), "title": str((item.get("extra") or {}).get("message") or item.get("check_id"))[:200], "location": f"{item.get('path')}:{(item.get('start') or {}).get('line')}"} for item in (data or {}).get("results") or [] if isinstance(item, dict)] _PARSERS = {"gitleaks": _parse_gitleaks, "trivy": _parse_trivy, "semgrep": _parse_semgrep} def run_local_scan(path: str, args, environ, *, which=shutil.which, run=subprocess.run) -> dict: """Run each requested scanner that is installed; collect findings and what was skipped.""" findings, ran, skipped = [], [], [] environment = _scanner_environment(environ) with tempfile.TemporaryDirectory(prefix="dvsentry-local-") as directory: for name in (args.scanner or LOCAL_SCANNERS): report = os.path.join(directory, f"{name}.json") command = _local_commands(name, path, report, args) if command is None: if args.staged and not args.scanner: continue # a pre-commit run checks secrets only; nothing to report reason = ("not used with --staged" if args.staged else "pass --semgrep-config with local rules") skipped.append({"scanner": name, "reason": reason}) continue if which(command[0]) is None: skipped.append({"scanner": name, "reason": "not installed"}) continue try: completed = run(command, env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=args.scanner_timeout, check=False) except subprocess.TimeoutExpired: skipped.append({"scanner": name, "reason": f"timed out after {args.scanner_timeout:g}s"}) continue try: with open(report, encoding="utf-8") as handle: content = handle.read() data = json.loads(content) if content.strip() else [] except (OSError, ValueError): stderr = (completed.stderr or b"").decode("utf-8", "replace").strip().splitlines() skipped.append({"scanner": name, "reason": f"failed (exit {completed.returncode})" + (f": {stderr[-1][:200]}" if stderr else "")}) continue findings.extend(_PARSERS[name](data)) ran.append(name) findings.sort(key=lambda item: (-SEVERITY_ORDER[item["severity"]], item["scanner"], item["location"] or "")) return {"path": path, "staged": bool(args.staged), "scanners_run": ran, "skipped": skipped, "findings": findings, "uploaded": False} def command_local(args, out, environ, *, which=shutil.which, run=subprocess.run) -> int: path = os.path.abspath(args.path) if not os.path.isdir(path): raise CliError(f"{args.path} is not a directory", EXIT_USAGE) result = run_local_scan(path, args, environ, which=which, run=run) threshold = SEVERITY_ORDER[args.fail_on] failing = [item for item in result["findings"] if SEVERITY_ORDER[item["severity"]] >= threshold] if args.json: _print_json({**result, "fail_on": args.fail_on, "failing": len(failing)}, out) else: for item in result["findings"]: out.write(f"{item['severity'].upper():<8} {item['scanner']:<8} {item['title']} {item['location'] or ''}\n") for item in result["skipped"]: out.write(f"skipped {item['scanner']}: {item['reason']}\n") out.write(f"{len(result['findings'])} findings, {len(failing)} at or above {args.fail_on}; " f"ran {', '.join(result['scanners_run']) or 'no scanner'}; nothing was uploaded\n") if not result["scanners_run"]: return EXIT_INCOMPLETE return EXIT_FAIL if failing else EXIT_PASS HOOK_MARKER = "# dvsentry local pre-commit hook" def command_install_hook(args, out, environ) -> int: top = args.repository hooks = os.path.join(top, ".git", "hooks") if not os.path.isdir(hooks): raise CliError(f"{top} is not the top of a Git working copy (no .git/hooks)", EXIT_USAGE) hook = os.path.join(hooks, "pre-commit") if os.path.exists(hook) and not args.force: with open(hook, encoding="utf-8", errors="replace") as handle: if HOOK_MARKER not in handle.read(): raise CliError(f"{hook} already exists; pass --force to replace it", EXIT_USAGE) script = os.path.abspath(args.script or __file__) content = (f"#!/bin/sh\n{HOOK_MARKER}: blocks a commit whose staged changes hold a secret.\n" f"# Runs offline; nothing is uploaded. Bypass once with: git commit --no-verify\n" f"exec {_shell_quote(sys.executable or 'python3')} {_shell_quote(script)} local --staged " f"--fail-on {args.fail_on} \"$(git rev-parse --show-toplevel)\"\n") with open(hook, "w", encoding="utf-8") as handle: handle.write(content) os.chmod(hook, 0o755) # noqa: S103 - a git hook must be executable out.write(f"installed {hook}\n") return EXIT_PASS def _shell_quote(value: str) -> str: return "'" + value.replace("'", "'\"'\"'") + "'" def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="dvsentry", description="DVSentry /v1 API client for CI pipelines.", epilog="Set DVSENTRY_URL and DVSENTRY_TOKEN (or DVSENTRY_TOKEN_FILE). " "Exit codes: 0 pass, 1 policy failed, 2 scan incomplete, " "3 usage error, 4 API error.") parser.add_argument("--version", action="version", version=f"dvsentry-cli {__version__}") commands = parser.add_subparsers(dest="command", required=True) def add(name, handler, help_text): command = commands.add_parser(name, help=help_text) command.set_defaults(handler=handler) command.add_argument("--json", action="store_true", help="print the API's JSON instead of text") return command add("whoami", command_whoami, "show the identity the token belongs to") repositories = add("repositories", command_repositories, "list repositories the token can see") repositories.add_argument("--limit", type=int, default=100) repositories.add_argument("--offset", type=int, default=0) def waiting(command): command.add_argument("--timeout", type=float, default=1800, help="seconds to wait for the scan (1800)") command.add_argument("--poll-interval", type=float, default=10, help="seconds between checks (10)") scan = add("scan", command_scan, "start a repository scan, wait for it and apply the policy gate") scan.add_argument("--repository-id", required=True) scan.add_argument("--scanner", action="append", required=True, help="smart or full (scanners picked for the repository's stack, as in the web form), " "or a scanner id; repeat ids for more than one") scan.add_argument("--ref", help="branch, tag or commit to scan") scan.add_argument("--path", action="append", help="path inside the repository; repeat for more than one") scan.add_argument("--environment", choices=("development", "staging", "production"), default="development") scan.add_argument("--strategy", choices=("full", "incremental"), default="full") scan.add_argument("--name", default="CI scan") scan.add_argument("--scan-timeout", type=int, default=600, help="the scanners' own time budget in seconds") scan.add_argument("--idempotency-key", help="reuse to make a retried CI step return the same scan") scan.add_argument("--no-wait", action="store_true", help="queue the scan and exit 0") waiting(scan) wait = add("wait", command_wait, "wait for a report to finish, then apply the policy gate") wait.add_argument("report_id") waiting(wait) gate = add("gate", command_gate, "apply the policy gate to a finished report") gate.add_argument("report_id") findings = add("findings", command_findings, "list a report's findings") findings.add_argument("report_id") findings.add_argument("--severity", choices=("critical", "high", "medium", "low", "info")) findings.add_argument("--limit", type=int, default=200) findings.add_argument("--offset", type=int, default=0) cancel = add("cancel", command_cancel, "cancel a queued or running report") cancel.add_argument("report_id") local = add("local", None, "scan a local folder offline with installed scanners; nothing is uploaded") local.set_defaults(handler=None, local_handler=command_local) local.add_argument("path", nargs="?", default=".") local.add_argument("--staged", action="store_true", help="only changes staged for commit (secret scan)") local.add_argument("--scanner", action="append", choices=LOCAL_SCANNERS, help="scanner to run; repeat for more than one (default: every installed one)") local.add_argument("--semgrep-config", help="a local Semgrep rules file or folder (registry rules are never fetched)") local.add_argument("--fail-on", choices=tuple(SEVERITY_ORDER), default="high", help="exit 1 when a finding is at or above this severity (high)") local.add_argument("--scanner-timeout", type=float, default=600, help="seconds per scanner (600)") hook = add("install-hook", None, "install a Git pre-commit hook that runs `local --staged`") hook.set_defaults(handler=None, local_handler=command_install_hook) hook.add_argument("--repository", default=".", help="top of the Git working copy (.)") hook.add_argument("--fail-on", choices=tuple(SEVERITY_ORDER), default="high") hook.add_argument("--script", help="path to this file to run from the hook (default: this file)") hook.add_argument("--force", action="store_true", help="replace an existing pre-commit hook") return parser def main(argv=None, environ=None, transport=_urllib_transport, out=None, err=None, sleep=time.sleep) -> int: environ = os.environ if environ is None else environ out = out or sys.stdout err = err or sys.stderr args = build_parser().parse_args(argv) try: if getattr(args, "local_handler", None) is not None: # Local commands never build a client: no URL, no token, no request. return args.local_handler(args, out, environ) client = Client(environ.get("DVSENTRY_URL", ""), _token_from_environment(environ), transport=transport, sleep=sleep) return args.handler(client, args, out) except CliError as error: err.write(f"dvsentry: {error}\n") return error.exit_code if __name__ == "__main__": raise SystemExit(main())