Exceptions

Every exception cloudg raises on purpose lives in cloudg.exceptions and derives from CloudGError, so one except CloudGError catches anything cloudg itself raised. Errors from the cloud SDKs and the operating system pass through untouched.

Module cloudg.exceptions · Also exported from cloudg · Last changed v0.6.1

The class tree¶

The narrower classes keep the built-in base they always had, so code written against an older release still works: ProviderSDKMissingError is still an ImportError, CredentialError is still a RuntimeError, and ConfigurationError is still a ValueError.

flowchart LR
  E["CloudGError"] --> CFG["ConfigurationError<br>(ValueError)"]
  E --> SDK["ProviderSDKMissingError<br>(ImportError)"]
  E --> COL["CollectionError<br>(RuntimeError)"]
  COL --> CRED["CredentialError"]
  COL --> ORG["OrganizationScopeError<br>(ValueError)"]
  E --> RES["ResilienceError"]
  E --> LOR["LiveOperationRejected"]
  E --> SCN["ScannerError"]
  SCN --> NI["ScannerNotInstalledError"]
  SCN --> TO["ScannerTimeoutError"]
  SCN --> EX["ScannerExecutionError"]
  SCN --> OUT["ScannerOutputError"]
  OUT --> RPE["ReportParseError"]
  SCN --> SF["ScannerFailed<br>(RuntimeError)"]

The built-in base each class keeps is in brackets.

ResilienceError and LiveOperationRejected live in cloudg.resilience, next to their subclasses (CircuitOpenError, DeadlineExceededError, RetryBudgetExhaustedError, CooldownActive, LiveOperationBusy, CallerQuotaExceeded, LiveOperationTimeout). cloudg.resilience covers them. Everything else imports from cloudg.exceptions or straight from cloudg:

from cloudg import CloudGError, ScannerError
from cloudg.exceptions import OrganizationScopeError, ScannerTimeoutError
from cloudg.resilience import CircuitOpenError, LiveOperationRejected

When each one is raised¶

Exception	Raised when	Useful attributes
`ConfigurationError`	A configuration value is missing, malformed or contradicts another. `cloudg.ingest.parse_report()` raises it for a tool name it does not support.
`ProviderSDKMissingError`	A provider SDK is not installed: boto3 for AWS, azure-identity for Azure, google-auth or google-cloud-asset for GCP. The message names the pip extra.	`provider`, `extra`, `missing`
`CollectionError`	Collecting from a provider failed as a whole, for example when GCP rejects both ListAssets and SearchAllResources for a scope.
`CredentialError`	Role assumption (`AssumeRole`, `AssumeRoleWithWebIdentity`) failed, no credentials resolved for `aws.role_arn`, or an account was asked for without a role to reach it.	`provider`, `detail`
`OrganizationScopeError`	An `--ou` (`aws.organization.include_ous`) filter matches nothing in the organization. cloudg stops instead of quietly mapping only the caller's account.
`ScannerNotInstalledError`	A scanner wrapper in strict mode finds no executable on `PATH`.	`executable`, `install_hint`
`ScannerTimeoutError`	A scanner ran longer than `timeout_seconds` and was killed.	`timeout_seconds`, `target`
`ScannerExecutionError`	A scanner process could not be started, or crashed.	`returncode`, `output`, `target`
`ScannerOutputError`	A scanner ran, but its output could not be read or parsed.	`path`
`ReportParseError`	A report handed to a wrapper's `parse_report(path, strict=True)` is unreadable, or a directory holds no report.	`path`
`ScannerFailed`	A scanner did not finish, as reported by `CloudGEngine.scan()` through `on_error`, or by `raise_for_errors()` when a wrapper recorded several failures.	`findings`, `errors`, `failures`

Every ScannerError also carries scanner (the wrapper's short name, such as "checkov"), detail (the reason in a few words) and target (the image, directory or file being scanned, when there is one). str(exc) is "<scanner>: <detail>".

Recorded failures and strict mode¶

The scanner wrappers record failures instead of raising them, so one broken scanner never stops a pipeline. Each wrapper keeps two lists:

errors: why the last run did not finish (timeouts, launch failures), as text;
failures: every ScannerError the wrapper met, including output it could not parse. Unparseable output does not stop a run, so it is in failures but not in errors.

There are two ways to turn those records into exceptions:

flowchart TD
  R["scanner.run()"] --> Q{"strict=True?"}
  Q -->|yes| RAISE["first failure raised at once"]
  Q -->|no| REC["failure appended to .failures"]
  REC --> D["run finishes, partial findings returned"]
  D --> RFE["scanner.raise_for_errors()"]
  RFE --> ONE["one failure: raised as it is"]
  RFE --> MANY["several: ScannerFailed with .failures"]
  RFE --> NONE["none: returns quietly"]

A missing executable is a warning by default and is not recorded as a failure, because "Trivy is not installed" is a setup choice rather than a broken run. Pass strict=True to get a ScannerNotInstalledError instead.

Examples¶

Each tab is a complete script. They run without cloud credentials: with no scanner installed, the strict scanner prints the install hint, and the report parser needs only a file.

strict=Trueraise_for_errors()Around CloudGEngineparse_report(path, strict=True)

The first failure is raised as soon as it happens, so the except clauses read like an ordinary call.

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

Without Checkov on PATH it prints:

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

Run everything first and keep the partial results, then decide what counts as a failure.

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, both images fail to parse, so the two failures come back together as one ScannerFailed:

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

CloudGEngine reports failures instead of raising them: collect() records them per service in the coverage records, and scanner failures reach the on_error hook as exception objects. A missing SDK or a refused role is easiest to catch before the run, by building the session yourself.

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

Without boto3 it stops at the first check:

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

build_aws_session() only contacts AWS when it has a role to assume, so a wrong key surfaces later, as failed services in result.coverage and lines in result.errors.

Each scanner wrapper has a parse_report class method for output that already exists. With strict=True, a file that is not a report raises ReportParseError instead of being skipped 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)

