Skip to content

Ingest and report

Already ran the scanners somewhere else? Feed cloudg their native output. No cloud credentials and no scanner binaries are needed, and the core package is enough.

Every script on this page is offline. Run make_sample_reports.py in an empty directory first, then copy its Trivy image report to the name the scripts read:

python make_sample_reports.py
cp trivy-image.json trivy.json

Merge reports and render them

cloudg ingest --prowler ./prowler-output/ --checkov ./results_json.json --trivy ./trivy.json -o ./reports
ingest_only.py
from cloudg import CloudGConfig, CloudGEngine

engine = CloudGEngine(CloudGConfig())
engine.on_phase_start = lambda phase: print("phase:", phase)

result = engine.run_from_reports_sync(
    {
        "prowler": ["./prowler-output/"],
        "checkov": ["./results_json.json"],
        "trivy": ["./trivy.json"],
    },
    output_dir="./reports",
)

print(result.total_findings, result.severity_breakdown)
print(result.report_paths)
for f in result.findings:
    print(f.severity.value, f.risk_score, f.source_tool, f.title)

docker run --rm \
  -v "$PWD:/app/input:ro" \
  -v "$PWD/reports:/app/reports" \
  ghcr.io/morpheuslord/cloudg:latest \
  ingest --prowler /app/input/prowler-output/ --checkov /app/input/results_json.json --trivy /app/input/trivy.json

$ python ingest_only.py
phase: ingest
phase: normalisation
phase: reporting
3 {'CRITICAL': 1, 'HIGH': 2}
{'json': PosixPath('reports/findings.json'), 'html': PosixPath('reports/report.html')}
CRITICAL 9.7 trivy [Trivy] CVE-2024-0001: openssl (alpine)
HIGH 7.5 prowler S3 bucket default encryption
HIGH 7.5 checkov [Checkov/terraform] Ensure S3 bucket has server-side encryption enabled

reports/report.html is the page shown in the sample report, filled with these three findings.

Parse first, decide later

parse_report() reads one report into Finding objects without normalising anything, so you can filter or enrich them before the merge. ingest_reports() does several at once and skips files it cannot read.

parse_first.py
import logging

from cloudg.ingest import SUPPORTED_TOOLS, ingest_reports, parse_report
from cloudg.normaliser import FindingsNormaliser

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")

print(SUPPORTED_TOOLS)
findings = parse_report("prowler", "./prowler-output/")
findings += parse_report("checkov", "./results_json.json")
print(len(findings), "findings from two reports")

# A typo'd tool name or a missing path is skipped by ingest_reports, not raised
more = ingest_reports({"trivy": ["./trivy.json", "./missing.json"], "trivvy": ["./trivy.json"]})
print(len(more), "findings from the batch")

try:
    parse_report("grype", "./trivy.json")
except ValueError as exc:
    print("ValueError:", exc)

scan_result = FindingsNormaliser().normalise(findings, more)
print(scan_result.summary["severity_breakdown"], len(scan_result.compliance), "control results")

$ python parse_first.py
INFO cloudg.scanners.prowler: Parsed 1 findings from Prowler (1 passed checks)
INFO cloudg.scanners.checkov: Parsed 1 findings from Checkov
INFO cloudg.scanners.trivy: Parsed 1 findings from Trivy for myrepo/app:latest
INFO cloudg.ingest: [trivy] 1 findings ingested from ./trivy.json
ERROR cloudg.ingest: Skipping trivy report ./missing.json: Report path does not exist: ./missing.json
ERROR cloudg.ingest: Skipping trivvy report ./trivy.json: Unsupported tool 'trivvy'. Supported: prowler, scoutsuite, checkov, trivy
INFO cloudg.normaliser: Normalising 3 total findings from 2 sources
INFO cloudg.normaliser: After deduplication: 3 findings
('prowler', 'scoutsuite', 'checkov', 'trivy')
2 findings from two reports
1 findings from the batch
ValueError: Unsupported tool 'grype'. Supported: prowler, scoutsuite, checkov, trivy
{'CRITICAL': 1, 'HIGH': 2} 31 control results

The unknown tool raises ConfigurationError, which is a ValueError, so the except ValueError in the script catches it. Exceptions has the strict variant that raises on an unreadable file as well.

Merge findings from your own tools

FindingsNormaliser takes lists of Finding objects from any source. Findings from two scanners about the same check on the same resource are merged into one, with both tools listed.

normalise_demo.py
from cloudg.normaliser import FindingsNormaliser
from cloudg.schema.models import Finding, Severity

BUCKET = "arn:aws:s3:::orders-archive"

def finding(tool, check, title, severity, **extra):
    return Finding(resource_id=BUCKET, resource_arn=BUCKET, source_tool=tool,
                   source_finding_id=check, title=title, description=title,
                   severity=severity, **extra)

