diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..2ac36fd --- /dev/null +++ b/examples/README.md @@ -0,0 +1,43 @@ +# Examples + +## `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.py +``` + +| 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`). + +### Try it + +```bash +export CORGEA_TOKEN= +./upload_checkmarx.py ./checkmarx ./checkmarx/report.xml +``` + +``` +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 +``` + +Stdlib only — no `pip install`. + +## `deps_skill.rs` + +```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.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()