cloudg.ingest.parse_report(tool, path), the function behind cloudg ingest, has no strict flag: it logs and skips unreadable files, and raises ConfigurationError (a ValueError) for an unknown tool and FileNotFoundError for a missing path. Call the wrapper's own parse_report when a bad file should stop the job.

Catching everything cloudg raises¶

catch_all.py

from cloudg import CloudGError
from cloudg.ingest import parse_report

try:
    findings = parse_report("trivvy", "./trivy.json")
except CloudGError as exc:
    print("cloudg refused:", exc)
except FileNotFoundError as exc:
    print("no such report:", exc)

$ python catch_all.py
cloudg refused: Unsupported tool 'trivvy'. Supported: prowler, scoutsuite, checkov, trivy

The misspelt tool name raises ConfigurationError, which the first clause catches. A missing file raises the built-in FileNotFoundError, which is not a CloudGError: except CloudGError never hides an operating system error, an SDK exception or a bug in your own code.

Reference¶

Every exception cloudg raises on purpose, in one importable place.

All of them derive from CloudGError, so one except CloudGError catches anything cloudg itself raised, while SDK and OS errors keep passing through untouched. The narrower classes also keep the built-in base they always had (ValueError, RuntimeError, ImportError), so existing except ValueError code goes on working.

CloudGError
├── ConfigurationError            (ValueError)
├── ProviderSDKMissingError       (ImportError)
├── CollectionError               (RuntimeError)
│   ├── CredentialError
│   └── OrganizationScopeError    (ValueError)
├── ScannerError
│   ├── ScannerNotInstalledError
│   ├── ScannerTimeoutError
│   ├── ScannerExecutionError
│   ├── ScannerOutputError
│   │   └── ReportParseError
│   └── ScannerFailed             (RuntimeError)
├── ResilienceError               (cloudg.resilience)
└── LiveOperationRejected         (cloudg.resilience)

Scanner wrappers record failures instead of raising, so one broken scanner never stops a pipeline. To get exceptions instead, pass strict=True to the wrapper, or call raise_for_errors after a run:

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

scanner = CheckovScanner(target_dir="infra/", timeout_seconds=600)
findings = scanner.run()
try:
    scanner.raise_for_errors()
except ScannerTimeoutError as exc:
    print(f"{exc.scanner} gave up after {exc.timeout_seconds}s")
except ScannerError as exc:
    print(f"{exc.scanner} failed: {exc.detail}")

This module imports nothing from the rest of cloudg, so any module can import it without a cycle.

CloudGError ¶

Bases: Exception

Base of every exception cloudg raises on purpose.

ConfigurationError ¶

Bases: CloudGError, ValueError

A configuration value is missing, malformed or contradicts another.

ProviderSDKMissingError ¶

ProviderSDKMissingError(
    provider: str, extra: str | None = None, missing: str | None = None
)

Bases: CloudGError, ImportError

A provider's SDK is not installed.

provider is "aws", "azure" or "gcp" and extra the pip extra that installs it, so the message can say pip install cloudg[gcp].

Source code in cloudg/exceptions.py

def __init__(self, provider: str, extra: str | None = None, missing: str | None = None) -> None:
    self.provider = provider
    self.extra = extra or provider
    self.missing = missing
    what = f" ({missing})" if missing else ""
    super().__init__(
        f"the {provider} SDK is not installed{what}; run: pip install 'cloudg[{self.extra}]'"
    )

CollectionError ¶

Bases: CloudGError, RuntimeError

Collecting inventory from a cloud provider failed.

CredentialError ¶

CredentialError(provider: str, detail: str)

Bases: CollectionError

No usable credentials were found for a provider, or they were refused.

Source code in cloudg/exceptions.py

def __init__(self, provider: str, detail: str) -> None:
    self.provider = provider
    self.detail = detail
    super().__init__(f"{provider}: {detail}")

OrganizationScopeError ¶

Bases: CollectionError, ValueError

