Skip to content

Errors and strict mode

cloudg is built to keep going: a scanner that times out, a service that refuses a call or a report that will not parse is recorded, and the rest of the run carries on. These scripts show how to turn those records into exceptions when your job should stop instead.

Every exception cloudg raises on purpose derives from CloudGError. Exceptions has the class tree and when each one is raised.

A scanner that raises

Offline. With strict=True the first failure is raised at once. Without Checkov on PATH, that is a ScannerNotInstalledError.

strict_scanner.py
"""Run Checkov through cloudg and stop at the first problem."""

from cloudg.exceptions import ScannerError, ScannerNotInstalledError, ScannerTimeoutError
from cloudg.scanners.checkov import CheckovScanner

scanner = CheckovScanner(target_dir="infra/", timeout_seconds=600, strict=True)

try:
    findings = scanner.run()
except ScannerNotInstalledError as exc:
    print(f"{exc.executable} is missing. {exc.install_hint}")
except ScannerTimeoutError as exc:
    print(f"gave up on {exc.target} after {exc.timeout_seconds:g}s")
except ScannerError as exc:  # launch failures, unreadable output
    print(f"{exc.scanner} failed: {exc.detail}")
else:
    print(len(findings), "findings")

$ python strict_scanner.py
checkov is missing. pip install checkov

Decide after the run

raise_for_errors() raises what the last run recorded: one failure as it is, several as a ScannerFailed that lists each of them. Nothing recorded, nothing raised.

raise_for_errors.py
"""Scan several images, keep every result, then decide what counts as failure."""

from cloudg.exceptions import ScannerError, ScannerFailed
from cloudg.scanners.trivy import TrivyScanner

scanner = TrivyScanner(timeout_seconds=900)
findings = scanner.scan_images(["nginx:1.27", "python:3.12-slim"])

print(len(findings), "findings,", len(scanner.failures), "failures recorded")
for failure in scanner.failures:
    print(" -", type(failure).__name__, failure.detail)

try:
    scanner.raise_for_errors()
except ScannerFailed as exc:  # more than one failure: all of them, in order
    for failure in exc.failures:
        print("failed:", failure)
except ScannerError as exc:  # exactly one failure, raised as it is
    print("failed:", exc)

With a trivy on PATH that prints something other than JSON:

$ python raise_for_errors.py
0 findings, 2 failures recorded
 - ScannerOutputError output for nginx:1.27 is not JSON: Expecting value: line 1 column 1 (char 0)
 - ScannerOutputError output for python:3.12-slim is not JSON: Expecting value: line 1 column 1 (char 0)
failed: trivy: output for nginx:1.27 is not JSON: Expecting value: line 1 column 1 (char 0)
failed: trivy: output for python:3.12-slim is not JSON: Expecting value: line 1 column 1 (char 0)

Check credentials before a run

CloudGEngine.collect() records failures per service instead of raising. Build the session yourself first and a missing SDK or a refused role stops the job before anything runs.

engine_errors.py
"""Check credentials before a run, and sort what the run reports by type."""

import sys

from cloudg import CloudGConfig, CloudGEngine
from cloudg.credentials import build_aws_session
from cloudg.exceptions import (
    CloudGError,
    CredentialError,
    ProviderSDKMissingError,
    ScannerFailed,
)

config = CloudGConfig(providers=["aws"])
config.aws.profile = "audit"
config.aws.regions = ["eu-west-1"]

# The engine records collection failures instead of raising them, so a
# missing SDK or a refused role is easiest to catch before the run starts.
try:
    build_aws_session(config.aws, "eu-west-1")
except ProviderSDKMissingError as exc:
    sys.exit(f"{exc.missing} is not installed: pip install 'cloudg[{exc.extra}]'")
except CredentialError as exc:  # a role assumption failed, or no role to reach an account
    sys.exit(f"{exc.provider} credentials were refused: {exc.detail}")
except Exception as exc:  # SDK errors pass through untouched (botocore's ProfileNotFound, ...)
    sys.exit(f"{type(exc).__name__}: {exc}")


def on_error(phase: str, exc: Exception) -> None:
    if isinstance(exc, ScannerFailed):
        print(f"{phase}: kept {len(exc.findings)} partial findings ({exc})")
    elif isinstance(exc, CloudGError):
        print(f"{phase}: {type(exc).__name__}: {exc}")
    else:  # SDK and OS errors pass through untouched
        print(f"{phase}: unexpected {type(exc).__name__}: {exc}")


engine = CloudGEngine(config)
engine.on_error = on_error
result = engine.run_pipeline_sync(output_dir="./reports")
print(result.errors or "complete run")

$ pip uninstall -y boto3
$ python engine_errors.py
boto3 is not installed: pip install 'cloudg[aws]'

On the CLI, check the exit status instead. cloudg run exits 3 when collection failed for every target, and 0 with a warning when only some failed:

cloudg run -p aws --profile audit --regions eu-west-1
echo $?

Stop at the first failed phase

Drive the phases yourself and check the on_error records between them. Exit status 2 means a phase failed, 0 a complete run.

fail_fast.py
import asyncio
import sys

from cloudg import CloudGConfig, CloudGEngine

class PhaseFailed(RuntimeError):
    pass

async def main() -> int:
    engine = CloudGEngine(CloudGConfig(providers=["aws"]))
    failures: list[tuple[str, Exception]] = []
    engine.on_error = lambda phase, exc: failures.append((phase, exc))

    def checkpoint(step: str) -> None:
        if failures:
            phase, exc = failures[0]
            raise PhaseFailed(f"{phase} failed during {step}: {exc}")

    collection = await engine.collect()
    checkpoint("collection")
    broken = [c.to_summary() for c in collection.coverage if c.failed_services]
    if broken or not collection.assets:
        raise PhaseFailed(f"collection incomplete: {broken or 'no assets'}")

    findings = await engine.scan(collection.assets, collection.edges, output_dir="./reports")
    checkpoint("scanning")

    analysis = await engine.analyze(collection.assets, collection.edges, findings, "./reports")
    checkpoint("analysis")

    scan_result = engine.normalise_findings(
        findings + analysis.reachability_findings, assets=collection.assets
    )
    print(scan_result.summary)
    return 0

try:
    sys.exit(asyncio.run(main()))
except PhaseFailed as exc:
    print(f"cloudg: {exc}", file=sys.stderr)
    sys.exit(2)

Without credentials, every AWS service fails and the script stops after the collection with exit status 2, before any scanner starts.

Reject an unreadable report

Offline. Each wrapper's parse_report(path, strict=True) raises ReportParseError for a file that is not a report, instead of skipping it with a warning.

strict_parse_report.py
"""Fail an ingest job when a report file is truncated or not a report at all."""

import sys

from cloudg.exceptions import ReportParseError
from cloudg.scanners.checkov import CheckovScanner

path = sys.argv[1] if len(sys.argv) > 1 else "results_json.json"

try:
    findings = CheckovScanner.parse_report(path, strict=True)
except ReportParseError as exc:
    sys.exit(f"{exc.scanner} report {exc.path} is unreadable: {exc.detail}")

print(len(findings), "findings in", path)

$ echo "not json" > results_json.json
$ python strict_parse_report.py results_json.json
checkov report results_json.json is unreadable: output is not JSON: Expecting value: line 1 column 1 (char 0)