Add Python example for uploading a Checkmarx report

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=<your token>
+./upload_checkmarx.py <code_path> <report_path>
+```
+
+| 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=<your 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]
+```

examples/checkmarx/report.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="utf-8"?>
+<CxXMLResults InitiatorName="svc-corgea" Owner="security" ScanId="1043271" ProjectName="checkmarx-demo" TeamFullPathOnReportDate="CxServer" DeepLink="https://checkmarx.example.com/CxWebClient/ViewerMain.aspx?scanid=1043271&amp;projectid=42" ScanStart="Monday, July 27, 2026 9:12:03 AM" Preset="Checkmarx Default" ScanTime="00h:04m:11s" LinesOfCodeScanned="48" FilesScanned="2" ReportCreationTime="Monday, July 27, 2026 9:18:44 AM" Team="CxServer" CheckmarxVersion="9.5.0" ScanType="Full" SourceOrigin="LocalPath" Visibility="Public">
+  <Query id="594" cweId="89" name="SQL_Injection" group="Python_High_Risk" Severity="High" QueryPath="Python\Cx\Python High Risk\SQL Injection Version:0" Language="Python" LanguageHash="1806663762" LanguageChangeDate="2026-01-14T00:00:00.0000000" SeverityIndex="3">
+    <Result NodeId="10432712" FileName="/src/db.py" Status="New" Line="14" Column="24" FalsePositive="False" Severity="High" AssignToUser="" state="0" Remark="" DeepLink="https://checkmarx.example.com/CxWebClient/ViewerMain.aspx?scanid=1043271&amp;projectid=42&amp;pathid=1" SeverityIndex="3">
+      <Path ResultId="1043271" PathId="1" SimilarityId="-1729315482">
+        <PathNode>
+          <FileName>/src/login.py</FileName>
+          <Line>6</Line>
+          <Column>18</Column>
+          <NodeId>1</NodeId>
+          <Name>username</Name>
+          <Type>ParamDecl</Type>
+          <Length>8</Length>
+          <Snippet>
+            <Line>
+              <Number>6</Number>
+              <Code>def authenticate(username, password):</Code>
+            </Line>
+          </Snippet>
+        </PathNode>
+        <PathNode>
+          <FileName>/src/db.py</FileName>
+          <Line>14</Line>
+          <Column>24</Column>
+          <NodeId>2</NodeId>
+          <Name>query</Name>
+          <Type>StringLiteral</Type>
+          <Length>5</Length>
+          <Snippet>
+            <Line>
+              <Number>14</Number>
+              <Code>    query = "SELECT id, role FROM users WHERE name = '" + username + "'"</Code>
+            </Line>
+          </Snippet>
+        </PathNode>
+      </Path>
+    </Result>
+  </Query>
+  <Query id="612" cweId="798" name="Use_Of_Hardcoded_Password" group="Python_Best_Coding_Practice" Severity="Medium" QueryPath="Python\Cx\Python Best Coding Practice\Use Of Hardcoded Password Version:0" Language="Python" LanguageHash="1806663762" LanguageChangeDate="2026-01-14T00:00:00.0000000" SeverityIndex="2">
+    <Result NodeId="10432713" FileName="/src/db.py" Status="New" Line="6" Column="16" FalsePositive="False" Severity="Medium" AssignToUser="" state="0" Remark="" DeepLink="https://checkmarx.example.com/CxWebClient/ViewerMain.aspx?scanid=1043271&amp;projectid=42&amp;pathid=2" SeverityIndex="2">
+      <Path ResultId="1043271" PathId="2" SimilarityId="1180231477">
+        <PathNode>
+          <FileName>/src/db.py</FileName>
+          <Line>6</Line>
+          <Column>16</Column>
+          <NodeId>1</NodeId>
+          <Name>DB_PASSWORD</Name>
+          <Type>StringLiteral</Type>
+          <Length>11</Length>
+          <Snippet>
+            <Line>
+              <Number>6</Number>
+              <Code>DB_PASSWORD = "s3cr3t-admin-pw"</Code>
+            </Line>
+          </Snippet>
+        </PathNode>
+      </Path>
+    </Result>
+  </Query>
+</CxXMLResults>

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()

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]}

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=<token>
+    ./upload_checkmarx.py <code_path> <report_path>
+
+Optional env:
+    CORGEA_URL   Corgea base URL (default: https://www.corgea.app)
+    PROJECT      Project name (default: basename of <code_path>)
+
+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("<?xml") and "<CxXMLResults" in report:
+        for el in ET.fromstring(report).iter():
+            tag = el.tag.rpartition("}")[2]
+            if tag == "Result" and el.get("FileName"):
+                paths.append(el.get("FileName", "").lstrip("/\\"))
+            elif tag == "FileName" and (el.text or "").strip():
+                paths.append(el.text.strip().lstrip("/\\"))
+    else:
+        data = json.loads(report)
+        if {"totalCount", "results", "scanID"} <= data.keys():
+            for r in data.get("results") or []:
+                for n in ((r.get("data") or {}).get("nodes") or []):
+                    if isinstance(n.get("fileName"), str):
+                        paths.append(n["fileName"][1:])
+        elif {"scanResults", "reportId"} <= data.keys():
+            for lang in (((data.get("scanResults") or {}).get("sast") or {}).get("languages") or []):
+                for q in lang.get("queries") or []:
+                    for v in q.get("vulnerabilities") or []:
+                        for n in v.get("nodes") or []:
+                            if isinstance(n.get("fileName"), str):
+                                paths.append(n["fileName"][1:])
+        else:
+            die("unrecognized Checkmarx report format")
+
+    seen: set[str] = set()
+    out: list[str] = []
+    for p in paths:
+        if p and p not in seen:
+            seen.add(p)
+            out.append(p)
+    return out
+
+
+def main() -> None:
+    if len(sys.argv) != 3:
+        die(f"usage: {sys.argv[0]} <code_path> <report_path>", 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()