The requested organization scope (an OU filter) names nothing in the org.

Raised instead of quietly mapping only the caller's account.

ScannerError ¶

ScannerError(scanner: str, detail: str, *, target: str | None = None)

Bases: CloudGError

A security scanner could not produce a complete result.

scanner is the wrapper's short name ("prowler", "checkov", ...), detail the reason in a few words and target what was being scanned (an image, a directory, a report file), when there is one.

Source code in cloudg/exceptions.py

def __init__(self, scanner: str, detail: str, *, target: str | None = None) -> None:
    self.scanner = scanner
    self.detail = detail
    self.target = target
    super().__init__(f"{scanner}: {detail}")

ScannerNotInstalledError ¶

ScannerNotInstalledError(scanner: str, executable: str, install_hint: str = '')

Bases: ScannerError

The scanner's executable is not on PATH.

Source code in cloudg/exceptions.py

def __init__(self, scanner: str, executable: str, install_hint: str = "") -> None:
    self.executable = executable
    self.install_hint = install_hint
    detail = f"{executable!r} is not on PATH"
    if install_hint:
        detail += f"; install with: {install_hint}"
    super().__init__(scanner, detail)

ScannerTimeoutError ¶

ScannerTimeoutError(scanner: str, timeout_seconds: float, *, target: str | None = None)

Bases: ScannerError

The scanner ran longer than its timeout_seconds and was killed.

Source code in cloudg/exceptions.py

def __init__(self, scanner: str, timeout_seconds: float, *, target: str | None = None) -> None:
    self.timeout_seconds = timeout_seconds
    detail = f"timed out after {timeout_seconds:g}s"
    if target:
        detail += f" scanning {target}"
    super().__init__(scanner, detail, target=target)

ScannerExecutionError ¶

ScannerExecutionError(
    scanner: str,
    detail: str,
    *,
    target: str | None = None,
    returncode: int | None = None,
    output: str = "",
)

Bases: ScannerError

The scanner process could not be started, or crashed.

returncode is the exit status when the process ran; output the tail of what it printed, when that was captured.

Source code in cloudg/exceptions.py

def __init__(
    self,
    scanner: str,
    detail: str,
    *,
    target: str | None = None,
    returncode: int | None = None,
    output: str = "",
) -> None:
    self.returncode = returncode
    self.output = output
    super().__init__(scanner, detail, target=target)

ScannerOutputError ¶

ScannerOutputError(
    scanner: str, detail: str, *, target: str | None = None, path: str | None = None
)

Bases: ScannerError

The scanner ran, but its output could not be read or parsed.

Source code in cloudg/exceptions.py

def __init__(
    self, scanner: str, detail: str, *, target: str | None = None, path: str | None = None
) -> None:
    self.path = path
    super().__init__(scanner, detail, target=target or path)

ReportParseError ¶

ReportParseError(
    scanner: str, detail: str, *, target: str | None = None, path: str | None = None
)

Bases: ScannerOutputError

An existing report handed to parse_report or cloudg ingest is unreadable.

Source code in cloudg/exceptions.py

def __init__(
    self, scanner: str, detail: str, *, target: str | None = None, path: str | None = None
) -> None:
    self.path = path
    super().__init__(scanner, detail, target=target or path)

ScannerFailed ¶

ScannerFailed(
    scanner: str,
    errors: list[str],
    findings: list[Any],
    failures: list[ScannerError] | None = None,
)

Bases: ScannerError, RuntimeError

A scanner started but did not finish (timeout, launch failure).

findings holds whatever it produced before that (for example the images Trivy did scan), so callers can keep partial results. errors are the recorded reasons as text and failures the matching ScannerError objects, when the wrapper recorded them.

Source code in cloudg/exceptions.py

def __init__(
    self,
    scanner: str,
    errors: list[str],
    findings: list[Any],
    failures: list[ScannerError] | None = None,
) -> None:
    self.errors = list(errors)
    self.findings = findings
    self.failures = list(failures or [])
    ScannerError.__init__(self, scanner, "; ".join(self.errors))
    # Keep the message format callers already match on: "<scanner> <reasons>"
    self.args = (f"{scanner} " + "; ".join(self.errors),)

cloudg.resilience

The rate limiter, circuit breakers and the errors they raise.
parse_report

Turning existing scanner output into findings.
Event hooks

on_error and the other callbacks on CloudGEngine.
Running scanners

What each scanner needs and how failures show up on the CLI.

Exceptions

The class tree¶

When each one is raised¶

Recorded failures and strict mode¶

Examples¶

Catching everything cloudg raises¶

Reference¶

CloudGError ¶

ConfigurationError ¶

ProviderSDKMissingError ¶

CollectionError ¶

CredentialError ¶

OrganizationScopeError ¶

ScannerError ¶

ScannerNotInstalledError ¶

ScannerTimeoutError ¶

ScannerExecutionError ¶

ScannerOutputError ¶

ReportParseError ¶

ScannerFailed ¶

Related¶