prowler = [
    finding("prowler", "prowler-aws-s3_bucket_default_encryption-123456789012-eu-west-1-orders-archive",
            "S3 bucket default encryption", Severity.MEDIUM),
    # the same check reported twice (two output formats, say): a within-scanner duplicate
    finding("prowler", "prowler-aws-s3_bucket_default_encryption-123456789012-eu-west-1-orders-archive",
            "S3 bucket default encryption", Severity.MEDIUM),
]
trivy = [
    # equivalent check (AVD-AWS-0088), same resource, same normalised title
    finding("trivy", "AVD-AWS-0088", "[Trivy] S3 bucket default encryption", Severity.HIGH),
    # a CVE with a CVSS score of 9.8 reported as MEDIUM
    finding("trivy", "CVE-2024-0001", "openssl: buffer overflow", Severity.MEDIUM, cvss_score=9.8),
]
untagged = [finding("custom", None, "CloudTrail logging disabled in eu-west-1", Severity.LOW)]

result = FindingsNormaliser().normalise(prowler, trivy, untagged)
for f in result.findings:
    print(f"{f.severity.value:8} {f.risk_score:4} {f.source_tool:16} {f.title} {f.compliance_frameworks[:4]}")
print(len(result.compliance), "compliance results")

$ python normalise_demo.py
CRITICAL  9.7 trivy            openssl: buffer overflow ['HIPAA', 'NIST-800-53', 'PCI-DSS', 'SOC2']
HIGH      7.5 trivy, prowler   [Trivy] S3 bucket default encryption ['GDPR-AWS', 'HIPAA-AWS', 'ISO27001-AWS', 'MITRE-ATTACK-AWS']
LOW       2.5 custom           CloudTrail logging disabled in eu-west-1 ['GDPR', 'HIPAA', 'ISO-27001', 'NIST-800-53']
38 compliance results

Gate a CI job on severity

Write a summary the pipeline can archive, and exit non-zero when anything critical or high turns up.

save_result.py
import json
import sys
from pathlib import Path

from cloudg import CloudGConfig, CloudGEngine

result = CloudGEngine(CloudGConfig()).run_from_reports_sync(
    {"checkov": ["./results_json.json"], "trivy": ["./trivy.json"]},
    output_dir="./reports",
)

# Headline numbers: plain JSON types only, safe to json.dumps
summary = result.to_summary()
print(json.dumps(summary, indent=2))

# Everything else: dump the pydantic models and stringify the paths yourself
payload = {
    "summary": summary,
    "findings": [f.model_dump(mode="json") for f in result.findings],
    "compliance": [c.model_dump(mode="json") for c in result.scan_result.compliance]
    if result.scan_result
    else [],
    "report_paths": {kind: str(path) for kind, path in result.report_paths.items()},
}
Path("./reports/run-summary.json").write_text(json.dumps(payload, indent=2))

breakdown = result.severity_breakdown            # only severities that occur
blocking = breakdown.get("CRITICAL", 0) + breakdown.get("HIGH", 0)
if blocking:
    print(f"{blocking} critical/high findings, failing the build", file=sys.stderr)
    sys.exit(1)

$ python save_result.py
{
  "total_assets": 0,
  "total_findings": 2,
  "severity_breakdown": {
    "CRITICAL": 1,
    "HIGH": 1
  },
  "providers_scanned": [],
  "regions_scanned": {},
  "graph_nodes": 0,
  "graph_edges": 0,
  "ontology_triples": 0,
  "attack_paths_count": 0,
  "duration_ms": 1219,
  "errors": []
}
2 critical/high findings, failing the build
$ echo $?
1
.github/workflows/scan-gate.yml
jobs:
  gate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install cloudg
      # results_json.json and trivy.json come from earlier steps or artifacts
      - run: python save_result.py
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: cloudg-reports
          path: reports/

CI and automation has complete workflows for GitHub Actions, GitLab and scheduled scans.

Re-render a report, or export the raw findings

cloudg report -i ./reports/findings.json -o ./rendered --format all
cloudg report -i ./reports/raw-findings.json -o ./rendered
export_findings.py
import json
from pathlib import Path

from cloudg.ingest import ingest_reports

findings = ingest_reports(
    {
        "prowler": ["./prowler-output/"],
        "checkov": ["./results_json.json"],
        "trivy": ["./trivy.json"],
    }
)
out = Path("./reports/raw-findings.json")
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps([f.model_dump(mode="json") for f in findings], indent=2))
print(f"wrote {len(findings)} findings to {out}")

$ python export_findings.py
wrote 3 findings to reports/raw-findings.json

cloudg report reads either file: a findings.json keeps its scan ID, compliance results and edges, and a raw-findings.json is normalised first.