From 7dbeb2e2572e9a95cecfa47339e2b274cd0023bd Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 5 Aug 2026 08:12:04 +0000 Subject: [PATCH 1/3] Add a Checkmarx report upload example that reproduces `corgea upload` A dependency-free Python script for pipelines that cannot run the CLI binary, plus a runnable CxXMLResults fixture and the sources it points at. Co-authored-by: Ibrahim Rahhal --- examples/README.md | 82 ++++ examples/checkmarx/report.xml | 60 +++ examples/checkmarx/src/db.py | 15 + examples/checkmarx/src/login.py | 11 + examples/upload_checkmarx_report.py | 555 ++++++++++++++++++++++++++++ 5 files changed, 723 insertions(+) create mode 100644 examples/README.md create mode 100644 examples/checkmarx/report.xml create mode 100644 examples/checkmarx/src/db.py create mode 100644 examples/checkmarx/src/login.py create mode 100644 examples/upload_checkmarx_report.py diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..1f20922 --- /dev/null +++ b/examples/README.md @@ -0,0 +1,82 @@ +# Examples + +## `upload_checkmarx_report.py` — upload a Checkmarx report over the HTTP API + +A standalone, dependency-free Python script that does what `corgea upload +` does, for pipelines that cannot run the CLI binary. If you +*can* run the CLI, prefer it: + +```bash +corgea upload report.xml --project-name my-service --wait +``` + +### Try it + +`examples/checkmarx/` is a self-contained fixture: a `CxXMLResults` report plus +the two vulnerable Python files it points at. + +```bash +export CORGEA_TOKEN= # or run `corgea login ` first +cd examples/checkmarx +python3 ../upload_checkmarx_report.py report.xml --project-name checkmarx-demo --wait +``` + +``` +Uploading 2 source file(s) referenced by the report... + uploaded src/db.py + uploaded src/login.py +Uploading the report as project 'checkmarx-demo' (engine=checkmarx)... + +Scan scan-abc-123 created. +https://www.corgea.app/project/42/?scan_id=scan-abc-123 +``` + +Run it from the root of the tree Checkmarx scanned, or point `--source-root` at +that tree: the report's paths are resolved relative to it. + +### Options + +| Flag | Meaning | +|------|---------| +| `--project-name` | Corgea project. Defaults to the git remote's repo name, else the source root's directory name. | +| `--source-root` | Directory the report's file paths are relative to. Defaults to the current directory. | +| `--wait` | Poll until the scan completes, then print a severity summary. | +| `--allow-missing-files` | Warn instead of failing when a referenced source file is absent. | +| `--url` / `--token` | Override `$CORGEA_URL` / `$CORGEA_TOKEN` and `~/.corgea/config.toml`. | + +Accepted reports: `CxXMLResults` XML, Checkmarx CLI JSON +(`totalCount`/`results`/`scanID`), and Checkmarx web JSON +(`scanResults`/`reportId`). All three upload under `engine=checkmarx`. + +### The API calls + +Corgea analyzes findings against the source they point at, so the referenced +files are uploaded first. A client-generated `run_id` is what ties those uploads +to the report that follows. + +``` +GET /api/v1/verify +POST /api/v1/code-upload?run_id=&path= # once per file, multipart "file" +POST /api/v1/scan-upload?engine=checkmarx&run_id=&project=&ci=&ci_platform= +POST /api/v1/git-config-upload?run_id= # if .git/config exists +GET /api/v1/scan/ # --wait, until status == "complete" +GET /api/v1/scan//issues?page=&page_size=30 # --wait +``` + +Every request carries `CORGEA-SOURCE` plus either `CORGEA-TOKEN: ` or, for +a JWT, `Authorization: Bearer `. `scan-upload` sends the report as the +raw body under `Content-Type: application/json` even when it is XML — `engine` +is what selects the parser server side. Reports over 50 MB are split into 1 MB +chunks carrying `Upload-Offset` and `Upload-Length`. + +`tests/cloud_commands_e2e/checkmarx_example.rs` runs this script and `corgea +upload` against the same stub server and asserts both produce the same request +sequence, so the example cannot drift from the CLI. + +## `deps_skill.rs` + +Prints or refreshes the generated section of `skills/corgea/SKILL.md`. + +```bash +cargo run --example deps_skill -- [print|check|update] +``` diff --git a/examples/checkmarx/report.xml b/examples/checkmarx/report.xml new file mode 100644 index 0000000..4a2a2c1 --- /dev/null +++ b/examples/checkmarx/report.xml @@ -0,0 +1,60 @@ + + + + + + + /src/login.py + 6 + 18 + 1 + username + ParamDecl + 8 + + + 6 + def authenticate(username, password): + + + + + /src/db.py + 14 + 24 + 2 + query + StringLiteral + 5 + + + 14 + query = "SELECT id, role FROM users WHERE name = '" + username + "'" + + + + + + + + + + + /src/db.py + 6 + 16 + 1 + DB_PASSWORD + StringLiteral + 11 + + + 6 + DB_PASSWORD = "s3cr3t-admin-pw" + + + + + + + diff --git a/examples/checkmarx/src/db.py b/examples/checkmarx/src/db.py new file mode 100644 index 0000000..3f95e18 --- /dev/null +++ b/examples/checkmarx/src/db.py @@ -0,0 +1,15 @@ +"""Deliberately vulnerable sample code for the Checkmarx upload example.""" + +import sqlite3 + +DB_HOST = "db.internal.example.com" +DB_PASSWORD = "s3cr3t-admin-pw" + + +def connect(): + return sqlite3.connect("app.db") + + +def find_user(connection, username): + query = "SELECT id, role FROM users WHERE name = '" + username + "'" + return connection.execute(query).fetchone() diff --git a/examples/checkmarx/src/login.py b/examples/checkmarx/src/login.py new file mode 100644 index 0000000..90d1f68 --- /dev/null +++ b/examples/checkmarx/src/login.py @@ -0,0 +1,11 @@ +"""Deliberately vulnerable sample code for the Checkmarx upload example.""" + +from db import connect, find_user + + +def authenticate(username, password): + connection = connect() + user = find_user(connection, username) + if user is None: + return None + return {"id": user[0], "role": user[1]} diff --git a/examples/upload_checkmarx_report.py b/examples/upload_checkmarx_report.py new file mode 100644 index 0000000..d4969dd --- /dev/null +++ b/examples/upload_checkmarx_report.py @@ -0,0 +1,555 @@ +#!/usr/bin/env python3 +"""Upload a Checkmarx report to Corgea over the raw HTTP API. + +This reproduces what `corgea upload ` does, so it can be used +as a reference for wiring Corgea into a pipeline that cannot run the CLI binary. +The request sequence is: + + GET /api/v1/verify + POST /api/v1/code-upload?run_id=&path= (per file) + POST /api/v1/scan-upload?engine=checkmarx&run_id=&project=... + POST /api/v1/git-config-upload?run_id= (if .git/config exists) + GET /api/v1/scan/ (--wait) + GET /api/v1/scan//issues?page=N&page_size=30 (--wait) + +Checkmarx findings reference source files by path. Corgea needs the matching +source, so every path named by the report is uploaded under a shared `run_id` +before the report itself; the `run_id` is what ties the two together server side. + +Usage: + export CORGEA_TOKEN= + ./upload_checkmarx_report.py checkmarx_report.xml --project-name my-service --wait + +Only the Python standard library is required. +""" + +from __future__ import annotations + +import argparse +import json +import mimetypes +import os +import re +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +import uuid +import xml.etree.ElementTree as ElementTree +from pathlib import Path +from typing import Any, Iterable + +DEFAULT_URL = "https://www.corgea.app" +API_BASE = "/api/v1" +ENGINE = "checkmarx" + +# Reports larger than this are streamed as `Upload-Offset`/`Upload-Length` +# chunks instead of a single body, matching the CLI's limits. +MAX_SINGLE_UPLOAD_BYTES = 50 * 1024 * 1024 +CHUNK_BYTES = 1024 * 1024 + +POLL_INTERVAL_SECONDS = 1 +ISSUE_PAGE_SIZE = 30 +URGENCY_ORDER = ("CR", "HI", "ME", "LO") + + +class CorgeaError(Exception): + """A request was rejected, or the report could not be parsed.""" + + +# -------------------------------------------------------------------------- +# Configuration +# -------------------------------------------------------------------------- + + +def read_cli_config() -> dict[str, str]: + """Read `~/.corgea/config.toml`, the file `corgea login` writes.""" + path = Path.home() / ".corgea" / "config.toml" + if not path.is_file(): + return {} + text = path.read_text(encoding="utf-8") + try: + import tomllib + + return {k: v for k, v in tomllib.loads(text).items() if isinstance(v, str)} + except ImportError: # Python < 3.11 + pairs = re.findall(r'^\s*(\w+)\s*=\s*"([^"]*)"\s*$', text, re.MULTILINE) + return dict(pairs) + + +def resolve_url(override: str | None) -> str: + config = read_cli_config() + url = override or os.environ.get("CORGEA_URL") or config.get("url") or DEFAULT_URL + return url.rstrip("/") + + +def resolve_token(override: str | None) -> str: + config = read_cli_config() + token = override or os.environ.get("CORGEA_TOKEN") or config.get("token") or "" + if not token: + raise CorgeaError( + "No Corgea token. Set CORGEA_TOKEN, pass --token, or run `corgea login `." + ) + return token + + +def auth_headers(token: str) -> dict[str, str]: + """A JWT goes in `Authorization`; an opaque token in `CORGEA-TOKEN`.""" + segments = token.split(".", 3) + if len(segments) == 3 and all(segments): + headers = {"Authorization": f"Bearer {token}"} + else: + headers = {"CORGEA-TOKEN": token} + headers["CORGEA-SOURCE"] = os.environ.get("CORGEA_SOURCE", "cli") + return headers + + +def sanitize_project_name(name: str) -> str: + return "".join(c if (c.isalnum() or c in "-_.") else "_" for c in name) + + +def determine_project_name(provided: str | None, root: Path) -> str: + """`--project-name`, else the git remote's repo name, else the directory.""" + if provided: + return sanitize_project_name(provided) + + git_config = root / ".git" / "config" + if git_config.is_file(): + match = re.search( + r'\[remote "origin"\][^\[]*?url\s*=\s*(\S+)', + git_config.read_text(encoding="utf-8", errors="replace"), + re.DOTALL, + ) + if match: + repo = match.group(1).removesuffix(".git").rstrip("/") + tail = re.split(r"[/:]", repo)[-1].strip() + if tail: + return sanitize_project_name(tail) + + return sanitize_project_name(root.resolve().name) + + +def ci_context(project: str, github_env: dict[str, str]) -> tuple[bool, str, str]: + """Corgea keys CI scans to `{repo}-{pr}` so a PR gets one project.""" + in_ci = "CI" in os.environ and "GITHUB_ACTIONS" in os.environ + platform = "github" if "GITHUB_ACTIONS" in os.environ else "unknown" + if in_ci: + project = f"{github_env['GITHUB_REPOSITORY']}-{github_env['GITHUB_PR']}" + return in_ci, platform, project + + +# -------------------------------------------------------------------------- +# Checkmarx report parsing +# -------------------------------------------------------------------------- + + +def _strip_leading_separator(path: str) -> str: + return path.lstrip("/\\") + + +def parse_checkmarx_xml(text: str) -> list[str]: + """`CxXMLResults`: paths live on `Result/@FileName` or `` nodes.""" + paths: list[str] = [] + for element in ElementTree.fromstring(text).iter(): + tag = element.tag.rpartition("}")[2] + if tag == "Result": + candidate = element.get("FileName", "") + elif tag == "FileName": + candidate = element.text or "" + else: + continue + candidate = _strip_leading_separator(candidate.strip()) + if candidate: + paths.append(candidate) + return paths + + +def _paths_from_nodes(nodes: Iterable[Any]) -> Iterable[str]: + # Checkmarx JSON prefixes every path with the scan root separator, which the + # CLI drops by removing the first character rather than stripping a set of + # separators. Mirror that so both produce identical `path` query values. + for node in nodes: + if isinstance(node, dict) and isinstance(node.get("fileName"), str): + yield node["fileName"][1:] + + +def parse_checkmarx_cli_json(data: dict[str, Any]) -> list[str]: + paths: list[str] = [] + for result in data.get("results") or []: + nodes = (result.get("data") or {}).get("nodes") or [] + paths.extend(_paths_from_nodes(nodes)) + return paths + + +def parse_checkmarx_web_json(data: dict[str, Any]) -> list[str]: + paths: list[str] = [] + sast = ((data.get("scanResults") or {}).get("sast") or {}) + for language in sast.get("languages") or []: + for query in language.get("queries") or []: + for vulnerability in query.get("vulnerabilities") or []: + paths.extend(_paths_from_nodes(vulnerability.get("nodes") or [])) + return paths + + +def parse_report(text: str) -> list[str]: + """Return the source paths a Checkmarx report references, in report order.""" + if text.startswith(" None: + self.base_url = base_url + self.headers = auth_headers(token) + self.timeout = timeout + + def _url(self, path: str, query: dict[str, str] | None = None) -> str: + url = f"{self.base_url}{API_BASE}{path}" + if query: + # `safe="/"` keeps repo-relative paths readable in the query string. + encoded = urllib.parse.urlencode( + query, quote_via=urllib.parse.quote, safe="/" + ) + url = f"{url}?{encoded}" + return url + + def request( + self, + method: str, + path: str, + query: dict[str, str] | None = None, + body: bytes | None = None, + headers: dict[str, str] | None = None, + ) -> tuple[int, dict[str, str], bytes]: + request = urllib.request.Request( + self._url(path, query), data=body, method=method + ) + for name, value in {**self.headers, **(headers or {})}.items(): + request.add_header(name, value) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + return response.status, dict(response.headers), response.read() + except urllib.error.HTTPError as error: + return error.code, dict(error.headers), error.read() + except urllib.error.URLError as error: + raise CorgeaError(f"{method} {path} failed: {error.reason}") from error + + def json_get(self, path: str, query: dict[str, str] | None = None) -> dict[str, Any]: + status, _, body = self.request("GET", path, query=query) + if status >= 400: + raise CorgeaError(f"GET {path} returned {status}: {body.decode(errors='replace')}") + return json.loads(body or b"{}") + + +def encode_multipart_file(field: str, path: Path) -> tuple[str, bytes]: + """Build a `multipart/form-data` body holding a single file part.""" + boundary = uuid.uuid4().hex + mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream" + head = ( + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="{field}"; filename="{path.name}"\r\n' + f"Content-Type: {mime}\r\n\r\n" + ).encode() + tail = f"\r\n--{boundary}--\r\n".encode() + return f"multipart/form-data; boundary={boundary}", head + path.read_bytes() + tail + + +# -------------------------------------------------------------------------- +# Upload steps +# -------------------------------------------------------------------------- + + +def verify_token(api: CorgeaApi) -> None: + body = api.json_get("/verify") + if body.get("status") != "ok": + raise CorgeaError(f"Token rejected by {api.base_url}: {body}") + + +def upload_source_files( + api: CorgeaApi, run_id: str, paths: list[str], root: Path, allow_missing: bool +) -> int: + """Upload each referenced file once, keyed to `run_id` by its repo-relative path.""" + uploaded = 0 + for path in dict.fromkeys(paths): # de-duplicate, keep report order + local = root / path + if not local.is_file(): + message = f"{path} is referenced by the report but missing under {root}" + if not allow_missing: + raise CorgeaError( + f"{message}. Run from the scanned source tree, pass --source-root, " + "or use --allow-missing-files to upload the report without it." + ) + print(f" warning: skipping {message}", file=sys.stderr) + continue + + content_type, body = encode_multipart_file("file", local) + status, _, response = api.request( + "POST", + "/code-upload", + query={"run_id": run_id, "path": path}, + body=body, + headers={"Content-Type": content_type}, + ) + if status >= 400: + raise CorgeaError( + f"code-upload of {path} returned {status}: {response.decode(errors='replace')}" + ) + uploaded += 1 + print(f" uploaded {path}") + return uploaded + + +def upload_report( + api: CorgeaApi, run_id: str, report: str, project: str, in_ci: bool, platform: str +) -> tuple[str, str | None]: + """POST the report and return `(scan_id, project_id)`.""" + query = { + "engine": ENGINE, + "run_id": run_id, + "project": project, + "ci": "true" if in_ci else "false", + "ci_platform": platform, + } + repo_data = os.environ.get("REPO_DATA", "") + if repo_data: + query["repo_data"] = repo_data + + payload = report.encode("utf-8") + # The endpoint is declared JSON even for Checkmarx XML; the `engine` + # query parameter is what selects the parser server side. + headers = {"Content-Type": "application/json"} + + if len(payload) <= MAX_SINGLE_UPLOAD_BYTES: + status, _, response = api.request( + "POST", "/scan-upload", query=query, body=payload, headers=headers + ) + else: + status, response_headers, response = _upload_report_in_chunks( + api, query, payload, headers + ) + del response_headers + + if status >= 400: + raise CorgeaError( + f"scan-upload returned {status}: {response.decode(errors='replace')}" + ) + + body = json.loads(response or b"{}") + scan_id = body.get("sast_scan_id") + if scan_id is None: + raise CorgeaError(f"scan-upload succeeded but returned no sast_scan_id: {body}") + project_id = body.get("project_id") + return str(scan_id), None if project_id is None else str(project_id) + + +def _upload_report_in_chunks( + api: CorgeaApi, query: dict[str, str], payload: bytes, headers: dict[str, str] +) -> tuple[int, dict[str, str], bytes]: + total = len(payload) + offset = 0 + status, response_headers, response = 0, {}, b"" + while offset < total: + chunk = payload[offset : offset + CHUNK_BYTES] + status, response_headers, response = api.request( + "POST", + "/scan-upload", + query=query, + body=chunk, + headers={ + **headers, + "Upload-Offset": str(offset), + "Upload-Length": str(total), + }, + ) + if status >= 400: + return status, response_headers, response + offset += len(chunk) + # A mismatch means chunks landed on different server instances; the + # assembled report would be corrupt, so stop rather than finish it. + acknowledged = response_headers.get("Upload-Offset") + if acknowledged is not None and acknowledged.isdigit(): + if int(acknowledged) != offset: + raise CorgeaError( + f"Upload offset mismatch: server has {acknowledged} bytes, expected {offset}." + ) + print(f" sent {offset}/{total} bytes") + return status, response_headers, response + + +def upload_git_config(api: CorgeaApi, run_id: str, root: Path) -> None: + """Optional: lets Corgea attach the scan to the right repo and branch.""" + git_config = root / ".git" / "config" + if not git_config.is_file(): + return + content_type, body = encode_multipart_file("file", git_config) + status, _, response = api.request( + "POST", + "/git-config-upload", + query={"run_id": run_id}, + body=body, + headers={"Content-Type": content_type}, + ) + if status >= 400: + print( + f" warning: git-config-upload returned {status}: " + f"{response.decode(errors='replace')}", + file=sys.stderr, + ) + + +def scan_url(base_url: str, scan_id: str, project_id: str | None, project: str) -> str: + if project_id: + return f"{base_url}/project/{project_id}/?scan_id={scan_id}" + return f"{base_url}/project/{urllib.parse.quote(project, safe='')}?scan_id={scan_id}" + + +def wait_for_scan(api: CorgeaApi, scan_id: str) -> None: + while True: + scan = api.json_get(f"/scan/{scan_id}") + status = scan.get("status", "") + if status == "complete": + return + print(f" scan status: {status or 'unknown'}") + time.sleep(POLL_INTERVAL_SECONDS) + + +def print_issue_summary(api: CorgeaApi, scan_id: str) -> None: + counts: dict[str, int] = {} + total = 0 + page = 1 + while True: + body = api.json_get( + f"/scan/{scan_id}/issues", + query={"page": str(page), "page_size": str(ISSUE_PAGE_SIZE)}, + ) + issues = body.get("issues") or [] + for issue in issues: + counts[issue.get("urgency", "?")] = counts.get(issue.get("urgency", "?"), 0) + 1 + total += 1 + if page >= int(body.get("total_pages") or 1) or not issues: + break + page += 1 + + print("\nScan Results:\n") + print(f"{'Classification':<20} | Count") + print(f"{'':-<20} | ") + for urgency in URGENCY_ORDER: + print(f"{urgency:<20} | {counts.get(urgency, 0)}") + print(f"{'':-<20} | ") + print(f"{'Total':<20} | {total}") + + +# -------------------------------------------------------------------------- +# Entry point +# -------------------------------------------------------------------------- + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Upload a Checkmarx report to Corgea via the HTTP API.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=( + "Examples:\n" + " %(prog)s checkmarx_report.xml\n" + " %(prog)s cx_results.json --project-name payments-api --wait\n" + ), + ) + parser.add_argument("report", help="Checkmarx report: CxXMLResults XML, or CLI/web JSON") + parser.add_argument( + "--project-name", + help="Corgea project. Defaults to the git repo name, else the source root's name.", + ) + parser.add_argument( + "--source-root", + default=".", + type=Path, + help="Directory the report's file paths are relative to (default: current directory).", + ) + parser.add_argument( + "--wait", + action="store_true", + help="Poll until the scan completes and print an issue summary.", + ) + parser.add_argument( + "--allow-missing-files", + action="store_true", + help="Warn instead of failing when a referenced source file is absent.", + ) + parser.add_argument("--url", help="Corgea base URL (default: $CORGEA_URL or the CLI config).") + parser.add_argument("--token", help="Corgea token (default: $CORGEA_TOKEN or the CLI config).") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + + report_path = Path(args.report) + if not report_path.is_file(): + raise CorgeaError(f"Report not found: {report_path}") + # A BOM would break both the XML declaration check and json.loads. + report = report_path.read_text(encoding="utf-8-sig").strip() + + paths = parse_report(report) + if not paths: + print("No findings in the report, nothing to upload.") + return 0 + + root = args.source_root + api = CorgeaApi(resolve_url(args.url), resolve_token(args.token)) + run_id = str(uuid.uuid4()) + + verify_token(api) + + project = determine_project_name(args.project_name, root) + in_ci, platform, project = ci_context(project, dict(os.environ)) + + print(f"Uploading {len(set(paths))} source file(s) referenced by the report...") + if upload_source_files(api, run_id, paths, root, args.allow_missing_files) == 0: + raise CorgeaError("No source files were uploaded; Corgea cannot analyze the findings.") + + print(f"Uploading the report as project '{project}' (engine={ENGINE})...") + scan_id, project_id = upload_report(api, run_id, report, project, in_ci, platform) + upload_git_config(api, run_id, root) + + url = scan_url(api.base_url, scan_id, project_id, project) + print(f"\nScan {scan_id} created.\n{url}") + + if args.wait: + print("\nWaiting for the scan to complete...") + wait_for_scan(api, scan_id) + print_issue_summary(api, scan_id) + print(f"\n{url}") + + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except CorgeaError as error: + print(f"error: {error}", file=sys.stderr) + sys.exit(1) + except KeyboardInterrupt: + sys.exit(130) From 99be6ce8b8bd897ab2c93fda7a88c41572f46791 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 5 Aug 2026 08:14:02 +0000 Subject: [PATCH 2/3] Simplify Checkmarx upload example to a 2-arg bash script Replace the elaborate Python helper with upload_checkmarx.sh that takes , creates the scan, and exits without waiting. Co-authored-by: Ibrahim Rahhal --- examples/README.md | 79 +--- examples/upload_checkmarx.sh | 127 +++++++ examples/upload_checkmarx_report.py | 555 ---------------------------- 3 files changed, 147 insertions(+), 614 deletions(-) create mode 100755 examples/upload_checkmarx.sh delete mode 100644 examples/upload_checkmarx_report.py diff --git a/examples/README.md b/examples/README.md index 1f20922..1acca19 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,82 +1,43 @@ # Examples -## `upload_checkmarx_report.py` — upload a Checkmarx report over the HTTP API +## `upload_checkmarx.sh` — upload a Checkmarx report -A standalone, dependency-free Python script that does what `corgea upload -` does, for pipelines that cannot run the CLI binary. If you -*can* run the CLI, prefer it: +Creates a Corgea scan from a Checkmarx report. Same API flow as +`corgea upload`, without waiting for the scan to finish. ```bash -corgea upload report.xml --project-name my-service --wait +export CORGEA_TOKEN= +./upload_checkmarx.sh ``` -### Try it +| Arg | Meaning | +|-----|---------| +| `code_path` | Root of the tree Checkmarx scanned (report paths are relative to this) | +| `report_path` | Checkmarx report: `CxXMLResults` XML, CLI JSON, or web JSON | + +Optional env vars: `CORGEA_URL` (default `https://www.corgea.app`), `PROJECT` +(default: basename of `code_path`). -`examples/checkmarx/` is a self-contained fixture: a `CxXMLResults` report plus -the two vulnerable Python files it points at. +### Try it ```bash -export CORGEA_TOKEN= # or run `corgea login ` first -cd examples/checkmarx -python3 ../upload_checkmarx_report.py report.xml --project-name checkmarx-demo --wait +export CORGEA_TOKEN= +./upload_checkmarx.sh ./checkmarx ./checkmarx/report.xml ``` ``` -Uploading 2 source file(s) referenced by the report... - uploaded src/db.py - uploaded src/login.py -Uploading the report as project 'checkmarx-demo' (engine=checkmarx)... - +Uploading 2 source file(s) from .../examples/checkmarx... + src/db.py + src/login.py +Uploading report as project 'checkmarx'... Scan scan-abc-123 created. https://www.corgea.app/project/42/?scan_id=scan-abc-123 ``` -Run it from the root of the tree Checkmarx scanned, or point `--source-root` at -that tree: the report's paths are resolved relative to it. - -### Options - -| Flag | Meaning | -|------|---------| -| `--project-name` | Corgea project. Defaults to the git remote's repo name, else the source root's directory name. | -| `--source-root` | Directory the report's file paths are relative to. Defaults to the current directory. | -| `--wait` | Poll until the scan completes, then print a severity summary. | -| `--allow-missing-files` | Warn instead of failing when a referenced source file is absent. | -| `--url` / `--token` | Override `$CORGEA_URL` / `$CORGEA_TOKEN` and `~/.corgea/config.toml`. | - -Accepted reports: `CxXMLResults` XML, Checkmarx CLI JSON -(`totalCount`/`results`/`scanID`), and Checkmarx web JSON -(`scanResults`/`reportId`). All three upload under `engine=checkmarx`. - -### The API calls - -Corgea analyzes findings against the source they point at, so the referenced -files are uploaded first. A client-generated `run_id` is what ties those uploads -to the report that follows. - -``` -GET /api/v1/verify -POST /api/v1/code-upload?run_id=&path= # once per file, multipart "file" -POST /api/v1/scan-upload?engine=checkmarx&run_id=&project=&ci=&ci_platform= -POST /api/v1/git-config-upload?run_id= # if .git/config exists -GET /api/v1/scan/ # --wait, until status == "complete" -GET /api/v1/scan//issues?page=&page_size=30 # --wait -``` - -Every request carries `CORGEA-SOURCE` plus either `CORGEA-TOKEN: ` or, for -a JWT, `Authorization: Bearer `. `scan-upload` sends the report as the -raw body under `Content-Type: application/json` even when it is XML — `engine` -is what selects the parser server side. Reports over 50 MB are split into 1 MB -chunks carrying `Upload-Offset` and `Upload-Length`. - -`tests/cloud_commands_e2e/checkmarx_example.rs` runs this script and `corgea -upload` against the same stub server and asserts both produce the same request -sequence, so the example cannot drift from the CLI. +Requires `curl` and `python3` (stdlib only — used to parse the report). ## `deps_skill.rs` -Prints or refreshes the generated section of `skills/corgea/SKILL.md`. - ```bash cargo run --example deps_skill -- [print|check|update] ``` diff --git a/examples/upload_checkmarx.sh b/examples/upload_checkmarx.sh new file mode 100755 index 0000000..07fd9ec --- /dev/null +++ b/examples/upload_checkmarx.sh @@ -0,0 +1,127 @@ +#!/usr/bin/env bash +# Upload a Checkmarx report to Corgea (same flow as `corgea upload`). +# +# Usage: +# export CORGEA_TOKEN= +# ./upload_checkmarx.sh +# +# Optional: +# CORGEA_URL Corgea base URL (default: https://www.corgea.app) +# PROJECT Project name (default: basename of ) +# +# Creates the scan and prints the scan URL. Does not wait for it to finish. +set -euo pipefail + +if [[ $# -ne 2 ]]; then + echo "usage: $0 " >&2 + exit 2 +fi + +CODE_PATH=$(cd "$1" && pwd) +REPORT_PATH=$(cd "$(dirname "$2")" && pwd)/$(basename "$2") +BASE_URL="${CORGEA_URL:-https://www.corgea.app}" +BASE_URL="${BASE_URL%/}" +TOKEN="${CORGEA_TOKEN:?set CORGEA_TOKEN}" +PROJECT="${PROJECT:-$(basename "$CODE_PATH")}" +PROJECT=$(printf '%s' "$PROJECT" | tr -c 'A-Za-z0-9._-' '_') +RUN_ID=$(python3 -c 'import uuid; print(uuid.uuid4())') +API="$BASE_URL/api/v1" + +if [[ ! -f "$REPORT_PATH" ]]; then + echo "error: report not found: $REPORT_PATH" >&2 + exit 1 +fi + +# Opaque tokens use CORGEA-TOKEN; JWTs (a.b.c) use Authorization: Bearer. +AUTH_ARGS=(-H "CORGEA-SOURCE: cli") +if [[ "$TOKEN" == *.*.* && "$TOKEN" != *..* ]]; then + AUTH_ARGS+=(-H "Authorization: Bearer $TOKEN") +else + AUTH_ARGS+=(-H "CORGEA-TOKEN: $TOKEN") +fi + +curl -fsS "${AUTH_ARGS[@]}" "$API/verify" >/dev/null + +# Extract repo-relative source paths from Checkmarx XML / CLI JSON / web JSON. +# Mirrors the CLI: strip a leading separator from each path named by the report. +mapfile -t PATHS < <(python3 - "$REPORT_PATH" <<'PY' +import json, sys, xml.etree.ElementTree as ET +from pathlib import Path + +text = Path(sys.argv[1]).read_text(encoding="utf-8-sig").strip() +paths = [] + +if text.startswith("&2 + exit 1 + fi + # Same as the CLI: path is passed raw in the query string. + curl -fsS "${AUTH_ARGS[@]}" \ + -F "file=@${file}" \ + "${API}/code-upload?run_id=${RUN_ID}&path=${rel}" \ + >/dev/null + echo " $rel" +done + +echo "Uploading report as project '$PROJECT'..." +RESPONSE=$(curl -fsS "${AUTH_ARGS[@]}" \ + -H "Content-Type: application/json" \ + --data-binary @"$REPORT_PATH" \ + "${API}/scan-upload?engine=checkmarx&run_id=${RUN_ID}&project=${PROJECT}&ci=false&ci_platform=unknown") + +SCAN_ID=$(printf '%s' "$RESPONSE" | python3 -c 'import json,sys; print(json.load(sys.stdin)["sast_scan_id"])') +PROJECT_ID=$(printf '%s' "$RESPONSE" | python3 -c 'import json,sys; print(json.load(sys.stdin).get("project_id") or "")') + +if [[ -f "$CODE_PATH/.git/config" ]]; then + curl -fsS "${AUTH_ARGS[@]}" \ + -F "file=@${CODE_PATH}/.git/config" \ + "${API}/git-config-upload?run_id=${RUN_ID}" >/dev/null || true +fi + +if [[ -n "$PROJECT_ID" ]]; then + echo "Scan $SCAN_ID created." + echo "${BASE_URL}/project/${PROJECT_ID}/?scan_id=${SCAN_ID}" +else + echo "Scan $SCAN_ID created." + echo "${BASE_URL}/project/${PROJECT}?scan_id=${SCAN_ID}" +fi diff --git a/examples/upload_checkmarx_report.py b/examples/upload_checkmarx_report.py deleted file mode 100644 index d4969dd..0000000 --- a/examples/upload_checkmarx_report.py +++ /dev/null @@ -1,555 +0,0 @@ -#!/usr/bin/env python3 -"""Upload a Checkmarx report to Corgea over the raw HTTP API. - -This reproduces what `corgea upload ` does, so it can be used -as a reference for wiring Corgea into a pipeline that cannot run the CLI binary. -The request sequence is: - - GET /api/v1/verify - POST /api/v1/code-upload?run_id=&path= (per file) - POST /api/v1/scan-upload?engine=checkmarx&run_id=&project=... - POST /api/v1/git-config-upload?run_id= (if .git/config exists) - GET /api/v1/scan/ (--wait) - GET /api/v1/scan//issues?page=N&page_size=30 (--wait) - -Checkmarx findings reference source files by path. Corgea needs the matching -source, so every path named by the report is uploaded under a shared `run_id` -before the report itself; the `run_id` is what ties the two together server side. - -Usage: - export CORGEA_TOKEN= - ./upload_checkmarx_report.py checkmarx_report.xml --project-name my-service --wait - -Only the Python standard library is required. -""" - -from __future__ import annotations - -import argparse -import json -import mimetypes -import os -import re -import sys -import time -import urllib.error -import urllib.parse -import urllib.request -import uuid -import xml.etree.ElementTree as ElementTree -from pathlib import Path -from typing import Any, Iterable - -DEFAULT_URL = "https://www.corgea.app" -API_BASE = "/api/v1" -ENGINE = "checkmarx" - -# Reports larger than this are streamed as `Upload-Offset`/`Upload-Length` -# chunks instead of a single body, matching the CLI's limits. -MAX_SINGLE_UPLOAD_BYTES = 50 * 1024 * 1024 -CHUNK_BYTES = 1024 * 1024 - -POLL_INTERVAL_SECONDS = 1 -ISSUE_PAGE_SIZE = 30 -URGENCY_ORDER = ("CR", "HI", "ME", "LO") - - -class CorgeaError(Exception): - """A request was rejected, or the report could not be parsed.""" - - -# -------------------------------------------------------------------------- -# Configuration -# -------------------------------------------------------------------------- - - -def read_cli_config() -> dict[str, str]: - """Read `~/.corgea/config.toml`, the file `corgea login` writes.""" - path = Path.home() / ".corgea" / "config.toml" - if not path.is_file(): - return {} - text = path.read_text(encoding="utf-8") - try: - import tomllib - - return {k: v for k, v in tomllib.loads(text).items() if isinstance(v, str)} - except ImportError: # Python < 3.11 - pairs = re.findall(r'^\s*(\w+)\s*=\s*"([^"]*)"\s*$', text, re.MULTILINE) - return dict(pairs) - - -def resolve_url(override: str | None) -> str: - config = read_cli_config() - url = override or os.environ.get("CORGEA_URL") or config.get("url") or DEFAULT_URL - return url.rstrip("/") - - -def resolve_token(override: str | None) -> str: - config = read_cli_config() - token = override or os.environ.get("CORGEA_TOKEN") or config.get("token") or "" - if not token: - raise CorgeaError( - "No Corgea token. Set CORGEA_TOKEN, pass --token, or run `corgea login `." - ) - return token - - -def auth_headers(token: str) -> dict[str, str]: - """A JWT goes in `Authorization`; an opaque token in `CORGEA-TOKEN`.""" - segments = token.split(".", 3) - if len(segments) == 3 and all(segments): - headers = {"Authorization": f"Bearer {token}"} - else: - headers = {"CORGEA-TOKEN": token} - headers["CORGEA-SOURCE"] = os.environ.get("CORGEA_SOURCE", "cli") - return headers - - -def sanitize_project_name(name: str) -> str: - return "".join(c if (c.isalnum() or c in "-_.") else "_" for c in name) - - -def determine_project_name(provided: str | None, root: Path) -> str: - """`--project-name`, else the git remote's repo name, else the directory.""" - if provided: - return sanitize_project_name(provided) - - git_config = root / ".git" / "config" - if git_config.is_file(): - match = re.search( - r'\[remote "origin"\][^\[]*?url\s*=\s*(\S+)', - git_config.read_text(encoding="utf-8", errors="replace"), - re.DOTALL, - ) - if match: - repo = match.group(1).removesuffix(".git").rstrip("/") - tail = re.split(r"[/:]", repo)[-1].strip() - if tail: - return sanitize_project_name(tail) - - return sanitize_project_name(root.resolve().name) - - -def ci_context(project: str, github_env: dict[str, str]) -> tuple[bool, str, str]: - """Corgea keys CI scans to `{repo}-{pr}` so a PR gets one project.""" - in_ci = "CI" in os.environ and "GITHUB_ACTIONS" in os.environ - platform = "github" if "GITHUB_ACTIONS" in os.environ else "unknown" - if in_ci: - project = f"{github_env['GITHUB_REPOSITORY']}-{github_env['GITHUB_PR']}" - return in_ci, platform, project - - -# -------------------------------------------------------------------------- -# Checkmarx report parsing -# -------------------------------------------------------------------------- - - -def _strip_leading_separator(path: str) -> str: - return path.lstrip("/\\") - - -def parse_checkmarx_xml(text: str) -> list[str]: - """`CxXMLResults`: paths live on `Result/@FileName` or `` nodes.""" - paths: list[str] = [] - for element in ElementTree.fromstring(text).iter(): - tag = element.tag.rpartition("}")[2] - if tag == "Result": - candidate = element.get("FileName", "") - elif tag == "FileName": - candidate = element.text or "" - else: - continue - candidate = _strip_leading_separator(candidate.strip()) - if candidate: - paths.append(candidate) - return paths - - -def _paths_from_nodes(nodes: Iterable[Any]) -> Iterable[str]: - # Checkmarx JSON prefixes every path with the scan root separator, which the - # CLI drops by removing the first character rather than stripping a set of - # separators. Mirror that so both produce identical `path` query values. - for node in nodes: - if isinstance(node, dict) and isinstance(node.get("fileName"), str): - yield node["fileName"][1:] - - -def parse_checkmarx_cli_json(data: dict[str, Any]) -> list[str]: - paths: list[str] = [] - for result in data.get("results") or []: - nodes = (result.get("data") or {}).get("nodes") or [] - paths.extend(_paths_from_nodes(nodes)) - return paths - - -def parse_checkmarx_web_json(data: dict[str, Any]) -> list[str]: - paths: list[str] = [] - sast = ((data.get("scanResults") or {}).get("sast") or {}) - for language in sast.get("languages") or []: - for query in language.get("queries") or []: - for vulnerability in query.get("vulnerabilities") or []: - paths.extend(_paths_from_nodes(vulnerability.get("nodes") or [])) - return paths - - -def parse_report(text: str) -> list[str]: - """Return the source paths a Checkmarx report references, in report order.""" - if text.startswith(" None: - self.base_url = base_url - self.headers = auth_headers(token) - self.timeout = timeout - - def _url(self, path: str, query: dict[str, str] | None = None) -> str: - url = f"{self.base_url}{API_BASE}{path}" - if query: - # `safe="/"` keeps repo-relative paths readable in the query string. - encoded = urllib.parse.urlencode( - query, quote_via=urllib.parse.quote, safe="/" - ) - url = f"{url}?{encoded}" - return url - - def request( - self, - method: str, - path: str, - query: dict[str, str] | None = None, - body: bytes | None = None, - headers: dict[str, str] | None = None, - ) -> tuple[int, dict[str, str], bytes]: - request = urllib.request.Request( - self._url(path, query), data=body, method=method - ) - for name, value in {**self.headers, **(headers or {})}.items(): - request.add_header(name, value) - try: - with urllib.request.urlopen(request, timeout=self.timeout) as response: - return response.status, dict(response.headers), response.read() - except urllib.error.HTTPError as error: - return error.code, dict(error.headers), error.read() - except urllib.error.URLError as error: - raise CorgeaError(f"{method} {path} failed: {error.reason}") from error - - def json_get(self, path: str, query: dict[str, str] | None = None) -> dict[str, Any]: - status, _, body = self.request("GET", path, query=query) - if status >= 400: - raise CorgeaError(f"GET {path} returned {status}: {body.decode(errors='replace')}") - return json.loads(body or b"{}") - - -def encode_multipart_file(field: str, path: Path) -> tuple[str, bytes]: - """Build a `multipart/form-data` body holding a single file part.""" - boundary = uuid.uuid4().hex - mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream" - head = ( - f"--{boundary}\r\n" - f'Content-Disposition: form-data; name="{field}"; filename="{path.name}"\r\n' - f"Content-Type: {mime}\r\n\r\n" - ).encode() - tail = f"\r\n--{boundary}--\r\n".encode() - return f"multipart/form-data; boundary={boundary}", head + path.read_bytes() + tail - - -# -------------------------------------------------------------------------- -# Upload steps -# -------------------------------------------------------------------------- - - -def verify_token(api: CorgeaApi) -> None: - body = api.json_get("/verify") - if body.get("status") != "ok": - raise CorgeaError(f"Token rejected by {api.base_url}: {body}") - - -def upload_source_files( - api: CorgeaApi, run_id: str, paths: list[str], root: Path, allow_missing: bool -) -> int: - """Upload each referenced file once, keyed to `run_id` by its repo-relative path.""" - uploaded = 0 - for path in dict.fromkeys(paths): # de-duplicate, keep report order - local = root / path - if not local.is_file(): - message = f"{path} is referenced by the report but missing under {root}" - if not allow_missing: - raise CorgeaError( - f"{message}. Run from the scanned source tree, pass --source-root, " - "or use --allow-missing-files to upload the report without it." - ) - print(f" warning: skipping {message}", file=sys.stderr) - continue - - content_type, body = encode_multipart_file("file", local) - status, _, response = api.request( - "POST", - "/code-upload", - query={"run_id": run_id, "path": path}, - body=body, - headers={"Content-Type": content_type}, - ) - if status >= 400: - raise CorgeaError( - f"code-upload of {path} returned {status}: {response.decode(errors='replace')}" - ) - uploaded += 1 - print(f" uploaded {path}") - return uploaded - - -def upload_report( - api: CorgeaApi, run_id: str, report: str, project: str, in_ci: bool, platform: str -) -> tuple[str, str | None]: - """POST the report and return `(scan_id, project_id)`.""" - query = { - "engine": ENGINE, - "run_id": run_id, - "project": project, - "ci": "true" if in_ci else "false", - "ci_platform": platform, - } - repo_data = os.environ.get("REPO_DATA", "") - if repo_data: - query["repo_data"] = repo_data - - payload = report.encode("utf-8") - # The endpoint is declared JSON even for Checkmarx XML; the `engine` - # query parameter is what selects the parser server side. - headers = {"Content-Type": "application/json"} - - if len(payload) <= MAX_SINGLE_UPLOAD_BYTES: - status, _, response = api.request( - "POST", "/scan-upload", query=query, body=payload, headers=headers - ) - else: - status, response_headers, response = _upload_report_in_chunks( - api, query, payload, headers - ) - del response_headers - - if status >= 400: - raise CorgeaError( - f"scan-upload returned {status}: {response.decode(errors='replace')}" - ) - - body = json.loads(response or b"{}") - scan_id = body.get("sast_scan_id") - if scan_id is None: - raise CorgeaError(f"scan-upload succeeded but returned no sast_scan_id: {body}") - project_id = body.get("project_id") - return str(scan_id), None if project_id is None else str(project_id) - - -def _upload_report_in_chunks( - api: CorgeaApi, query: dict[str, str], payload: bytes, headers: dict[str, str] -) -> tuple[int, dict[str, str], bytes]: - total = len(payload) - offset = 0 - status, response_headers, response = 0, {}, b"" - while offset < total: - chunk = payload[offset : offset + CHUNK_BYTES] - status, response_headers, response = api.request( - "POST", - "/scan-upload", - query=query, - body=chunk, - headers={ - **headers, - "Upload-Offset": str(offset), - "Upload-Length": str(total), - }, - ) - if status >= 400: - return status, response_headers, response - offset += len(chunk) - # A mismatch means chunks landed on different server instances; the - # assembled report would be corrupt, so stop rather than finish it. - acknowledged = response_headers.get("Upload-Offset") - if acknowledged is not None and acknowledged.isdigit(): - if int(acknowledged) != offset: - raise CorgeaError( - f"Upload offset mismatch: server has {acknowledged} bytes, expected {offset}." - ) - print(f" sent {offset}/{total} bytes") - return status, response_headers, response - - -def upload_git_config(api: CorgeaApi, run_id: str, root: Path) -> None: - """Optional: lets Corgea attach the scan to the right repo and branch.""" - git_config = root / ".git" / "config" - if not git_config.is_file(): - return - content_type, body = encode_multipart_file("file", git_config) - status, _, response = api.request( - "POST", - "/git-config-upload", - query={"run_id": run_id}, - body=body, - headers={"Content-Type": content_type}, - ) - if status >= 400: - print( - f" warning: git-config-upload returned {status}: " - f"{response.decode(errors='replace')}", - file=sys.stderr, - ) - - -def scan_url(base_url: str, scan_id: str, project_id: str | None, project: str) -> str: - if project_id: - return f"{base_url}/project/{project_id}/?scan_id={scan_id}" - return f"{base_url}/project/{urllib.parse.quote(project, safe='')}?scan_id={scan_id}" - - -def wait_for_scan(api: CorgeaApi, scan_id: str) -> None: - while True: - scan = api.json_get(f"/scan/{scan_id}") - status = scan.get("status", "") - if status == "complete": - return - print(f" scan status: {status or 'unknown'}") - time.sleep(POLL_INTERVAL_SECONDS) - - -def print_issue_summary(api: CorgeaApi, scan_id: str) -> None: - counts: dict[str, int] = {} - total = 0 - page = 1 - while True: - body = api.json_get( - f"/scan/{scan_id}/issues", - query={"page": str(page), "page_size": str(ISSUE_PAGE_SIZE)}, - ) - issues = body.get("issues") or [] - for issue in issues: - counts[issue.get("urgency", "?")] = counts.get(issue.get("urgency", "?"), 0) + 1 - total += 1 - if page >= int(body.get("total_pages") or 1) or not issues: - break - page += 1 - - print("\nScan Results:\n") - print(f"{'Classification':<20} | Count") - print(f"{'':-<20} | ") - for urgency in URGENCY_ORDER: - print(f"{urgency:<20} | {counts.get(urgency, 0)}") - print(f"{'':-<20} | ") - print(f"{'Total':<20} | {total}") - - -# -------------------------------------------------------------------------- -# Entry point -# -------------------------------------------------------------------------- - - -def parse_args(argv: list[str] | None = None) -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Upload a Checkmarx report to Corgea via the HTTP API.", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=( - "Examples:\n" - " %(prog)s checkmarx_report.xml\n" - " %(prog)s cx_results.json --project-name payments-api --wait\n" - ), - ) - parser.add_argument("report", help="Checkmarx report: CxXMLResults XML, or CLI/web JSON") - parser.add_argument( - "--project-name", - help="Corgea project. Defaults to the git repo name, else the source root's name.", - ) - parser.add_argument( - "--source-root", - default=".", - type=Path, - help="Directory the report's file paths are relative to (default: current directory).", - ) - parser.add_argument( - "--wait", - action="store_true", - help="Poll until the scan completes and print an issue summary.", - ) - parser.add_argument( - "--allow-missing-files", - action="store_true", - help="Warn instead of failing when a referenced source file is absent.", - ) - parser.add_argument("--url", help="Corgea base URL (default: $CORGEA_URL or the CLI config).") - parser.add_argument("--token", help="Corgea token (default: $CORGEA_TOKEN or the CLI config).") - return parser.parse_args(argv) - - -def main(argv: list[str] | None = None) -> int: - args = parse_args(argv) - - report_path = Path(args.report) - if not report_path.is_file(): - raise CorgeaError(f"Report not found: {report_path}") - # A BOM would break both the XML declaration check and json.loads. - report = report_path.read_text(encoding="utf-8-sig").strip() - - paths = parse_report(report) - if not paths: - print("No findings in the report, nothing to upload.") - return 0 - - root = args.source_root - api = CorgeaApi(resolve_url(args.url), resolve_token(args.token)) - run_id = str(uuid.uuid4()) - - verify_token(api) - - project = determine_project_name(args.project_name, root) - in_ci, platform, project = ci_context(project, dict(os.environ)) - - print(f"Uploading {len(set(paths))} source file(s) referenced by the report...") - if upload_source_files(api, run_id, paths, root, args.allow_missing_files) == 0: - raise CorgeaError("No source files were uploaded; Corgea cannot analyze the findings.") - - print(f"Uploading the report as project '{project}' (engine={ENGINE})...") - scan_id, project_id = upload_report(api, run_id, report, project, in_ci, platform) - upload_git_config(api, run_id, root) - - url = scan_url(api.base_url, scan_id, project_id, project) - print(f"\nScan {scan_id} created.\n{url}") - - if args.wait: - print("\nWaiting for the scan to complete...") - wait_for_scan(api, scan_id) - print_issue_summary(api, scan_id) - print(f"\n{url}") - - return 0 - - -if __name__ == "__main__": - try: - sys.exit(main()) - except CorgeaError as error: - print(f"error: {error}", file=sys.stderr) - sys.exit(1) - except KeyboardInterrupt: - sys.exit(130) From fcfa9f636a634a0be337c5b55cd0bfb2c79b2292 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 5 Aug 2026 08:15:41 +0000 Subject: [PATCH 3/3] Replace Checkmarx upload example with a simple Python script Two args: code_path and report_path. Creates the scan and exits without waiting. Stdlib only. Co-authored-by: Ibrahim Rahhal --- examples/README.md | 8 +- examples/upload_checkmarx.py | 194 +++++++++++++++++++++++++++++++++++ examples/upload_checkmarx.sh | 127 ----------------------- 3 files changed, 198 insertions(+), 131 deletions(-) create mode 100755 examples/upload_checkmarx.py delete mode 100755 examples/upload_checkmarx.sh diff --git a/examples/README.md b/examples/README.md index 1acca19..2ac36fd 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,13 +1,13 @@ # Examples -## `upload_checkmarx.sh` — upload a Checkmarx report +## `upload_checkmarx.py` — upload a Checkmarx report Creates a Corgea scan from a Checkmarx report. Same API flow as `corgea upload`, without waiting for the scan to finish. ```bash export CORGEA_TOKEN= -./upload_checkmarx.sh +./upload_checkmarx.py ``` | Arg | Meaning | @@ -22,7 +22,7 @@ Optional env vars: `CORGEA_URL` (default `https://www.corgea.app`), `PROJECT` ```bash export CORGEA_TOKEN= -./upload_checkmarx.sh ./checkmarx ./checkmarx/report.xml +./upload_checkmarx.py ./checkmarx ./checkmarx/report.xml ``` ``` @@ -34,7 +34,7 @@ Scan scan-abc-123 created. https://www.corgea.app/project/42/?scan_id=scan-abc-123 ``` -Requires `curl` and `python3` (stdlib only — used to parse the report). +Stdlib only — no `pip install`. ## `deps_skill.rs` diff --git a/examples/upload_checkmarx.py b/examples/upload_checkmarx.py new file mode 100755 index 0000000..e6aa8d8 --- /dev/null +++ b/examples/upload_checkmarx.py @@ -0,0 +1,194 @@ +#!/usr/bin/env python3 +"""Upload a Checkmarx report to Corgea (same flow as `corgea upload`). + +Usage: + export CORGEA_TOKEN= + ./upload_checkmarx.py + +Optional env: + CORGEA_URL Corgea base URL (default: https://www.corgea.app) + PROJECT Project name (default: basename of ) + +Creates the scan and prints the scan URL. Does not wait for it to finish. +Requires only the Python standard library. +""" + +from __future__ import annotations + +import json +import mimetypes +import os +import sys +import urllib.error +import urllib.parse +import urllib.request +import uuid +import xml.etree.ElementTree as ET +from pathlib import Path + +DEFAULT_URL = "https://www.corgea.app" + + +def die(msg: str, code: int = 1) -> None: + print(f"error: {msg}", file=sys.stderr) + sys.exit(code) + + +def auth_headers(token: str) -> dict[str, str]: + parts = token.split(".", 3) + if len(parts) == 3 and all(parts): + headers = {"Authorization": f"Bearer {token}"} + else: + headers = {"CORGEA-TOKEN": token} + headers["CORGEA-SOURCE"] = "cli" + return headers + + +def request( + method: str, + url: str, + headers: dict[str, str], + data: bytes | None = None, + extra_headers: dict[str, str] | None = None, +) -> bytes: + req = urllib.request.Request(url, data=data, method=method) + for k, v in {**headers, **(extra_headers or {})}.items(): + req.add_header(k, v) + try: + with urllib.request.urlopen(req, timeout=150) as resp: + return resp.read() + except urllib.error.HTTPError as e: + die(f"{method} {url} -> {e.code}: {e.read().decode(errors='replace')}") + except urllib.error.URLError as e: + die(f"{method} {url} failed: {e.reason}") + + +def multipart_file(path: Path) -> tuple[str, bytes]: + boundary = uuid.uuid4().hex + mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream" + head = ( + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="file"; filename="{path.name}"\r\n' + f"Content-Type: {mime}\r\n\r\n" + ).encode() + return f"multipart/form-data; boundary={boundary}", head + path.read_bytes() + f"\r\n--{boundary}--\r\n".encode() + + +def extract_paths(report: str) -> list[str]: + paths: list[str] = [] + if report.startswith(" None: + if len(sys.argv) != 3: + die(f"usage: {sys.argv[0]} ", code=2) + + code_path = Path(sys.argv[1]).resolve() + report_path = Path(sys.argv[2]).resolve() + if not report_path.is_file(): + die(f"report not found: {report_path}") + + token = os.environ.get("CORGEA_TOKEN") + if not token: + die("set CORGEA_TOKEN") + base = os.environ.get("CORGEA_URL", DEFAULT_URL).rstrip("/") + project = os.environ.get("PROJECT", code_path.name) + project = "".join(c if (c.isalnum() or c in "-_.") else "_" for c in project) + run_id = str(uuid.uuid4()) + api = f"{base}/api/v1" + headers = auth_headers(token) + + report = report_path.read_text(encoding="utf-8-sig").strip() + paths = extract_paths(report) + if not paths: + print("no findings in report, nothing to upload") + return + + request("GET", f"{api}/verify", headers) + + print(f"Uploading {len(paths)} source file(s) from {code_path}...") + for rel in paths: + file = code_path / rel + if not file.is_file(): + die(f"{rel} referenced by the report but missing under {code_path}") + ctype, body = multipart_file(file) + # Same as the CLI: path is passed raw in the query string. + url = f"{api}/code-upload?run_id={run_id}&path={rel}" + request("POST", url, headers, data=body, extra_headers={"Content-Type": ctype}) + print(f" {rel}") + + print(f"Uploading report as project '{project}'...") + qs = urllib.parse.urlencode( + { + "engine": "checkmarx", + "run_id": run_id, + "project": project, + "ci": "false", + "ci_platform": "unknown", + } + ) + resp = request( + "POST", + f"{api}/scan-upload?{qs}", + headers, + data=report.encode("utf-8"), + extra_headers={"Content-Type": "application/json"}, + ) + data = json.loads(resp) + scan_id = str(data["sast_scan_id"]) + project_id = data.get("project_id") + + git_config = code_path / ".git" / "config" + if git_config.is_file(): + ctype, body = multipart_file(git_config) + req = urllib.request.Request( + f"{api}/git-config-upload?run_id={run_id}", + data=body, + method="POST", + ) + for k, v in {**headers, "Content-Type": ctype}.items(): + req.add_header(k, v) + try: + urllib.request.urlopen(req, timeout=150).read() + except urllib.error.URLError: + pass + + print(f"Scan {scan_id} created.") + if project_id is not None: + print(f"{base}/project/{project_id}/?scan_id={scan_id}") + else: + print(f"{base}/project/{urllib.parse.quote(project, safe='')}?scan_id={scan_id}") + + +if __name__ == "__main__": + main() diff --git a/examples/upload_checkmarx.sh b/examples/upload_checkmarx.sh deleted file mode 100755 index 07fd9ec..0000000 --- a/examples/upload_checkmarx.sh +++ /dev/null @@ -1,127 +0,0 @@ -#!/usr/bin/env bash -# Upload a Checkmarx report to Corgea (same flow as `corgea upload`). -# -# Usage: -# export CORGEA_TOKEN= -# ./upload_checkmarx.sh -# -# Optional: -# CORGEA_URL Corgea base URL (default: https://www.corgea.app) -# PROJECT Project name (default: basename of ) -# -# Creates the scan and prints the scan URL. Does not wait for it to finish. -set -euo pipefail - -if [[ $# -ne 2 ]]; then - echo "usage: $0 " >&2 - exit 2 -fi - -CODE_PATH=$(cd "$1" && pwd) -REPORT_PATH=$(cd "$(dirname "$2")" && pwd)/$(basename "$2") -BASE_URL="${CORGEA_URL:-https://www.corgea.app}" -BASE_URL="${BASE_URL%/}" -TOKEN="${CORGEA_TOKEN:?set CORGEA_TOKEN}" -PROJECT="${PROJECT:-$(basename "$CODE_PATH")}" -PROJECT=$(printf '%s' "$PROJECT" | tr -c 'A-Za-z0-9._-' '_') -RUN_ID=$(python3 -c 'import uuid; print(uuid.uuid4())') -API="$BASE_URL/api/v1" - -if [[ ! -f "$REPORT_PATH" ]]; then - echo "error: report not found: $REPORT_PATH" >&2 - exit 1 -fi - -# Opaque tokens use CORGEA-TOKEN; JWTs (a.b.c) use Authorization: Bearer. -AUTH_ARGS=(-H "CORGEA-SOURCE: cli") -if [[ "$TOKEN" == *.*.* && "$TOKEN" != *..* ]]; then - AUTH_ARGS+=(-H "Authorization: Bearer $TOKEN") -else - AUTH_ARGS+=(-H "CORGEA-TOKEN: $TOKEN") -fi - -curl -fsS "${AUTH_ARGS[@]}" "$API/verify" >/dev/null - -# Extract repo-relative source paths from Checkmarx XML / CLI JSON / web JSON. -# Mirrors the CLI: strip a leading separator from each path named by the report. -mapfile -t PATHS < <(python3 - "$REPORT_PATH" <<'PY' -import json, sys, xml.etree.ElementTree as ET -from pathlib import Path - -text = Path(sys.argv[1]).read_text(encoding="utf-8-sig").strip() -paths = [] - -if text.startswith("&2 - exit 1 - fi - # Same as the CLI: path is passed raw in the query string. - curl -fsS "${AUTH_ARGS[@]}" \ - -F "file=@${file}" \ - "${API}/code-upload?run_id=${RUN_ID}&path=${rel}" \ - >/dev/null - echo " $rel" -done - -echo "Uploading report as project '$PROJECT'..." -RESPONSE=$(curl -fsS "${AUTH_ARGS[@]}" \ - -H "Content-Type: application/json" \ - --data-binary @"$REPORT_PATH" \ - "${API}/scan-upload?engine=checkmarx&run_id=${RUN_ID}&project=${PROJECT}&ci=false&ci_platform=unknown") - -SCAN_ID=$(printf '%s' "$RESPONSE" | python3 -c 'import json,sys; print(json.load(sys.stdin)["sast_scan_id"])') -PROJECT_ID=$(printf '%s' "$RESPONSE" | python3 -c 'import json,sys; print(json.load(sys.stdin).get("project_id") or "")') - -if [[ -f "$CODE_PATH/.git/config" ]]; then - curl -fsS "${AUTH_ARGS[@]}" \ - -F "file=@${CODE_PATH}/.git/config" \ - "${API}/git-config-upload?run_id=${RUN_ID}" >/dev/null || true -fi - -if [[ -n "$PROJECT_ID" ]]; then - echo "Scan $SCAN_ID created." - echo "${BASE_URL}/project/${PROJECT_ID}/?scan_id=${SCAN_ID}" -else - echo "Scan $SCAN_ID created." - echo "${BASE_URL}/project/${PROJECT}?scan_id=${SCAN_ID}" -fi