#!/usr/bin/env python3 """CodeGuard CI — scan a repository's code and dependencies with CodeGuard, and fail the job when the scan fails your policy. One scan covers the code and its dependencies. The job's exit status is the scan's verdict under the policy your organisation set in CodeGuard: 0 passed the scan passes the policy 1 failed the scan fails it (each failing finding is printed: the package and the version that fixes it, or the file and line) 2 not decided part of the analysis did not finish (pass ``--allow-undecided`` to let the job pass) 3 error the scan could not be run or read (configuration, network, a failed scan) In a pull or merge request the scan is compared with the latest scan of the target branch, and the policy judges only what the change introduces (when the policy says so — the default). Standard library only (Python 3.9 or later). Settings come from the command line or the environment: CODEGUARD_API_URL the CodeGuard API (default https://sast.codesguard.io) CODEGUARD_TOKEN a CI token (Security → CI tokens; it starts with ``cgci_``) CODEGUARD_ORG_ID your organisation's id CODEGUARD_PROJECT_ID the project to scan into (optional for a token made for one project) CODEGUARD_BASE_BRANCH the branch to compare with (default: the pull/merge request's target) Run ``python3 codeguard_ci.py --help`` for every option. """ from __future__ import annotations import argparse import fnmatch import hashlib import http.client import json import os import socket import sys import tempfile import time import urllib.error import urllib.parse import urllib.request import uuid import zipfile from typing import Any __version__ = "1.0.0" EXIT_PASSED, EXIT_FAILED, EXIT_UNDECIDED, EXIT_ERROR = 0, 1, 2, 3 DEFAULT_API_URL = "https://sast.codesguard.io" DEFAULT_EXCLUDES = (".git", ".hg", ".svn") TERMINAL = ("completed", "failed", "cancelled") #: GitHub shows at most 10 error annotations per step. GITHUB_ANNOTATIONS = 10 #: How many failing findings the log lists one by one (the count is always printed). PRINTED = 50 class CiError(Exception): """Something that stops the job with ``EXIT_ERROR``, with what to do about it.""" class _NoRedirect(urllib.request.HTTPRedirectHandler): """A redirect is refused, never followed: following one would send the token to whatever address it names (and turn an upload into a GET). The address to configure is the one it names.""" def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 — urllib's signature raise CiError(f"CodeGuard answered {code}, a redirect to {newurl}: set CODEGUARD_API_URL to that " "address (the token is never sent on to another one)") _OPENER = urllib.request.build_opener(_NoRedirect) #: Failures of the connection itself, retried: a refused or reset connection, a timeout (on Python #: 3.9 a read timeout is ``socket.timeout``, not ``TimeoutError``), an answer cut short. _TRANSIENT = (urllib.error.URLError, socket.timeout, TimeoutError, ConnectionError, http.client.HTTPException, OSError) # ---- the API ---------------------------------------------------------------------------------------- class Api: def __init__(self, base_url: str, token: str, org_id: str, *, retries: int = 5): self.base = base_url.rstrip("/") self.token = token self.org = org_id self.retries = retries def _request(self, method: str, path: str, *, body: bytes | None = None, content_type: str | None = None, query: dict[str, str] | None = None, accept: str = "application/json") -> tuple[int, bytes]: url = f"{self.base}/orgs/{self.org}{path}" if query: url += "?" + urllib.parse.urlencode(query) headers = {"Authorization": f"Bearer {self.token}", "Accept": accept, "User-Agent": f"codeguard-ci/{__version__}"} if content_type: headers["Content-Type"] = content_type delay = 2.0 for attempt in range(self.retries + 1): request = urllib.request.Request(url, data=body, method=method, headers=headers) try: with _OPENER.open(request, timeout=300) as resp: # noqa: S310 — the configured API return resp.status, resp.read() except urllib.error.HTTPError as exc: payload = exc.read() # A server error or a busy gateway is retried; an answer about the request is not. if exc.code >= 500 and attempt < self.retries: time.sleep(delay) delay = min(delay * 2, 60) continue return exc.code, payload except _TRANSIENT as exc: if attempt < self.retries: time.sleep(delay) delay = min(delay * 2, 60) continue raise CiError(f"CodeGuard could not be reached at {self.base}: {exc}") from exc raise CiError(f"CodeGuard did not answer {method} {path}") # pragma: no cover def json(self, method: str, path: str, payload: Any = None, *, query: dict[str, str] | None = None, expect: tuple[int, ...] = (200, 201, 202)) -> Any: body = json.dumps(payload).encode("utf-8") if payload is not None else None status, data = self._request(method, path, body=body, content_type="application/json" if body else None, query=query) if status not in expect: raise CiError(_refusal(method, path, status, data)) return _json(method, path, data) def multipart(self, path: str, fields: dict[str, str], file_field: str, filename: str, content: bytes, file_type: str = "application/zip") -> Any: boundary = f"----codeguard{uuid.uuid4().hex}" parts: list[bytes] = [] for name, value in fields.items(): parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'.encode()) parts.append( f'--{boundary}\r\nContent-Disposition: form-data; name="{file_field}"; filename="{filename}"\r\n' f"Content-Type: {file_type}\r\n\r\n".encode() ) parts.append(content) parts.append(f"\r\n--{boundary}--\r\n".encode()) status, data = self._request("POST", path, body=b"".join(parts), content_type=f"multipart/form-data; boundary={boundary}") if status not in (200, 201): raise CiError(_refusal("POST", path, status, data)) return _json("POST", path, data) def download(self, path: str, accept: str) -> tuple[int, bytes]: return self._request("GET", path, accept=accept) def _json(method: str, path: str, data: bytes) -> Any: if not data: return None try: return json.loads(data.decode("utf-8")) except (UnicodeDecodeError, ValueError) as exc: raise CiError(f"CodeGuard answered {method} {path} with something that is not JSON " f"({data[:80]!r}…): check CODEGUARD_API_URL") from exc def _refusal(method: str, path: str, status: int, data: bytes) -> str: detail: Any = "" try: detail = json.loads(data.decode("utf-8")).get("detail", "") except (ValueError, AttributeError): detail = data[:300].decode("utf-8", "replace") if isinstance(detail, dict): detail = detail.get("message") or json.dumps(detail) elif isinstance(detail, list): detail = "; ".join(str(d.get("msg", d)) if isinstance(d, dict) else str(d) for d in detail) hint = { 401: " — check CODEGUARD_TOKEN", 402: " — your plan does not include this", 403: " — the token cannot use this organisation or project", 404: " — check CODEGUARD_ORG_ID and CODEGUARD_PROJECT_ID", }.get(status, "") return f"CodeGuard refused {method} {path} ({status}): {detail or 'no reason given'}{hint}" # ---- where the code came from -------------------------------------------------------------------- def detect_source(env: dict[str, str]) -> tuple[dict[str, str], str | None]: """What the CI system says about the code (repository, branch, commit, pull request, run address) and the branch a pull/merge request targets. Only what it states; nothing is guessed.""" source: dict[str, str] = {} base: str | None = None if env.get("GITHUB_ACTIONS") == "true": source["provider"] = "github" repo = env.get("GITHUB_REPOSITORY", "") server = env.get("GITHUB_SERVER_URL", "https://github.com") source["repository"] = repo source["branch"] = env.get("GITHUB_HEAD_REF") or env.get("GITHUB_REF_NAME", "") source["ref"] = env.get("GITHUB_REF", "") source["commit"] = env.get("GITHUB_SHA", "") ref = env.get("GITHUB_REF", "") if ref.startswith("refs/pull/"): source["pull_request"] = ref.split("/")[2] if repo and env.get("GITHUB_RUN_ID"): source["run_url"] = f"{server}/{repo}/actions/runs/{env['GITHUB_RUN_ID']}" base = env.get("GITHUB_BASE_REF") or None elif env.get("GITLAB_CI") == "true": source["provider"] = "gitlab" source["repository"] = env.get("CI_PROJECT_PATH", "") source["branch"] = env.get("CI_MERGE_REQUEST_SOURCE_BRANCH_NAME") or env.get("CI_COMMIT_REF_NAME", "") source["ref"] = env.get("CI_COMMIT_REF_NAME", "") source["commit"] = env.get("CI_COMMIT_SHA", "") if env.get("CI_MERGE_REQUEST_IID"): source["pull_request"] = env["CI_MERGE_REQUEST_IID"] if env.get("CI_JOB_URL"): source["run_url"] = env["CI_JOB_URL"] base = env.get("CI_MERGE_REQUEST_TARGET_BRANCH_NAME") or None else: source["provider"] = "other" limits = {"repository": 512, "branch": 256, "ref": 512, "commit": 64, "pull_request": 64, "run_url": 1024} return {k: v[: limits.get(k, 64)] for k, v in source.items() if v}, base # ---- the archive --------------------------------------------------------------------------------- def archive(root: str, excludes: list[str], skip: set[str]) -> tuple[str, int, str]: """``root`` as a zip in a temporary file (paths relative to it, in a stable order): its path, size and SHA-256. Directories named in ``excludes`` (or matching a pattern) are left out, and the files in ``skip`` (this job's own outputs).""" root = os.path.abspath(root) if not os.path.isdir(root): raise CiError(f"{root} is not a directory") fd, path = tempfile.mkstemp(prefix="codeguard-", suffix=".zip") os.close(fd) files = 0 def excluded(rel: str) -> bool: parts = rel.split("/") return any(fnmatch.fnmatch(p, pattern) for pattern in excludes for p in parts) or \ any(fnmatch.fnmatch(rel, pattern) for pattern in excludes) with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED, strict_timestamps=False) as zf: for current, dirs, names in os.walk(root): rel_dir = os.path.relpath(current, root).replace(os.sep, "/") dirs[:] = sorted(d for d in dirs if not excluded(d if rel_dir == "." else f"{rel_dir}/{d}")) for name in sorted(names): full = os.path.join(current, name) rel = name if rel_dir == "." else f"{rel_dir}/{name}" if excluded(rel) or os.path.abspath(full) in skip or not os.path.isfile(full) or os.path.islink(full): continue zf.write(full, rel) files += 1 if files == 0: os.unlink(path) raise CiError(f"nothing to scan under {root}") digest = hashlib.sha256() with open(path, "rb") as fh: for block in iter(lambda: fh.read(1 << 20), b""): digest.update(block) return path, os.path.getsize(path), digest.hexdigest() def upload(api: Api, project: str, path: str, size: int, sha256: str, name: str) -> str: """Upload the archive (in one request, or in chunks above the service's threshold); its id.""" config = api.json("GET", "/uploads/config") if size > int(config["max_upload_bytes"]): raise CiError(f"the archive is {size:,} bytes; CodeGuard accepts at most {int(config['max_upload_bytes']):,}") if size <= int(config["chunk_threshold_bytes"]): with open(path, "rb") as fh: return api.multipart(f"/projects/{project}/uploads", {}, "file", name, fh.read())["id"] started = api.json("POST", f"/projects/{project}/uploads/init", {"filename": name, "size_bytes": size, "content_type": "application/zip"}) upload_id, chunk = started["upload_id"], int(started["chunk_size"]) with open(path, "rb") as fh: for index, block in enumerate(iter(lambda: fh.read(chunk), b"")): api.multipart(f"/projects/{project}/uploads/{upload_id}/chunk", {"index": str(index)}, "chunk", name, block) api.json("POST", f"/projects/{project}/uploads/{upload_id}/complete", {"sha256": sha256}) return upload_id # ---- the verdict --------------------------------------------------------------------------------- def wait(api: Api, scan_id: str, *, interval: float, out=sys.stdout) -> dict[str, Any]: """Poll the scan until it is done — no limit of our own: a scan takes what it takes, and the CI system's own job timeout bounds the wait.""" started = time.monotonic() last_said = 0.0 while True: scan = api.json("GET", f"/scans/{scan_id}") if scan.get("status") in TERMINAL: return scan if time.monotonic() - last_said >= 60: print(f" … the scan is {scan.get('status')} ({int(time.monotonic() - started)}s)", file=out, flush=True) last_said = time.monotonic() time.sleep(interval) def describe(item: dict[str, Any]) -> str: """One failing finding in one line: the package, the version and what fixes it — or the file and line — and the rules it fails.""" where = item.get("file_path") or "" if where and item.get("line"): where = f"{where}:{item['line']}" if item.get("family") == "license": text = f"{item.get('package')} {item.get('version') or ''} — license {item.get('license') or 'unknown'}".rstrip() elif item.get("family") == "dependency": text = f"{item.get('package')} {item.get('version') or ''} ({item.get('ecosystem') or 'package'}) — {item.get('advisory_id')}, {item.get('severity')}" fix = item.get("upgrade_to") or ", ".join(item.get("fixed_in") or []) text += f" — fixed in {fix}" if fix else " — no fixed version published" else: text = f"{item.get('title')} ({item.get('severity')})" if item.get("new"): text += " — new in this change" return f"{text}{f' — {where}' if where else ''}" def sentences(reasons: list[str] | None) -> str: """A verdict's reasons as running text: each one a sentence. (A failing rule's reason is the rule's own line, written as a list item without a full stop.)""" out = [] for reason in reasons or []: text = str(reason).strip() if text: text = text[:1].upper() + text[1:] out.append(text if text[-1] in ".!?" else f"{text}.") return " ".join(out) def _escape_data(text: str) -> str: """A workflow command's message, escaped as GitHub reads it.""" return text.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A") def _escape_property(text: str) -> str: """A workflow command's property value (``file``, ``title``), escaped as GitHub reads it.""" return _escape_data(text).replace(":", "%3A").replace(",", "%2C") def annotation(item: dict[str, Any], prefix: str) -> str | None: """A GitHub error annotation on the line the finding is at (``prefix``: where the scanned path is in the repository).""" path = item.get("file_path") if not path: return None file = f"{prefix}{path}" if prefix else path title = f"CodeGuard: {item.get('title') or 'finding'}" line = f",line={int(item['line'])}" if item.get("line") else "" return f"::error file={_escape_property(file)}{line},title={_escape_property(title)}::{_escape_data(describe(item))}" def summary_markdown(scan: dict[str, Any], gate: dict[str, Any] | None, app_url: str | None) -> str: lines = ["## CodeGuard scan", ""] if gate is None: lines.append(f"The scan ended **{scan.get('status')}** without a verdict.") return "\n".join(lines) + "\n" icon = {"passed": "✅", "failed": "❌", "not_decided": "⚠️"}.get(gate["status"], "") lines.append(_one_line(f"{icon} **{gate['title']}** — {sentences(gate.get('reasons'))}")) lines.append("") for rule in gate.get("rules") or []: lines.append(_one_line(f"- {'✖' if rule['failed'] else '✓'} {rule['sentence']}")) failing = gate.get("failing") or [] if failing: lines += ["", "| Finding | Rules |", "| --- | --- |"] for item in failing[:PRINTED]: lines.append(f"| {_one_line(describe(item)).replace('|', '/')} | {', '.join(item.get('rules') or [])} |") if gate.get("failing_total", 0) > PRINTED: lines.append(f"| … and {gate['failing_total'] - PRINTED} more | |") if app_url: lines += ["", f"[Open the scan in CodeGuard]({app_url})"] return "\n".join(lines) + "\n" def code_quality(gate: dict[str, Any] | None, prefix: str) -> list[dict[str, Any]]: """The failing findings as a GitLab code quality report (shown on the merge request).""" severity = {"critical": "blocker", "high": "critical", "medium": "major", "low": "minor", "info": "info"} out = [] for item in (gate or {}).get("failing") or []: path = item.get("file_path") or "" out.append({ "description": describe(item), "check_name": "codeguard/" + "+".join(item.get("rules") or ["policy"]), # The finding's identity in every scan (a scan's record id changes each pipeline, and the # merge request then saw every finding as new and every earlier one as resolved). "fingerprint": hashlib.sha256(str(item.get("key") or item.get("record_id")).encode()).hexdigest(), "severity": severity.get(str(item.get("severity")), "major"), "location": {"path": f"{prefix}{path}" if prefix else path, "lines": {"begin": int(item.get("line") or 1)}}, }) return out # ---- the job ------------------------------------------------------------------------------------- class _Parser(argparse.ArgumentParser): """An option the job cannot run with is a configuration error — exit 3 — never "not decided" (argparse's own exit status is 2, which this job reserves for an undecided verdict).""" def error(self, message: str): # type: ignore[override] self.print_usage(sys.stderr) self.exit(EXIT_ERROR, f"{self.prog}: error: {message}\n") def parse_args(argv: list[str], env: dict[str, str]) -> argparse.Namespace: p = _Parser(description="Scan with CodeGuard and fail the job when the scan fails your policy.") p.add_argument("--api-url", default=env.get("CODEGUARD_API_URL") or DEFAULT_API_URL) p.add_argument("--token", default=env.get("CODEGUARD_TOKEN"), help="a CI token (default: CODEGUARD_TOKEN)") p.add_argument("--org", default=env.get("CODEGUARD_ORG_ID"), help="your organisation's id") p.add_argument("--project", default=env.get("CODEGUARD_PROJECT_ID"), help="the project to scan into") p.add_argument("--path", default=env.get("CODEGUARD_PATH") or ".", help="what to scan (default: here)") p.add_argument("--exclude", action="append", default=[], help="a directory or pattern to leave out (repeatable)") p.add_argument("--base-branch", default=env.get("CODEGUARD_BASE_BRANCH"), help="compare with this branch's latest scan (default: the pull/merge request's target)") p.add_argument("--no-diff", action="store_true", help="judge the whole scan, never only what a change introduces") p.add_argument("--sarif", help="write the results file (SARIF 2.1.0) here, for code scanning") p.add_argument("--summary", default=env.get("GITHUB_STEP_SUMMARY"), help="write a markdown summary here") p.add_argument("--codequality", help="write a GitLab code quality report here") p.add_argument("--app-url", default=env.get("CODEGUARD_APP_URL"), help="the CodeGuard app, to link the scan") p.add_argument("--allow-undecided", action="store_true", help="exit 0 when the verdict is not decided") p.add_argument("--poll-interval", type=float, default=10.0) p.add_argument("--version", action="version", version=f"codeguard-ci {__version__}") args = p.parse_args(argv) missing = [name for name, value in (("--token / CODEGUARD_TOKEN", args.token), ("--org / CODEGUARD_ORG_ID", args.org)) if not value] if missing: p.error("missing " + " and ".join(missing)) return args def _one_line(text: Any) -> str: """Text from the scan, printed on ONE log line: a line break inside it can never start a line of its own — on GitHub a line that starts with ``::`` is a command to the runner.""" return " ".join(str(text).splitlines()) def _say_error(message: str, github: bool, out) -> None: print(f"::error::{_escape_data(message)}" if github else f"CodeGuard: {_one_line(message)}", file=out, flush=True) def run(argv: list[str] | None = None, env: dict[str, str] | None = None, out=sys.stdout) -> int: """The job: its exit status is the verdict (0 passed, 1 failed, 2 not decided) or 3 — anything that stops the job before a verdict is read, unforeseen failures included, is an error of the job and never reads as a failed scan.""" env = dict(os.environ if env is None else env) github = env.get("GITHUB_ACTIONS") == "true" try: return _run(sys.argv[1:] if argv is None else argv, env, out, github) except CiError as exc: _say_error(str(exc), github, out) except Exception as exc: # noqa: BLE001 — reported as the job's error, never as a verdict _say_error(f"unexpected error: {exc!r}", github, out) return EXIT_ERROR def _run(argv: list[str], env: dict[str, str], out, github: bool) -> int: args = parse_args(argv, env) api = Api(args.api_url, args.token, args.org) who = api.json("GET", "/ci/whoami") project = args.project or who.get("project_id") if not project: raise CiError("name the project to scan into: --project or CODEGUARD_PROJECT_ID") source, detected_base = detect_source(env) base_branch = None if args.no_diff else (args.base_branch or detected_base) workspace = os.path.abspath(env.get("GITHUB_WORKSPACE") or env.get("CI_PROJECT_DIR") or os.getcwd()) scanned = os.path.abspath(args.path) prefix = os.path.relpath(scanned, workspace).replace(os.sep, "/") prefix = "" if prefix in (".", "") or prefix.startswith("..") else prefix.rstrip("/") + "/" skip = {os.path.abspath(p) for p in (args.sarif, args.summary, args.codequality) if p} print(f"CodeGuard: packing {scanned}", file=out, flush=True) path, size, sha256 = archive(scanned, [*DEFAULT_EXCLUDES, *args.exclude], skip) try: name = (source.get("repository", "").split("/")[-1] or os.path.basename(scanned) or "source") + ".zip" print(f"CodeGuard: uploading {size:,} bytes", file=out, flush=True) upload_id = upload(api, project, path, size, sha256, name) finally: os.unlink(path) body: dict[str, Any] = {"upload_id": upload_id, "source": source} if base_branch: body["base_branch"] = base_branch scan = api.json("POST", f"/projects/{project}/scans", body) scan_id = scan["id"] app_url = f"{args.app_url.rstrip('/')}/dashboard/sast/scans/{scan_id}" if args.app_url else None print(f"CodeGuard: scan {scan_id} started" + (f", compared with the latest scan of {base_branch}" if base_branch else ""), file=out, flush=True) scan = wait(api, scan_id, interval=args.poll_interval, out=out) gate = scan.get("policy_result") if isinstance(scan.get("policy_result"), dict) else None if args.sarif: # The results file is for code scanning; it never changes the verdict's exit status. try: status, data = api.download(f"/scans/{scan_id}/sarif", "application/sarif+json") except CiError as exc: status, data = 0, b"" print(f"CodeGuard: the results file could not be downloaded: {_one_line(exc)}", file=out) if status == 200: with open(args.sarif, "wb") as fh: fh.write(data) print(f"CodeGuard: results file written to {args.sarif}", file=out) elif status: print(f"CodeGuard: the scan has no results file ({status})", file=out) if args.summary: with open(args.summary, "a", encoding="utf-8") as fh: fh.write(summary_markdown(scan, gate, app_url)) if args.codequality: with open(args.codequality, "w", encoding="utf-8") as fh: json.dump(code_quality(gate, prefix), fh, indent=2) if gate is None: reason = scan.get("last_error") or f"the scan ended {scan.get('status')} with no results" _say_error(f"CodeGuard: {reason}" if github else str(reason), github, out) return EXIT_ERROR print(_one_line(f"CodeGuard: {gate['title']} — {sentences(gate.get('reasons'))}"), file=out) for rule in gate.get("rules") or []: print(_one_line(f" {'✖' if rule['failed'] else '✓'} {rule['sentence']}"), file=out) failing = gate.get("failing") or [] for item in failing[:PRINTED]: print(_one_line(f" - {describe(item)}"), file=out) if gate.get("failing_total", 0) > PRINTED: print(f" … and {gate['failing_total'] - PRINTED} more", file=out) if github: for item in failing[:GITHUB_ANNOTATIONS]: line = annotation(item, prefix) if line: print(line, file=out) if app_url: print(f"CodeGuard: {app_url}", file=out) status = gate.get("status") if status == "passed": return EXIT_PASSED if status == "failed": return EXIT_FAILED return EXIT_PASSED if args.allow_undecided else EXIT_UNDECIDED if __name__ == "__main__": sys.exit(run())