Python API Reference

automated_security_helper.interactions.run_ash_scan

ScanOptions

Bases: BaseModel

All parameters for a single run_ash_scan invocation.

Source code in automated_security_helper/interactions/run_ash_scan.py

class ScanOptions(BaseModel):
    """All parameters for a single run_ash_scan invocation."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    # Core paths
    source_dir: Path
    output_dir: Path

    # Workspace mode. When set, source_dir is the workspace ROOT and each project
    # is scanned with its own source_dir inside it. None means single-directory
    # mode, which is the only shape the rest of this module handles.
    workspace_plan: Optional[WorkspacePlan] = None
    allow_missing_projects: bool = False

    # General scan options
    config: Optional[str] = None
    config_overrides: Optional[List[str]] = Field(default_factory=list)
    # A check every `extends` base of the config must also pass, applied by every
    # config read below. The MCP server passes the calling session's allowed
    # config roots (cli/mcp/sandbox.config_base_gate); None keeps the CLI rule.
    config_base_gate: Optional[Callable[[Path], bool]] = Field(
        default=None, exclude=True
    )
    # The config was written by the caller (an MCP client's upload), so its sandbox
    # settings are restrict-only; trusted_config_path is the operator's config the
    # grants come from instead. Both go to resolve_config, in local mode only.
    untrusted_config: bool = False
    trusted_config_path: Optional[str] = None
    offline: bool = False
    strategy: ExecutionStrategy = ExecutionStrategy.PARALLEL
    scanners: Optional[List[str]] = Field(default_factory=list)
    excluded_scanners: Optional[List[str]] = Field(default_factory=list)
    progress: bool = True
    output_formats: Optional[List[ExportFormat]] = Field(default_factory=list)
    cleanup: bool = False
    phases: Optional[List[ExecutionPhase]] = Field(
        default_factory=lambda: [
            ExecutionPhase.CONVERT,
            ExecutionPhase.SCAN,
            ExecutionPhase.REPORT,
        ]
    )
    inspect: bool = False
    existing_results: Optional[str] = None
    python_based_plugins_only: bool = False
    quiet: bool = False
    simple: bool = False
    verbose: bool = False
    debug: bool = False
    color: bool = True
    fail_on_findings: Optional[bool] = None
    # Fail when a selected scanner did not complete. None means "not set on the
    # command line", which defers to the config file and then to off. Kept
    # separate from fail_on_findings rather than folded into it because the two
    # answer different questions: whether anything was found, and whether the
    # scanners that were supposed to look actually ran.
    fail_on_incomplete_scanners: Optional[bool] = None
    ignore_suppressions: bool = False
    min_severity: str = "low"
    changed_files_only: bool = False
    base_ref: str = "origin/main"
    # One shard of a split scan, or both None for an ordinary whole scan. Left
    # unvalidated here on purpose: ScanPhase._execute_phase is the single place
    # that refuses an unusable pair, so a second rule in this model could drift
    # from it and start rejecting a selection the scan phase would have accepted
    # (or the reverse). The CLI validates early for the operator's benefit; that
    # is presentation, not the contract.
    shard_index: Optional[int] = None
    shard_count: Optional[int] = None
    mode: RunMode = RunMode.local
    show_summary: bool = True
    log_level: AshLogLevel = AshLogLevel.INFO

    # Container-specific
    build: bool = True
    run: bool = True
    force: bool = False
    oci_runner: Optional[str] = None
    build_target: Optional[BuildTarget] = None
    offline_semgrep_rulesets: str = "p/ci"
    container_uid: Optional[str] = None
    container_gid: Optional[str] = None
    ash_revision_to_install: Optional[str] = None
    custom_containerfile: Optional[str] = None
    custom_build_arg: Optional[List[str]] = Field(default_factory=list)
    ash_plugin_modules: Optional[List[str]] = Field(default_factory=list)
    container_network: str = "bridge"

    @field_validator("source_dir", "output_dir", mode="before")
    @classmethod
    def _coerce_absolute_path(cls, v):
        return Path(v).absolute()

    @field_validator(
        "scanners",
        "excluded_scanners",
        "output_formats",
        "config_overrides",
        "custom_build_arg",
        "ash_plugin_modules",
        mode="before",
    )
    @classmethod
    def _none_to_empty_list(cls, v):
        return v if v is not None else []

    @field_validator("phases", mode="before")
    @classmethod
    def _phases_default(cls, v):
        if v is None:
            return [ExecutionPhase.CONVERT, ExecutionPhase.SCAN, ExecutionPhase.REPORT]
        return v

ScanIncompleteness dataclass

Every reason a finished scan exits 1 for coverage rather than for a crash.

One field per arm of the exit-1 half of _compute_exit_code, which reads its verdict from this object rather than from its own calls to the primitives. That is what makes it safe for anything else to act on: the MCP runner reports a scan incomplete instead of failed from this object, and if the two read the question separately, the status a client sees and the exit code the CLI returns could disagree on the same results.

Truthy when any field is set. Empty for results is None, which also exits 1 but means no results exist. That is a crash, and it does not belong in here.

Source code in automated_security_helper/interactions/run_ash_scan.py

@dataclass(frozen=True)
class ScanIncompleteness:
    """Every reason a finished scan exits 1 for coverage rather than for a crash.

    One field per arm of the exit-1 half of ``_compute_exit_code``, which reads its
    verdict from this object rather than from its own calls to the primitives. That
    is what makes it safe for anything else to act on: the MCP runner reports a scan
    ``incomplete`` instead of ``failed`` from this object, and if the two read the
    question separately, the status a client sees and the exit code the CLI returns
    could disagree on the same results.

    Truthy when any field is set. Empty for ``results is None``, which also exits
    1 but means no results exist. That is a crash, and it does not belong in here.
    """

    #: ``incomplete_scanners`` rows: (name, display status).
    scanners: Tuple[Tuple[str, str], ...] = ()
    #: Every scanner's recorded status, the raw token rather than the display
    #: string, so a payload can say MISSING or ERROR without parsing prose.
    observed: Tuple[Tuple[str, str], ...] = ()
    #: The set-level arm: scanners were expected or recorded, and none reached a
    #: verdict.
    no_scanner_ran: bool = False
    #: ``incomplete_converters`` rows: (name, reason).
    converters: Tuple[Tuple[str, str], ...] = ()
    #: ``unevaluated_rules``. Not behind the completeness gate.
    unevaluated_rules: Tuple[str, ...] = ()
    #: Content databases past their declared bound under ``content_db_staleness:
    #: fail``. Not behind the completeness gate either; its opt-out is
    #: ``--allow-stale-content-db`` or ``content_db_staleness: warn``, under which
    #: nothing is recorded here.
    stale_content_databases: Tuple[ContentDbAgeRecord, ...] = ()

    def __bool__(self) -> bool:
        return bool(
            self.scanners
            or self.no_scanner_ran
            or self.converters
            or self.unevaluated_rules
            or self.stale_content_databases
        )

    def to_payload(self) -> Dict[str, object]:
        """The structured form an MCP client reads. JSON-serializable."""
        raw_status = dict(self.observed)
        scanners = []
        for name, detail in self.scanners:
            status = raw_status.get(name, "")
            scanners.append(
                {
                    "scanner": name,
                    "status": status,
                    "reason": incomplete_scanner_reason(status),
                    "detail": detail,
                }
            )
        return {
            "incomplete_scanners": scanners,
            "no_scanner_ran": self.no_scanner_ran,
            "incomplete_converters": [
                {"converter": name, "reason": reason}
                for name, reason in self.converters
            ],
            "unevaluated_rules": list(self.unevaluated_rules),
            "stale_content_databases": [
                record.to_dict() for record in self.stale_content_databases
            ],
        }

to_payload()

The structured form an MCP client reads. JSON-serializable.

Source code in automated_security_helper/interactions/run_ash_scan.py

def to_payload(self) -> Dict[str, object]:
    """The structured form an MCP client reads. JSON-serializable."""
    raw_status = dict(self.observed)
    scanners = []
    for name, detail in self.scanners:
        status = raw_status.get(name, "")
        scanners.append(
            {
                "scanner": name,
                "status": status,
                "reason": incomplete_scanner_reason(status),
                "detail": detail,
            }
        )
    return {
        "incomplete_scanners": scanners,
        "no_scanner_ran": self.no_scanner_ran,
        "incomplete_converters": [
            {"converter": name, "reason": reason}
            for name, reason in self.converters
        ],
        "unevaluated_rules": list(self.unevaluated_rules),
        "stale_content_databases": [
            record.to_dict() for record in self.stale_content_databases
        ],
    }

ScanIncompleteExit

Bases: SystemExit

sys.exit(1) for a scan that finished with partial coverage.

A SystemExit with code 1, so the CLI's exit code and every caller that catches SystemExit see what they saw before. It adds the structured reason. Exit 1 also means a crash, and an in-process caller that has only the code cannot tell the two apart. Its other choices are parsing the log or re-reading a results file that a crash after the SCAN phase leaves behind looking just like a finished scan's.

Raised only where _compute_exit_code returned 1 with results in hand. That function returns 1 with results only from the coverage arms, so the exception's presence is the signal and incompleteness is what it was.

Source code in automated_security_helper/interactions/run_ash_scan.py

class ScanIncompleteExit(SystemExit):
    """``sys.exit(1)`` for a scan that finished with partial coverage.

    A ``SystemExit`` with code 1, so the CLI's exit code and every caller that
    catches ``SystemExit`` see what they saw before. It adds the structured reason.
    Exit 1 also means a crash, and an in-process caller that has only the code
    cannot tell the two apart. Its other choices are parsing the log or re-reading
    a results file that a crash after the SCAN phase leaves behind looking just
    like a finished scan's.

    Raised only where ``_compute_exit_code`` returned 1 with results in hand. That
    function returns 1 with results only from the coverage arms, so the
    exception's presence is the signal and ``incompleteness`` is what it was.
    """

    def __init__(
        self, incompleteness: ScanIncompleteness, results: AshAggregatedResults
    ):
        super().__init__(1)
        self.incompleteness = incompleteness
        self.results = results

scanner_statuses(results)

(name, status) for every scanner in results, in scanner-name order.

What :func:no_scanner_ran reads, and read through get_unified_scanner_metrics for the same reason :func:incomplete_scanners does: that function is what every reporter and the metrics table already use, so the set-level gate answers from the statuses the operator was shown rather than from a second, independently-derived read of results.scanner_results.

Deliberately just the pairs. An earlier form of this shared one pass with incomplete_scanners by filtering these pairs, which stopped being possible when that function grew a second arm reading the per-metric target counters -- a shortfall is not visible in a (name, status) pair. Keeping this narrow is what makes the two gates independently correct; the cost is one extra pass over the metrics, which _compute_exit_code already takes for the findings count.

Source code in automated_security_helper/interactions/run_ash_scan.py

def scanner_statuses(
    results: Optional[AshAggregatedResults],
) -> List[tuple[str, str]]:
    """(name, status) for every scanner in *results*, in scanner-name order.

    What :func:`no_scanner_ran` reads, and read through
    ``get_unified_scanner_metrics`` for the same reason
    :func:`incomplete_scanners` does: that function is what every reporter and the
    metrics table already use, so the set-level gate answers from the statuses the
    operator was shown rather than from a second, independently-derived read of
    ``results.scanner_results``.

    Deliberately just the pairs. An earlier form of this shared one pass with
    ``incomplete_scanners`` by filtering these pairs, which stopped being possible
    when that function grew a second arm reading the per-metric target counters --
    a shortfall is not visible in a (name, status) pair. Keeping this narrow is
    what makes the two gates independently correct; the cost is one extra pass over
    the metrics, which ``_compute_exit_code`` already takes for the findings count.
    """
    if results is None:
        return []
    return [
        (metric.scanner_name, metric.status)
        for metric in get_unified_scanner_metrics(asharp_model=results)
    ]

no_scanner_ran(observed, expected=None)

True when this run reached no verdict about the target.

Two conditions, and the second exists because the first could not express it.

1. observed is non-empty and none of its scanners reached a verdict. SKIPPED has to be tolerated one entry at a time -- it is how sharding and --exclude-scanners record work a run was never meant to do -- so a file in which every entry is SKIPPED clears the per-scanner pass while having measured nothing.

2. observed is empty AND expected is not. An empty scanner set used to be exempt unconditionally, on the reasoning that it is reachable from a legitimate --phases convert run. That reasoning is sound but covers two different states, and a boolean over observed alone cannot separate them: "the scan phase was not requested", which is benign, and "the scan phase ran and had nothing to run", which is the silent-zero case this gate exists for.

expected is what separates them, and it needs no new state to do it. ScanPhase is what records metadata.expected_scanners, so a recorded roster means the scan phase ran; no roster and no scanners means it never did.

Defaults to None so a results file written by a version that recorded no roster -- which ash merge reads, from whatever ASH produced each shard -- keeps the old benign reading rather than becoming a failure on upgrade.

Source code in automated_security_helper/interactions/run_ash_scan.py

def no_scanner_ran(
    observed: List[tuple[str, str]],
    expected: Optional[List[str]] = None,
) -> bool:
    """True when this run reached no verdict about the target.

    Two conditions, and the second exists because the first could not express it.

    1. *observed* is non-empty and none of its scanners reached a verdict. SKIPPED
       has to be tolerated one entry at a time -- it is how sharding and
       ``--exclude-scanners`` record work a run was never meant to do -- so a file
       in which every entry is SKIPPED clears the per-scanner pass while having
       measured nothing.

    2. *observed* is empty AND *expected* is not. An empty scanner set used to be
       exempt unconditionally, on the reasoning that it is reachable from a
       legitimate ``--phases convert`` run. That reasoning is sound but covers two
       different states, and a boolean over ``observed`` alone cannot separate
       them: "the scan phase was not requested", which is benign, and "the scan
       phase ran and had nothing to run", which is the silent-zero case this gate
       exists for.

    *expected* is what separates them, and it needs no new state to do it.
    ``ScanPhase`` is what records ``metadata.expected_scanners``, so a recorded
    roster means the scan phase ran; no roster and no scanners means it never did.

    Defaults to None so a results file written by a version that recorded no roster
    -- which ``ash merge`` reads, from whatever ASH produced each shard -- keeps the
    old benign reading rather than becoming a failure on upgrade.
    """
    if not observed:
        return bool(expected)
    return not any(status in _RAN_SCANNER_STATUSES for _, status in observed)

incomplete_scanners(results)

(name, status) for every scanner that was selected and did not complete.

"Did not complete" covers two distinct failures, and it did not always cover the second:

1. The scanner never produced a result -- status ERROR or MISSING.
2. The scanner ran, reported a status, and could not evaluate some of the targets it was given.

Only (1) was selected on, because that is a status test and (2) does not move a target's status. ScanResultsContainer.determine_status returns ERROR only once targets_failed >= targets_attempted, so a scanner that lost some of its input keeps whatever the severity gate gave it and the gate could not see it. Measured on this repository under its own config: cdk-nag attempts 10 targets and fails 4, its per-target container still reports PASSED, and the gate exited 0. Two of those four are real CloudFormation templates that went unscanned.

Do not read "(2) does not change the status" as "nothing in this change touches a status". A sibling commit ORs any_target_errored into the error flag, so a target tree that lost ALL of its targets -- which determine_status does report as ERROR -- now reaches the rolled-up status where previously only the "source" report was consulted, and even that only when the scanner was absent from scanner_results. That is a status change, it is not opted in, and it is the thing the CHANGELOG's "Behavior changes" entry describes. The two arms are separable: partial loss stays out of the status and is expressed here; total loss on any tree is a status.

Case (2) is expressed HERE and not as a status, and not by widening _INCOMPLETE_SCANNER_STATUSES. The reason is a caller rather than taste. cli.merge._completed keys on status against that same set to answer a different question -- whether a shard's scanner ran at all -- and _verify_shard_contributions refuses a merge outright where a shard completed none of the scanners it owned. A scanner that lost one target of ten ran, so any status-shaped expression of partial coverage would propagate into shard refusal and start rejecting healthy shards, failing the merge far from the code that caused it. That is the concrete cost of adding a ScannerStatus member for this, and the reason none was added.

Worth being precise about the residual risk, because two earlier versions of this comment got it wrong in opposite directions. One said _completed inspects a ScannerTargetStatusInfo, which declares no target counters, so the protection is structural rather than conventional. That is false, and measurably so: the model sets extra="allow", so counters written into scanner_results land in model_extra and a getattr for them succeeds. Nothing structural stops _completed reading coverage; what stops it is that it does not, which is a behavior and therefore something a test can hold. tests/unit/interactions/test_fail_on_partial_target_coverage.py holds it, and mutating _completed to consult the counters reddens it.

The other claimed no reporter and no summary table sees anything new, and this change is the reason both do. ScannerMetrics gained targets_attempted and targets_failed; the console table, the markdown report and ash.flat.json all carry them, and the first two grew an "Incomplete coverage" section. Measured on this repository, ash.summary.md gained ### Incomplete coverage and ash.flat.json gained the two keys. What is genuinely untouched is narrower and worth naming exactly: _completed, and the DEFAULT exit code, which reaches this function only once _resolve_fail_on_incomplete_scanners returns true. tests/unit/cli/test_merge.py pins the boundary from the merge side.

Precedence between the two arms is on TOTALITY, not on status. Total loss satisfies the coverage condition too -- failed >= attempted implies failed > 0 -- and appending counts there would give an ERROR row a parenthetical it never had while saying nothing the status does not already say, so total loss reports the bare status.

A PARTIAL shortfall reports its counts whatever the status is, and that is a correction rather than a preference. Selecting the bare-status arm on status alone made this function unable to deliver what it exists for in the case it was written for: an ERROR scanner that is only partly incomplete took the bare arm and printed cdk-nag: ERROR, never cdk-nag: ERROR (4 of 10 targets unevaluated), so the counts the operator needs to tell "the tool is absent" from "the tool ran and skipped four templates" were dropped by the routing. An ERROR with no counters available still falls to the bare form, because there is no honest denominator to print.

Read through get_unified_scanner_metrics rather than off results.scanner_results directly, so the gate and the report cannot disagree: that function is what every reporter and the metrics table already use, and it is where excluded-versus-missing precedence is decided. The target counters are read from the same rows for the same reason -- ScannerMetrics is what the summary table prints, so the gate fails on exactly the numbers the operator was shown rather than on a second, independently-derived count.

An allowlist narrowing -- --scanners bandit -- does not trip this, because the scanners it leaves out are recorded SKIPPED. That was not always true: the scan phase used to validate a scanner's dependencies before checking whether it had been selected, so on a host without cfn-nag, grype and syft a --scanners bandit run reported those three MISSING while the six tool-present scanners it left out reported SKIPPED. Which status an unselected scanner got therefore depended on whether its tool happened to be installed. See core/phases/scan_phase.py for the ordering that fixed it.

Filtering here against opts.scanners was the alternative and is rejected: it would make the exit code disagree with the status the report prints for the same scanner, and it has no counterpart in ash merge, which has no scanner selection to consult. Fixing the recorded status instead makes both agree.

Tested against _COMPLETE_SCANNER_STATUSES and not against _INCOMPLETE_SCANNER_STATUSES, though the two are complements over the enum. metric.status is a plain string that may have come from a results file this version did not write, and only the allowlist form treats a status outside the enum entirely as incomplete rather than as a scanner that ran.

Parameters:

Name	Type	Description	Default
results	Optional[AshAggregatedResults]	The aggregated results, or None when the scan produced none.	required

Returns:

Type	Description
List[tuple[str, str]]	Pairs in scanner-name order, empty when every selected scanner completed.
List[tuple[str, str]]	The second element is the scanner's own status for a status-based
List[tuple[str, str]]	incompleteness, and that status followed by the unevaluated-target counts
List[tuple[str, str]]	for a coverage-based one. It is a display string, not a status token:
List[tuple[str, str]]	both callers interpolate it into a message and neither parses it.

Source code in automated_security_helper/interactions/run_ash_scan.py

def incomplete_scanners(
    results: Optional[AshAggregatedResults],
) -> List[tuple[str, str]]:
    """(name, status) for every scanner that was selected and did not complete.

    "Did not complete" covers two distinct failures, and it did not always cover
    the second:

    1. The scanner never produced a result -- status ERROR or MISSING.
    2. The scanner ran, reported a status, and could not evaluate some of the
       targets it was given.

    Only (1) was selected on, because that is a status test and (2) does not move
    a target's status. ``ScanResultsContainer.determine_status`` returns ERROR only
    once ``targets_failed >= targets_attempted``, so a scanner that lost some of
    its input keeps whatever the severity gate gave it and the gate could not see
    it. Measured on this repository under its own config: cdk-nag attempts 10
    targets and fails 4, its per-target container still reports PASSED, and the
    gate exited 0. Two of those four are real CloudFormation templates that went
    unscanned.

    Do not read "(2) does not change the status" as "nothing in this change
    touches a status". A sibling commit ORs ``any_target_errored`` into the
    ``error`` flag, so a target tree that lost ALL of its targets -- which
    ``determine_status`` does report as ERROR -- now reaches the rolled-up status
    where previously only the ``"source"`` report was consulted, and even that only
    when the scanner was absent from ``scanner_results``. That is a status change,
    it is not opted in, and it is the thing the CHANGELOG's "Behavior changes"
    entry describes. The two arms are separable: partial loss stays out of the
    status and is expressed here; total loss on any tree is a status.

    Case (2) is expressed HERE and not as a status, and not by widening
    ``_INCOMPLETE_SCANNER_STATUSES``. The reason is a caller rather than taste.
    ``cli.merge._completed`` keys on status against that same set to answer a
    different question -- whether a shard's scanner ran at all -- and
    ``_verify_shard_contributions`` refuses a merge outright where a shard
    completed none of the scanners it owned. A scanner that lost one target of ten
    ran, so any status-shaped expression of partial coverage would propagate into
    shard refusal and start rejecting healthy shards, failing the merge far from
    the code that caused it. That is the concrete cost of adding a
    ``ScannerStatus`` member for this, and the reason none was added.

    Worth being precise about the residual risk, because two earlier versions of
    this comment got it wrong in opposite directions. One said ``_completed``
    inspects a ``ScannerTargetStatusInfo``, which declares no target counters, so
    the protection is structural rather than conventional. That is false, and
    measurably so: the model sets ``extra="allow"``, so counters written into
    ``scanner_results`` land in ``model_extra`` and a ``getattr`` for them
    succeeds. Nothing structural stops ``_completed`` reading coverage; what stops
    it is that it does not, which is a behavior and therefore something a test can
    hold. ``tests/unit/interactions/test_fail_on_partial_target_coverage.py``
    holds it, and mutating ``_completed`` to consult the counters reddens it.

    The other claimed no reporter and no summary table sees anything new, and this
    change is the reason both do. ``ScannerMetrics`` gained ``targets_attempted``
    and ``targets_failed``; the console table, the markdown report and
    ``ash.flat.json`` all carry them, and the first two grew an "Incomplete
    coverage" section. Measured on this repository, ``ash.summary.md`` gained
    ``### Incomplete coverage`` and ``ash.flat.json`` gained the two keys. What is
    genuinely untouched is narrower and worth naming exactly: ``_completed``, and
    the DEFAULT exit code, which reaches this function only once
    ``_resolve_fail_on_incomplete_scanners`` returns true.
    ``tests/unit/cli/test_merge.py`` pins the boundary from the merge side.

    Precedence between the two arms is on TOTALITY, not on status. Total loss
    satisfies the coverage condition too -- ``failed >= attempted`` implies
    ``failed > 0`` -- and appending counts there would give an ERROR row a
    parenthetical it never had while saying nothing the status does not already
    say, so total loss reports the bare status.

    A PARTIAL shortfall reports its counts whatever the status is, and that is a
    correction rather than a preference. Selecting the bare-status arm on status
    alone made this function unable to deliver what it exists for in the case it
    was written for: an ERROR scanner that is only partly incomplete took the bare
    arm and printed ``cdk-nag: ERROR``, never ``cdk-nag: ERROR (4 of 10 targets
    unevaluated)``, so the counts the operator needs to tell "the tool is absent"
    from "the tool ran and skipped four templates" were dropped by the routing.
    An ERROR with no counters available still falls to the bare form, because there
    is no honest denominator to print.

    Read through ``get_unified_scanner_metrics`` rather than off
    ``results.scanner_results`` directly, so the gate and the report cannot
    disagree: that function is what every reporter and the metrics table already
    use, and it is where excluded-versus-missing precedence is decided. The target
    counters are read from the same rows for the same reason -- ``ScannerMetrics``
    is what the summary table prints, so the gate fails on exactly the numbers the
    operator was shown rather than on a second, independently-derived count.

    An allowlist narrowing -- ``--scanners bandit`` -- does not trip this, because
    the scanners it leaves out are recorded SKIPPED. That was not always true: the
    scan phase used to validate a scanner's dependencies before checking whether it
    had been selected, so on a host without cfn-nag, grype and syft a
    ``--scanners bandit`` run reported those three MISSING while the six
    tool-present scanners it left out reported SKIPPED. Which status an unselected
    scanner got therefore depended on whether its tool happened to be installed.
    See ``core/phases/scan_phase.py`` for the ordering that fixed it.

    Filtering here against ``opts.scanners`` was the alternative and is rejected:
    it would make the exit code disagree with the status the report prints for the
    same scanner, and it has no counterpart in ``ash merge``, which has no scanner
    selection to consult. Fixing the recorded status instead makes both agree.

    Tested against ``_COMPLETE_SCANNER_STATUSES`` and not against
    ``_INCOMPLETE_SCANNER_STATUSES``, though the two are complements over the enum.
    ``metric.status`` is a plain string that may have come from a results file this
    version did not write, and only the allowlist form treats a status outside the
    enum entirely as incomplete rather than as a scanner that ran.

    Args:
        results: The aggregated results, or None when the scan produced none.

    Returns:
        Pairs in scanner-name order, empty when every selected scanner completed.
        The second element is the scanner's own status for a status-based
        incompleteness, and that status followed by the unevaluated-target counts
        for a coverage-based one. It is a display string, not a status token:
        both callers interpolate it into a message and neither parses it.
    """
    if results is None:
        return []

    # A scanner that ran against a content database past its declared bound, under the
    # default `content_db_staleness: fail`, is incomplete in the sense this function
    # means: its findings are real, and its silence about anything published since the
    # database was built is not. Listed here so the callers that read only this list --
    # the workspace layer and `ash merge` -- name it. Under `warn` the record is not
    # enforced and is not listed.
    #
    # The exit code does NOT come from this arm. A stale database fails the scan
    # whether or not `fail_on_incomplete_scanners` is on, so `scan_incompleteness`
    # carries it in its own field, which `_compute_exit_code` reads, and drops a row
    # listed here only for its stale database so the one condition is not reported
    # twice. See `_listed_only_for_a_stale_database`.
    stale_by_scanner: dict[str, list[str]] = {}
    for record in stale_content_databases(results, enforced_only=True):
        stale_by_scanner.setdefault(record.scanner, []).append(record.name)

    def _with_stale(name: str, status: str) -> tuple[str, str]:
        names = stale_by_scanner.pop(name, None)
        if names:
            status = f"{status} (stale content database: {', '.join(names)})"
        return (name, status)

    listed: list[tuple[str, str]] = []
    for metric in get_unified_scanner_metrics(asharp_model=results):
        # Against the allowlist, not against _INCOMPLETE_SCANNER_STATUSES, and the
        # two are not interchangeable here even though they partition the enum.
        # `metric.status` is a plain string that may have come from a results file
        # this version did not write -- `ash merge` reads shard results from
        # whatever ASH produced each one -- and only the allowlist form treats a
        # status outside the enum entirely as incomplete rather than as a scanner
        # that ran. This arm arrived from one side of a merge spelled as
        # `in _INCOMPLETE_SCANNER_STATUSES`, which is the denylist that inversion
        # replaced; `TestStatusClassificationFailsClosed` is what catches it.
        status_is_incomplete = metric.status not in _COMPLETE_SCANNER_STATUSES
        shortfall = _partial_coverage(metric)

        # No usable counters. The status is the only thing there is to report, so
        # this is the arm an ERROR or MISSING with no denominator falls to.
        if shortfall is None:
            if status_is_incomplete:
                listed.append(_with_stale(metric.scanner_name, metric.status))
            elif metric.scanner_name in stale_by_scanner:
                listed.append(_with_stale(metric.scanner_name, metric.status))
            continue

        attempted, failed = shortfall
        # Total loss, and a status that already says so. The counts would add a
        # parenthetical that repeats the status.
        if status_is_incomplete and failed >= attempted:
            listed.append(_with_stale(metric.scanner_name, metric.status))
            continue

        # A partial shortfall against a known denominator, whatever the status.
        # Selecting on status ahead of this is what dropped the counts from the
        # measured case.
        listed.append(
            _with_stale(
                metric.scanner_name,
                f"{metric.status} ({failed} of {attempted} targets unevaluated)",
            )
        )
    return listed

incomplete_converters(results)

(name, reason) for every converter that was meant to run and did not.

Conversion produces the second set of targets the scanners are given: notebooks become Python, archives become their contents. A converter that crashed or whose tool was absent therefore costs scan coverage, and it did so with no effect on any verdict -- the scanners that ran reported PASSED on the targets they were handed, and nothing asked whether the targets that should have existed did.

Two conditions, read off the recorded row rather than re-derived:

1. failure is set -- the converter raised, or was dropped by the plugin filter for a reason its own dependency check did not explain. ConvertPhase records both.
2. dependencies_satisfied is False -- its external tool was not available.

excluded is checked first and wins over both. It is the converter-side counterpart of a SKIPPED scanner: work the run was never meant to do, which is what a config-disabled converter and one dropped by --python-based-plugins-only are. A gate that failed on those would fail every run that turns conversion off, which is a supported configuration.

Read with getattr because a results file written by an older version carries rows without failure, and ash merge reads shard results from whatever ASH produced each one.

Parameters:

Name	Type	Description	Default
results	Optional[AshAggregatedResults]	The aggregated results, or None when the scan produced none.	required

Returns:

Type	Description
List[tuple[str, str]]	Pairs in the order the rows were recorded, empty when every converter either
List[tuple[str, str]]	ran or was excluded. The second element is a display string, not a token;
List[tuple[str, str]]	the caller interpolates it into a message and does not parse it.

Source code in automated_security_helper/interactions/run_ash_scan.py

def incomplete_converters(
    results: Optional[AshAggregatedResults],
) -> List[tuple[str, str]]:
    """(name, reason) for every converter that was meant to run and did not.

    Conversion produces the second set of targets the scanners are given: notebooks
    become Python, archives become their contents. A converter that crashed or whose
    tool was absent therefore costs scan coverage, and it did so with no effect on
    any verdict -- the scanners that ran reported PASSED on the targets they were
    handed, and nothing asked whether the targets that should have existed did.

    Two conditions, read off the recorded row rather than re-derived:

    1. ``failure`` is set -- the converter raised, or was dropped by the plugin
       filter for a reason its own dependency check did not explain. ConvertPhase
       records both.
    2. ``dependencies_satisfied`` is False -- its external tool was not available.

    ``excluded`` is checked first and wins over both. It is the converter-side
    counterpart of a SKIPPED scanner: work the run was never meant to do, which is
    what a config-disabled converter and one dropped by
    ``--python-based-plugins-only`` are. A gate that failed on those would fail every
    run that turns conversion off, which is a supported configuration.

    Read with ``getattr`` because a results file written by an older version carries
    rows without ``failure``, and ``ash merge`` reads shard results from whatever
    ASH produced each one.

    Args:
        results: The aggregated results, or None when the scan produced none.

    Returns:
        Pairs in the order the rows were recorded, empty when every converter either
        ran or was excluded. The second element is a display string, not a token;
        the caller interpolates it into a message and does not parse it.
    """
    if results is None:
        return []

    listed: list[tuple[str, str]] = []
    for name, row in (getattr(results, "converter_results", None) or {}).items():
        if getattr(row, "excluded", False):
            continue
        failure = getattr(row, "failure", None)
        if failure:
            listed.append((name, str(failure)))
        elif getattr(row, "dependencies_satisfied", True) is False:
            # An unavailable converter costs coverage only if it had inputs. The count
            # is established without its tool, which is what makes it answerable for a
            # converter that never ran -- see
            # ``ConverterPluginBase.candidate_input_count``.
            #
            # ONLY an explicit 0 exempts the row. None means the converter reports no
            # count, and that is treated exactly as this function treated every row
            # before the field existed: still incomplete. So a converter that has not
            # opted in, one whose count raised, and a results file written by a version
            # predating the field all keep the strict answer, and nothing is exempted by
            # missing information. ``is 0`` rather than ``== 0`` to keep False out:
            # ``bool`` is an ``int`` subclass and ``False == 0``, so a row carrying a
            # boolean would otherwise silence the gate.
            candidates = getattr(row, "candidate_inputs", None)
            if candidates is not None and not isinstance(candidates, bool):
                if candidates == 0:
                    continue
            listed.append((name, "dependencies unavailable, so it never ran"))
    return listed

print_incompleteness_message(incomplete_scanner_rows, incomplete_converter_rows, stale_database_records=())

Explain an exit code of 1 in terms of what actually caused it.

WHY THIS IS A FUNCTION RATHER THAN THREE BRANCHES INLINE

It was inline, and the converter arm had no branch of its own, so it fell through to "Exiting due to exception during ASH scan" -- which is false. Nothing raises on that path: _compute_exit_code returns 1 from the converter gate after a logger.error. Both Nix CI legs exited that way, and the wording sent a reader through 1572 log lines hunting a traceback that was never written, with the real cause recorded only in a log line seventy lines earlier.

Inline it also could not be tested. The enclosing function runs a whole scan, so no unit test could assert on the wording, which is how a message this wrong survived. Extracted, the arm selection is a pure function of two lists and tests/unit/interactions/test_fail_on_incomplete_converters.py asserts it directly.

THE SCANNER ARM WINS WHEN BOTH ARE NON-EMPTY, deliberately. A scanner that did not run is the more specific and more serious finding, and conversion shortfalls often follow from the same absent environment. The converter arm is still reachable on its own, which is the case Nix mode produces.

STALE CONTENT DATABASES ARE PRINTED FIRST AND ALONGSIDE, not as a third arm of the precedence above. Their advice is different (refresh the database, or opt out with --allow-stale-content-db), and that arm is not behind the completeness gate, so a run can trip it together with either of the others. A scanner listed only for its stale database never reaches incomplete_scanner_rows: scan_incompleteness carries it in the stale records instead, so the install-the-tool advice below is not given for a tool that ran.

Source code in automated_security_helper/interactions/run_ash_scan.py

def print_incompleteness_message(
    incomplete_scanner_rows: List[tuple[str, str]],
    incomplete_converter_rows: List[tuple[str, str]],
    stale_database_records: Sequence[ContentDbAgeRecord] = (),
) -> None:
    """Explain an exit code of 1 in terms of what actually caused it.

    WHY THIS IS A FUNCTION RATHER THAN THREE BRANCHES INLINE
    ------------------------------------------------------
    It was inline, and the converter arm had no branch of its own, so it fell through to
    "Exiting due to exception during ASH scan" -- which is false. Nothing raises on that
    path: ``_compute_exit_code`` returns 1 from the converter gate after a
    ``logger.error``. Both Nix CI legs exited that way, and the wording sent a reader
    through 1572 log lines hunting a traceback that was never written, with the real
    cause recorded only in a log line seventy lines earlier.

    Inline it also could not be tested. The enclosing function runs a whole scan, so no
    unit test could assert on the wording, which is how a message this wrong survived.
    Extracted, the arm selection is a pure function of two lists and
    ``tests/unit/interactions/test_fail_on_incomplete_converters.py`` asserts it
    directly.

    THE SCANNER ARM WINS WHEN BOTH ARE NON-EMPTY, deliberately. A scanner that did not
    run is the more specific and more serious finding, and conversion shortfalls often
    follow from the same absent environment. The converter arm is still reachable on its
    own, which is the case Nix mode produces.

    STALE CONTENT DATABASES ARE PRINTED FIRST AND ALONGSIDE, not as a third arm of
    the precedence above. Their advice is different (refresh the database, or opt out
    with ``--allow-stale-content-db``), and that arm is not behind the completeness
    gate, so a run can trip it together with either of the others. A scanner listed
    only for its stale database never reaches ``incomplete_scanner_rows``:
    ``scan_incompleteness`` carries it in the stale records instead, so the
    install-the-tool advice below is not given for a tool that ran.
    """
    if stale_database_records:
        print(
            "\n[bold red]ERROR (1) Exiting because "
            f"{len(stale_database_records)} content database(s) are past their "
            "declared age bound[/bold red]"
        )
        for _record in stale_database_records:
            print(f"  [red]{escape_markup(_record.message())}[/red]")
    if incomplete_scanner_rows:
        # "did not run" would be false for the coverage case: that scanner ran,
        # reported a status, and could not read some of its targets. Sending an
        # operator to install a tool that is already installed is the specific
        # wrong turn this wording avoids.
        print(
            "\n[bold red]ERROR (1) Exiting because the scan was incomplete: "
            f"{len(incomplete_scanner_rows)} selected scanner(s) did not evaluate "
            "everything they were given[/bold red]"
        )
        for _name, _status in incomplete_scanner_rows:
            print(f"  [red]{_name}: {_status}[/red]")
        print(
            "[yellow]ERROR means the scanner ran and failed; MISSING means its "
            "dependencies were unavailable; a target count means the scanner ran "
            "but could not read that many of its inputs. Install the missing "
            "tools, fix or exclude the unreadable targets, exclude the scanners "
            "with --exclude-scanners, or drop --fail-on-incomplete-scanners to "
            "accept a partial scan.[/yellow]"
        )
    elif incomplete_converter_rows:
        # Names the converter, because that is the actionable part -- the fix is to
        # install that converter's tool or accept the partial scan, and neither is
        # discoverable from "an exception occurred".
        print(
            "\n[bold red]ERROR (1) Exiting because conversion was incomplete: "
            f"{len(incomplete_converter_rows)} converter(s) did not process the "
            "files they were given[/bold red]"
        )
        for _name, _reason in incomplete_converter_rows:
            print(f"  [red]{_name}: {_reason}[/red]")
        print(
            "[yellow]A converter that did not run leaves its inputs unscanned, so "
            "the scanners saw fewer files than this repository holds. Install the "
            "converter's tool, disable the converter if its inputs are not wanted, "
            "or drop --fail-on-incomplete-scanners to accept a partial scan. A "
            "converter with nothing to convert never reports here.[/yellow]"
        )
    elif not stale_database_records:
        print("[bold red]ERROR (1) Exiting due to exception during ASH scan[/bold red]")

unevaluated_rules(results)

Every rule a scanner reported, at run level, that it could not evaluate.

WHY THIS IS A SECOND FUNCTION RATHER THAN PART OF incomplete_scanners

incomplete_scanners answers "which scanners lost whole targets", and it answers it from the target counters. A rule that raised mid-evaluation loses neither a scanner nor a target: the scanner ran, the target was read, most of the rules reached a verdict, and one did not. Counting it as a lost target would overstate in a way that is measurable rather than theoretical -- ScanResultsContainer.determine_status returns ERROR once targets_failed >= targets_attempted, so a rule that raises on every template (which is the normal case for one that cannot resolve an intrinsic) would report the scanner as having evaluated nothing, and the cdk-nag scanner's own log line would read "No rules were evaluated" about a run that evaluated all but one of them.

So the granularity is the rule, and the fact is read from where SARIF already puts it.

WHAT IT READS, AND WHY THAT IS THE RIGHT CHANNEL

invocation.toolExecutionNotifications is, in the schema's own words, "A list of runtime conditions detected by the tool during the analysis", and notification.associatedRule is "A reference used to locate the rule descriptor associated with this notification". A rule raising instead of returning a verdict is a runtime condition, and the rule it happened to is what to associate it with. The cdk-nag scanner already writes exactly that.

Reading it here rather than inventing a counter is what makes this gate scanner-agnostic: any scanner -- or any externally-produced SARIF that ASH ingests -- which reports an error-level runtime condition is reporting that part of its analysis did not run, and that is the thing being gated on.

ONLY level == "error". warning is the field's default and a tool may use it for conditions that cost no coverage, so gating on it would fail scans for notes. The cdk-nag scanner sets error deliberately and says so.

.value RATHER THAN str() ON THE LEVEL. Level is a str-mixin enum, so Enum.__str__ still wins and str(Level.error) renders "Level.error", which matches nothing. Comparing the raw member against "error" works because of the str mixin, but only for a model built in-process; a model round-tripped through JSON carries a plain string. Both shapes are handled by taking .value when it is there.

Read from the in-memory model rather than re-reading reports/ash.sarif. Verified end to end against a real cdk-nag run: the notifications survive sanitize_sarif_paths, apply_suppressions_to_sarif, attach_scanner_details, merge_sarif_report, and a second scanner merging into the same aggregate. Nothing in that chain rewrites them, which is why no disk read is needed to see them.

SUPPRESSIONS ARE HONORED, AND THAT IS NOT A CONVENIENCE

A notification carries no suppression of its own -- SARIF puts suppressions on results -- so read naively this gate would be unsuppressable, and an operator who has already reviewed a rule's failure and accepted not knowing its verdict would have no way to say so. That is not hypothetical: this repository's own .ash/.ash.yaml carries fifteen such entries under the heading "rules that threw and never ran", each with a reviewed reason, and its own note calls naming them "the only way to keep the exit code honest". A gate that ignored them would fail ASH's own default scan with no escape hatch, which is a worse defect than the one being fixed.

So a rule is reported only when at least one of its not-evaluated results is unsuppressed. The results are what suppression applies to, and consulting them is what lets the existing mechanism reach a fact recorded somewhere it cannot be attached.

kind is the filter rather than the cdk-nag property bag, so this stays generic to SARIF. Restricting to not-evaluated rows matters: one rule can throw on one resource while reaching a verdict on another, and counting an ordinary unsuppressed finding as evidence would report a rule whose only failure was suppressed.

A rule with an error-level notification and NO matching result is reported. Absence of a result is not evidence of suppression, and defaulting to silence there would reintroduce the silent pass through the one shape nothing checks.

Returns:

Type	Description
List[str]	Rule ids in sorted order, deduplicated, with the notification's message
List[str]	substituted for a notification that names no rule so a condition is never
List[str]	silently dropped for lacking an id. Empty when every rule was evaluated,
List[str]	which is the case for every scanner that reports no such condition at all.

Source code in automated_security_helper/interactions/run_ash_scan.py

def unevaluated_rules(results: Optional[AshAggregatedResults]) -> List[str]:
    """Every rule a scanner reported, at run level, that it could not evaluate.

    WHY THIS IS A SECOND FUNCTION RATHER THAN PART OF ``incomplete_scanners``
    -----------------------------------------------------------------------
    ``incomplete_scanners`` answers "which scanners lost whole targets", and it
    answers it from the target counters. A rule that raised mid-evaluation loses
    neither a scanner nor a target: the scanner ran, the target was read, most of
    the rules reached a verdict, and one did not. Counting it as a lost target
    would overstate in a way that is measurable rather than theoretical --
    ``ScanResultsContainer.determine_status`` returns ERROR once
    ``targets_failed >= targets_attempted``, so a rule that raises on every
    template (which is the normal case for one that cannot resolve an intrinsic)
    would report the scanner as having evaluated nothing, and the cdk-nag
    scanner's own log line would read "No rules were evaluated" about a run that
    evaluated all but one of them.

    So the granularity is the rule, and the fact is read from where SARIF already
    puts it.

    WHAT IT READS, AND WHY THAT IS THE RIGHT CHANNEL
    -----------------------------------------------
    ``invocation.toolExecutionNotifications`` is, in the schema's own words, "A
    list of runtime conditions detected by the tool during the analysis", and
    ``notification.associatedRule`` is "A reference used to locate the rule
    descriptor associated with this notification". A rule raising instead of
    returning a verdict is a runtime condition, and the rule it happened to is
    what to associate it with. The cdk-nag scanner already writes exactly that.

    Reading it here rather than inventing a counter is what makes this gate
    scanner-agnostic: any scanner -- or any externally-produced SARIF that ASH
    ingests -- which reports an error-level runtime condition is reporting that
    part of its analysis did not run, and that is the thing being gated on.

    ONLY ``level == "error"``. ``warning`` is the field's default and a tool may
    use it for conditions that cost no coverage, so gating on it would fail scans
    for notes. The cdk-nag scanner sets ``error`` deliberately and says so.

    ``.value`` RATHER THAN ``str()`` ON THE LEVEL. ``Level`` is a str-mixin enum,
    so ``Enum.__str__`` still wins and ``str(Level.error)`` renders
    ``"Level.error"``, which matches nothing. Comparing the raw member against
    ``"error"`` works because of the str mixin, but only for a model built
    in-process; a model round-tripped through JSON carries a plain string. Both
    shapes are handled by taking ``.value`` when it is there.

    Read from the in-memory model rather than re-reading ``reports/ash.sarif``.
    Verified end to end against a real cdk-nag run: the notifications survive
    ``sanitize_sarif_paths``, ``apply_suppressions_to_sarif``,
    ``attach_scanner_details``, ``merge_sarif_report``, and a second scanner
    merging into the same aggregate. Nothing in that chain rewrites them, which
    is why no disk read is needed to see them.

    SUPPRESSIONS ARE HONORED, AND THAT IS NOT A CONVENIENCE
    ------------------------------------------------------
    A notification carries no suppression of its own -- SARIF puts suppressions on
    results -- so read naively this gate would be unsuppressable, and an operator
    who has already reviewed a rule's failure and accepted not knowing its verdict
    would have no way to say so. That is not hypothetical: this repository's own
    ``.ash/.ash.yaml`` carries fifteen such entries under the heading "rules that
    threw and never ran", each with a reviewed reason, and its own note calls
    naming them "the only way to keep the exit code honest". A gate that ignored
    them would fail ASH's own default scan with no escape hatch, which is a worse
    defect than the one being fixed.

    So a rule is reported only when at least one of its not-evaluated results is
    unsuppressed. The results are what suppression applies to, and consulting them
    is what lets the existing mechanism reach a fact recorded somewhere it cannot
    be attached.

    ``kind`` is the filter rather than the cdk-nag property bag, so this stays
    generic to SARIF. Restricting to not-evaluated rows matters: one rule can throw
    on one resource while reaching a verdict on another, and counting an ordinary
    unsuppressed finding as evidence would report a rule whose only failure was
    suppressed.

    A rule with an error-level notification and NO matching result is reported.
    Absence of a result is not evidence of suppression, and defaulting to silence
    there would reintroduce the silent pass through the one shape nothing checks.

    Returns:
        Rule ids in sorted order, deduplicated, with the notification's message
        substituted for a notification that names no rule so a condition is never
        silently dropped for lacking an id. Empty when every rule was evaluated,
        which is the case for every scanner that reports no such condition at all.
    """
    sarif = getattr(results, "sarif", None)
    if sarif is None:
        return []

    reported: set[str] = set()
    for run in getattr(sarif, "runs", None) or []:
        # Per run, because a rule id is only unique within the tool that reported
        # it and the aggregate holds one run per merged scanner.
        unsuppressed: set[str] = set()
        has_result: set[str] = set()
        for result in getattr(run, "results", None) or []:
            kind = getattr(result, "kind", None)
            if getattr(kind, "value", kind) != "notApplicable":
                continue
            rule_id = str(getattr(result, "ruleId", "") or "")
            has_result.add(rule_id)
            if not getattr(result, "suppressions", None):
                unsuppressed.add(rule_id)

        for invocation in getattr(run, "invocations", None) or []:
            for notification in (
                getattr(invocation, "toolExecutionNotifications", None) or []
            ):
                level = getattr(notification, "level", None)
                if getattr(level, "value", level) != "error":
                    continue
                associated = getattr(notification, "associatedRule", None)
                rule_id = getattr(getattr(associated, "root", None), "id", None)
                if rule_id:
                    rule_id = str(rule_id)
                    if rule_id in has_result and rule_id not in unsuppressed:
                        continue
                    reported.add(rule_id)
                    continue
                message = getattr(getattr(notification, "message", None), "root", None)
                text = str(getattr(message, "text", "") or "").strip()
                reported.add(text or "an unnamed rule")
    return sorted(reported)

incomplete_scanner_reason(status)

The reason token for a scanner incomplete_scanners listed, from its status.

A scanner listed with a status the per-scanner pass accepts (PASSED, FAILED, SKIPPED) is there for the targets it lost, not for its status.

Source code in automated_security_helper/interactions/run_ash_scan.py

def incomplete_scanner_reason(status: str) -> str:
    """The reason token for a scanner ``incomplete_scanners`` listed, from its status.

    A scanner listed with a status the per-scanner pass accepts (PASSED, FAILED,
    SKIPPED) is there for the targets it lost, not for its status.
    """
    if status in _INCOMPLETE_SCANNER_REASONS:
        return _INCOMPLETE_SCANNER_REASONS[status]
    if status in _COMPLETE_SCANNER_STATUSES:
        return "partial_coverage"
    return "unrecognized_status"

scan_incompleteness(results, *, gate, one_shard_of_a_split=False)

Ask the exit-1 coverage questions of results, under the gate as resolved.

gate is _resolve_fail_on_incomplete_scanners's answer. When it is off, only the unevaluated-rule arm is asked, because that arm is the only one _compute_exit_code asks with the gate off. Pass gate=True to get the facts whatever the gate says, which is what a report that states coverage needs.

one_shard_of_a_split exempts the set-level arm for the reason _compute_exit_code gives at that arm.

A stale content database is reported in stale_content_databases whatever the gate says, and ONLY there. incomplete_scanners also lists its scanner, with the database named in the status, so that the workspace layer and ash merge see it; a row that is there for nothing else is dropped from scanners here. Otherwise the one condition would be reported twice with the gate on -- once as an incomplete scanner whose reason reads partial_coverage and whose advice is to install a tool that ran, and once as a stale database -- and once with it off, so the MCP payload and the console message would change shape with a gate the staleness check does not consult. A scanner that is stale AND lost targets or did not run keeps its row: that part is a coverage gap in its own right.

Source code in automated_security_helper/interactions/run_ash_scan.py

def scan_incompleteness(
    results: Optional[AshAggregatedResults],
    *,
    gate: bool,
    one_shard_of_a_split: bool = False,
) -> ScanIncompleteness:
    """Ask the exit-1 coverage questions of *results*, under the gate as resolved.

    ``gate`` is ``_resolve_fail_on_incomplete_scanners``'s answer. When it is off,
    only the unevaluated-rule arm is asked, because that arm is the only one
    ``_compute_exit_code`` asks with the gate off. Pass ``gate=True`` to get the
    facts whatever the gate says, which is what a report that states coverage
    needs.

    ``one_shard_of_a_split`` exempts the set-level arm for the reason
    ``_compute_exit_code`` gives at that arm.

    A stale content database is reported in ``stale_content_databases`` whatever the
    gate says, and ONLY there. ``incomplete_scanners`` also lists its scanner, with
    the database named in the status, so that the workspace layer and ``ash merge``
    see it; a row that is there for nothing else is dropped from ``scanners`` here.
    Otherwise the one condition would be reported twice with the gate on -- once as
    an incomplete scanner whose reason reads ``partial_coverage`` and whose advice is
    to install a tool that ran, and once as a stale database -- and once with it off,
    so the MCP payload and the console message would change shape with a gate the
    staleness check does not consult. A scanner that is stale AND lost targets or
    did not run keeps its row: that part is a coverage gap in its own right.
    """
    if results is None:
        return ScanIncompleteness()
    unevaluated = tuple(unevaluated_rules(results))
    stale = tuple(stale_content_databases(results, enforced_only=True))
    if not gate:
        return ScanIncompleteness(
            unevaluated_rules=unevaluated, stale_content_databases=stale
        )
    observed = scanner_statuses(results)
    expected_roster = list(
        getattr(getattr(results, "metadata", None), "expected_scanners", None) or []
    )
    stale_scanners = {record.scanner for record in stale}
    return ScanIncompleteness(
        scanners=tuple(
            (name, status)
            for name, status in incomplete_scanners(results)
            if not _listed_only_for_a_stale_database(name, status, stale_scanners)
        ),
        observed=tuple(observed),
        no_scanner_ran=(
            not one_shard_of_a_split and no_scanner_ran(observed, expected_roster)
        ),
        converters=tuple(incomplete_converters(results)),
        unevaluated_rules=unevaluated,
        stale_content_databases=stale,
    )

build_project_scan_settings(opts)

Build the per-project settings record a workspace run scans from.

Module-level and public because there are two callers, not one: the CLI's _run_workspace_mode and the MCP surface in automated_security_helper.cli.mcp.workspace. They assemble their ScanOptions differently -- one from typer arguments, one from MCP tool parameters -- but the record handed to execute_workspace has to come from one construction.

Why it is extracted rather than written twice

ProjectScanSettings has 24 fields and every one of them is optional with a plausible default, so a second construction that omits a field produces a valid record and a scan that runs to completion with a setting the caller never chose. Nothing raises. The two worst omissions are config_overrides, where dropping it silently scans with different configuration, and ignore_suppressions, where the default is the lenient direction.

What it owns, and why the boundary is here

Both derived inputs are computed inside: the workspace execution config, via :func:_resolve_workspace_execution_config, which supplies max_parallel_projects and project_timeout; and the phases list, which is the only field with branching behind it. A builder that took either as an argument would push part of the construction back out to its callers, which is where the duplication started.

Note what it does not own. Setting ASH_OFFLINE stays with the caller: it mutates process state and has to be unset in a finally, which a builder returning a value cannot do.

Failure modes

An unreadable ASH config at the workspace root does not raise here. _resolve_workspace_execution_config warns and falls back to the defaults, because these are scheduling knobs -- refusing the whole scan over a typo in one would be a poor trade, and on the MCP path it would surface as an internal error for what is really an operator's config file.

Source code in automated_security_helper/interactions/run_ash_scan.py

def build_project_scan_settings(opts: ScanOptions) -> "ProjectScanSettings":
    """Build the per-project settings record a workspace run scans from.

    Module-level and public because there are two callers, not one: the CLI's
    ``_run_workspace_mode`` and the MCP surface in
    ``automated_security_helper.cli.mcp.workspace``. They assemble their
    ``ScanOptions`` differently -- one from typer arguments, one from MCP tool
    parameters -- but the record handed to ``execute_workspace`` has to come from
    one construction.

    Why it is extracted rather than written twice
    ---------------------------------------------
    ``ProjectScanSettings`` has 24 fields and every one of them is optional with a
    plausible default, so a second construction that omits a field produces a
    valid record and a scan that runs to completion with a setting the caller
    never chose. Nothing raises. The two worst omissions are ``config_overrides``,
    where dropping it silently scans with different configuration, and
    ``ignore_suppressions``, where the default is the lenient direction.

    What it owns, and why the boundary is here
    ------------------------------------------
    Both derived inputs are computed inside: the workspace execution config, via
    :func:`_resolve_workspace_execution_config`, which supplies
    ``max_parallel_projects`` and ``project_timeout``; and the ``phases`` list,
    which is the only field with branching behind it. A builder that took either
    as an argument would push part of the construction back out to its callers,
    which is where the duplication started.

    Note what it does *not* own. Setting ``ASH_OFFLINE`` stays with the caller:
    it mutates process state and has to be unset in a ``finally``, which a
    builder returning a value cannot do.

    Failure modes
    -------------
    An unreadable ASH config at the workspace root does not raise here.
    ``_resolve_workspace_execution_config`` warns and falls back to the defaults,
    because these are scheduling knobs -- refusing the whole scan over a typo in
    one would be a poor trade, and on the MCP path it would surface as an
    internal error for what is really an operator's config file.
    """
    from automated_security_helper.workspace.execution import ProjectScanSettings

    if opts.shard_index is not None or opts.shard_count is not None:
        # Same reasoning as the missing-plan check above: the CLI refuses this
        # combination with an operator-facing message (see
        # cli.scan._validate_shard_options), so reaching here means a programmatic
        # caller passed both. Raised rather than ignored because ProjectScanSettings
        # has no shard fields, so ignoring is not a degraded mode -- it is every
        # shard scanning every project with every scanner, and a merge multiplying
        # each finding by the shard count. RuntimeError rather than
        # ShardSelectionError, which would report a caller bug as though the
        # operator's shard arguments were at fault; theirs are fine, the
        # combination is not.
        raise RuntimeError(
            "Sharding is not supported in workspace mode, and workspace mode "
            "cannot silently ignore it: every shard would scan every project in "
            "full and the merged report would count each finding once per shard. "
            "Scan the workspace whole, or scan one project per job with "
            "source_dir and shard that."
        )

    workspace_config = _resolve_workspace_execution_config(opts)

    phases: List[str] = []
    for phase, name in (
        (ExecutionPhase.CONVERT, "convert"),
        (ExecutionPhase.SCAN, "scan"),
        (ExecutionPhase.REPORT, "report"),
    ):
        if phase in (opts.phases or []):
            phases.append(name)
    if ExecutionPhase.INSPECT in (opts.phases or []) or opts.inspect:
        phases.append("inspect")
    if not phases:
        phases = ["convert", "scan", "report"]

    return ProjectScanSettings(
        output_dir=opts.output_dir,
        phases=tuple(phases),
        enabled_scanners=tuple(opts.scanners or []),
        excluded_scanners=tuple(opts.excluded_scanners or []),
        output_formats=tuple(
            getattr(fmt, "value", str(fmt)) for fmt in (opts.output_formats or [])
        ),
        config_overrides=tuple(opts.config_overrides or []),
        ash_plugin_modules=tuple(opts.ash_plugin_modules or []),
        strategy=getattr(opts.strategy, "value", str(opts.strategy)),
        offline=opts.offline,
        python_based_plugins_only=opts.python_based_plugins_only,
        ignore_suppressions=opts.ignore_suppressions,
        min_severity=opts.min_severity,
        fail_on_findings=opts.fail_on_findings,
        fail_on_incomplete_scanners=opts.fail_on_incomplete_scanners,
        changed_files_only=opts.changed_files_only,
        base_ref=opts.base_ref,
        precommit=opts.mode == RunMode.precommit,
        cleanup=opts.cleanup,
        verbose=opts.verbose,
        debug=opts.debug,
        simple=opts.simple,
        color_system=(
            "windows"
            if platform.system() == "Windows"
            else "auto"
            if opts.color
            else None
        ),
        max_parallel_projects=workspace_config.resolved_max_parallel_projects(),
        project_timeout=workspace_config.project_timeout,
        allow_missing_projects=opts.allow_missing_projects,
        # ``opts.config`` in workspace mode is the fallback for a project that
        # declares no config of its own, not a config for the workspace: each
        # project's own ``.ash.yaml`` still wins, which is what makes one
        # workspace scannable across differently-configured repositories.
        #
        # The caller has to pass the same value as ``resolve_workspace``'s
        # ``default_config``, because the plan's reported threshold comes from
        # resolution and the scan's config from execution. Dropping it here -- which
        # this builder did -- made an MCP client's `profile` argument accepted,
        # threaded through two layers, and silently ignored, which is the same
        # shape of defect as the ``ASH_OFFLINE`` one recorded in cli/mcp/workspace.
        default_config_path=opts.config,
    )

run_ash_scan(source_dir=None, output_dir=None, config=None, config_overrides=None, offline=False, strategy=ExecutionStrategy.PARALLEL, scanners=None, exclude_scanners=None, progress=True, output_formats=None, cleanup=False, phases=None, inspect=False, existing_results=None, python_based_plugins_only=False, quiet=False, simple=False, verbose=False, debug=False, color=True, fail_on_findings=None, fail_on_incomplete_scanners=None, ignore_suppressions=False, min_severity='low', changed_files_only=False, base_ref='origin/main', shard_index=None, shard_count=None, mode=RunMode.local, show_summary=True, log_level=AshLogLevel.INFO, build=True, run=True, force=False, oci_runner=None, build_target=None, offline_semgrep_rulesets='p/ci', container_uid=None, container_gid=None, ash_revision_to_install=None, custom_containerfile=None, custom_build_arg=None, ash_plugin_modules=None, container_network='bridge', workspace_plan=None, allow_missing_projects=False, config_base_gate=None, untrusted_config=False, trusted_config_path=None, *args, **kwargs)

Run an ASH scan against source_dir, outputting results to output_dir.

When workspace_plan is given, source_dir is the workspace root and each project in the plan is scanned in its own scope. See :mod:automated_security_helper.workspace.execution.

Source code in automated_security_helper/interactions/run_ash_scan.py

def run_ash_scan(
    source_dir: str | Path | None = None,
    output_dir: str | Path | None = None,
    config: str | None = None,
    config_overrides: List[str] | None = None,
    offline: bool = False,
    strategy: ExecutionStrategy = ExecutionStrategy.PARALLEL,
    scanners: List[str] | None = None,
    exclude_scanners: List[str] | None = None,
    progress: bool = True,
    output_formats: List[ExportFormat] | None = None,
    cleanup: bool = False,
    phases: List[ExecutionPhase] | None = None,
    inspect: bool = False,
    existing_results: str | None = None,
    python_based_plugins_only: bool = False,
    quiet: bool = False,
    simple: bool = False,
    verbose: bool = False,
    debug: bool = False,
    color: bool = True,
    fail_on_findings: bool | None = None,
    fail_on_incomplete_scanners: bool | None = None,
    ignore_suppressions: bool = False,
    min_severity: str = "low",
    changed_files_only: bool = False,
    base_ref: str = "origin/main",
    shard_index: int | None = None,
    shard_count: int | None = None,
    mode: RunMode = RunMode.local,
    show_summary: bool = True,
    log_level: AshLogLevel = AshLogLevel.INFO,
    # Container-specific args
    build: bool = True,
    run: bool = True,
    force: bool = False,
    oci_runner: str | None = None,
    build_target: BuildTarget | None = None,
    offline_semgrep_rulesets: str = "p/ci",
    container_uid: str | None = None,
    container_gid: str | None = None,
    ash_revision_to_install: str | None = None,
    custom_containerfile: str | None = None,
    custom_build_arg: List[str] | None = None,
    ash_plugin_modules: List[str] | None = None,
    container_network: str = "bridge",
    workspace_plan: "WorkspacePlan | None" = None,
    allow_missing_projects: bool = False,
    config_base_gate: Optional[Callable[[Path], bool]] = None,
    untrusted_config: bool = False,
    trusted_config_path: Optional[str] = None,
    *args,
    **kwargs,
):
    """Run an ASH scan against source_dir, outputting results to output_dir.

    When *workspace_plan* is given, *source_dir* is the workspace root and each
    project in the plan is scanned in its own scope. See
    :mod:`automated_security_helper.workspace.execution`.
    """
    scan_start_time = time.time()

    # Resolve cwd-based defaults at call time (not import time).
    _source_dir: Path = (
        Path(source_dir).absolute() if source_dir is not None else Path.cwd()
    )
    _output_dir: Path = (
        Path(output_dir).absolute()
        if output_dir is not None
        else Path.cwd().joinpath(".ash", "ash_output")
    )

    opts = ScanOptions(
        source_dir=_source_dir,
        output_dir=_output_dir,
        config=config,
        config_overrides=config_overrides,
        offline=offline,
        strategy=strategy,
        scanners=scanners,
        excluded_scanners=exclude_scanners,
        progress=progress,
        output_formats=output_formats,
        cleanup=cleanup,
        phases=phases,
        inspect=inspect,
        existing_results=existing_results,
        python_based_plugins_only=python_based_plugins_only,
        quiet=quiet,
        simple=simple,
        verbose=verbose,
        debug=debug,
        color=color,
        fail_on_findings=fail_on_findings,
        fail_on_incomplete_scanners=fail_on_incomplete_scanners,
        ignore_suppressions=ignore_suppressions,
        min_severity=min_severity,
        changed_files_only=changed_files_only,
        base_ref=base_ref,
        shard_index=shard_index,
        shard_count=shard_count,
        mode=mode,
        show_summary=show_summary,
        log_level=log_level,
        build=build,
        run=run,
        force=force,
        oci_runner=oci_runner,
        build_target=build_target,
        offline_semgrep_rulesets=offline_semgrep_rulesets,
        container_uid=container_uid,
        container_gid=container_gid,
        ash_revision_to_install=ash_revision_to_install,
        custom_containerfile=custom_containerfile,
        custom_build_arg=custom_build_arg,
        ash_plugin_modules=ash_plugin_modules,
        container_network=container_network,
        workspace_plan=workspace_plan,
        allow_missing_projects=allow_missing_projects,
        config_base_gate=config_base_gate,
        untrusted_config=untrusted_config,
        trusted_config_path=trusted_config_path,
    )

    # Only the local orchestrator resolves with these. Container and nix mode run
    # a separate ASH that is handed the config path alone, and workspace mode
    # resolves each project's own config, so in any of them the restriction would
    # be dropped without a word.
    if opts.untrusted_config and (
        opts.mode in (RunMode.container, RunMode.nix) or opts.workspace_plan is not None
    ):
        raise ValueError(
            "untrusted_config is applied only to a single-directory scan in local "
            f"mode, not to mode={opts.mode.value!r}"
            + (" with a workspace plan" if opts.workspace_plan is not None else "")
        )

    _apply_log_level_env(opts)
    _refuse_symlinked_output_dir(opts)
    logger = _setup_logger(opts)

    if opts.workspace_plan is not None and opts.mode != RunMode.container:
        # Workspace mode owns its own verdict and its own summary. It does not go
        # through _compute_exit_code, which answers for one directory against one
        # threshold and has no way to express "project A failed, project B did
        # not".
        workspace_result = _run_workspace_mode(opts, logger)
        if opts.show_summary:
            _print_workspace_summary(workspace_result, opts, scan_start_time)
        if workspace_result.exit_code != 0:
            sys.exit(workspace_result.exit_code)
        return workspace_result

    config_fail_on_findings: Optional[bool] = _resolve_config_fail_on_findings(opts)
    config_fail_on_incomplete_scanners: Optional[bool] = (
        _resolve_config_fail_on_incomplete_scanners(opts)
    )
    results: Optional[AshAggregatedResults]
    if opts.mode == RunMode.container:
        results = _run_container_mode(
            opts,
            logger,
            resolved_fail_on_findings=config_fail_on_findings,
            resolved_fail_on_incomplete_scanners=config_fail_on_incomplete_scanners,
        )
    elif opts.mode == RunMode.nix:
        # No resolved-flag arguments here, unlike container mode, and that asymmetry is
        # deliberate rather than an oversight in the merge that brought these two together.
        # Container mode forwards the flags into the CLI invocation it runs inside the
        # container, because that inner process computes its own verdict. Nix mode returns
        # the parsed results and the verdict is computed once, below, by
        # _compute_exit_code -- which already receives config_fail_on_incomplete_scanners,
        # so nix runs honour it through the shared path.
        results = _run_nix_mode(opts, logger)
    else:
        results, _local_config_fof = _run_local_mode(opts, logger)
        # _run_local_mode resolves config via the live orchestrator; prefer that
        # value over the file-based pre-read when it differs (e.g. config_overrides
        # applied by the orchestrator may alter fail_on_findings).
        if _local_config_fof is not None:
            config_fail_on_findings = _local_config_fof

    if opts.workspace_plan is not None:
        # Container mode ran `ash --workspace` inside the container, so the
        # verdict was already computed there by the same code. Re-deriving it on
        # the host from a merged model would answer a different question.
        workspace_payload = getattr(results, "workspace", None)
        exit_code = (
            int(workspace_payload.exit_code)
            if workspace_payload is not None
            else int(WorkspaceExitCode.INTERNAL_ERROR)
        )
        if workspace_payload is None:
            logger.error(
                "The container produced no workspace payload, so no per-project "
                "verdict is available."
            )
        if exit_code != 0:
            sys.exit(exit_code)
        return results

    exit_code = _compute_exit_code(
        results,
        opts,
        config_fail_on_findings,
        config_fail_on_incomplete_scanners,
    )

    if opts.show_summary:
        scanner_metrics = (
            get_unified_scanner_metrics(asharp_model=results) if results else []
        )
        actionable_findings = sum(item.actionable for item in scanner_metrics)
        _print_summary(results, opts, scan_start_time, actionable_findings)

        if exit_code == 2 and not opts.quiet:
            actionable_count = sum(
                item.actionable
                for item in (
                    get_unified_scanner_metrics(asharp_model=results) if results else []
                )
            )
            print("\n[yellow]=== ASH Exit Codes ===[/yellow]")
            print(
                "  0: Success - No actionable findings or not configured to fail on findings"
            )
            print("  1: Error during execution")
            print(
                f"  2: Actionable findings detected when configured with `fail_on_findings: true`."
                f" Default is True. Current value: {opts.fail_on_findings if opts.fail_on_findings is not None else True}"
            )
            print(
                f"[bold red]ERROR (2) Exiting due to {actionable_count} actionable findings found in ASH scan[/bold red]"
            )

    _incompleteness = ScanIncompleteness()
    if exit_code == 1:
        # An incomplete scan and a crash share exit 1, so the message has to be
        # chosen from the cause rather than the code. Printing "an exception
        # occurred" for a run whose scanners simply were not installed sends the
        # operator looking for a traceback that does not exist.
        #
        # The same object _compute_exit_code reached its verdict from, so the
        # message, the exit code and the exception below agree on the cause. The
        # scanner and converter rows are empty when the gate is off, because both
        # arms live behind it; the stale-database records are not, because that arm
        # is not behind the gate. The message keeps the scanner arm's precedence
        # over the converter arm when a run trips both, since a scanner that did not
        # run is the more specific finding.
        _incompleteness = scan_incompleteness(
            results,
            gate=_resolve_fail_on_incomplete_scanners(
                results, opts, config_fail_on_incomplete_scanners
            ),
            one_shard_of_a_split=(
                opts.shard_index is not None or opts.shard_count is not None
            ),
        )
        print_incompleteness_message(
            list(_incompleteness.scanners),
            list(_incompleteness.converters),
            list(_incompleteness.stale_content_databases),
        )

    if exit_code == 1 and results is not None and _incompleteness:
        # Still exit 1. The exception adds the reason, for an in-process caller
        # such as the MCP runner that has to tell this from a crash.
        raise ScanIncompleteExit(_incompleteness, results)
    if exit_code != 0:
        sys.exit(exit_code)

    return results

automated_security_helper.models.core

Core models for security findings.

IgnorePathWithReason

Bases: BaseModel

Represents a path exclusion entry.

Source code in automated_security_helper/models/core.py

class IgnorePathWithReason(BaseModel):
    """Represents a path exclusion entry."""

    path: Annotated[str, Field(..., description="Path or pattern to exclude")]
    reason: Annotated[str, Field(..., description="Reason for exclusion")]
    expiration: Annotated[
        str | None, Field(None, description="(Optional) Expiration date (YYYY-MM-DD)")
    ] = None

    def matches_path(self, file_path: str) -> bool:
        """Return True if ``file_path`` matches this entry's path pattern.

        Supports exact matches, simple globs (``*.py``), and recursive globs
        (``tests/**/*.py``). Matching is case-insensitive for OS portability.
        """
        return _path_pattern_matches(file_path, self.path)

matches_path(file_path)

Return True if file_path matches this entry's path pattern.

Supports exact matches, simple globs (*.py), and recursive globs (tests/**/*.py). Matching is case-insensitive for OS portability.

Source code in automated_security_helper/models/core.py

def matches_path(self, file_path: str) -> bool:
    """Return True if ``file_path`` matches this entry's path pattern.

    Supports exact matches, simple globs (``*.py``), and recursive globs
    (``tests/**/*.py``). Matching is case-insensitive for OS portability.
    """
    return _path_pattern_matches(file_path, self.path)

ToolArgs

Bases: BaseModel

Base class for tool argument dictionaries.

Source code in automated_security_helper/models/core.py

class ToolArgs(BaseModel):
    """Base class for tool argument dictionaries."""

    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

    output_arg: str | None = None
    scan_path_arg: str | None = None
    format_arg: str | None = None
    format_arg_value: str | None = None
    extra_args: List[ToolExtraArg] = []

AshSuppression

Bases: IgnorePathWithReason

Represents a finding suppression rule.

Source code in automated_security_helper/models/core.py

class AshSuppression(IgnorePathWithReason):
    """Represents a finding suppression rule."""

    rule_id: Annotated[str | None, Field(None, description="Rule ID to suppress")] = (
        None
    )
    line_start: Annotated[
        int | None, Field(None, description="(Optional) Starting line number")
    ] = None
    line_end: Annotated[
        int | None, Field(None, description="(Optional) Ending line number")
    ] = None
    package_name: Annotated[
        str | None,
        Field(
            None,
            description=(
                "(Optional) Only suppress findings about this package (glob, "
                "case-insensitive). Findings that do not report a package name "
                "never match."
            ),
        ),
    ] = None
    package_version: Annotated[
        str | None,
        Field(
            None,
            description=(
                "(Optional) Only suppress findings about this installed package "
                "version (glob, case-insensitive). Findings that do not report an "
                "installed version never match."
            ),
        ),
    ] = None
    package_path: Annotated[
        str | None,
        Field(
            None,
            description=(
                "(Optional) Only suppress findings about the package copy installed "
                "at this path, relative to the scan root (glob, supports **), e.g. "
                "'deploy/cdk/node_modules/aws-cdk-lib/node_modules/brace-expansion'. "
                "Separates two copies with the same name and version. Findings that "
                "do not report an install path never match."
            ),
        ),
    ] = None
    symbol: Annotated[
        str | None,
        Field(
            None,
            description=(
                "(Optional) Only suppress findings whose lines lie inside the "
                "definition with this qualified name, e.g. 'MyClass.my_method', "
                "in the finding's file. Names are dotted from the top of the "
                "file, exact and case-sensitive. Resolved by parsing the file "
                "with tree-sitter (Python, JavaScript, TypeScript, Java), which "
                "needs the 'symbols' extra; a finding whose file cannot be "
                "parsed never matches."
            ),
        ),
    ] = None

    @field_validator("line_end")
    @classmethod
    def validate_line_range(cls, v, values):
        """Validate that line_end is greater than or equal to line_start if both are provided."""
        if (
            v is not None
            and hasattr(values, "data")
            and values.data is not None
            and values.data.get("line_start") is not None
            and v < values.data["line_start"]
        ):
            raise ValueError("line_end must be greater than or equal to line_start")
        return v

    @field_validator("package_name", "package_version", "package_path")
    @classmethod
    def validate_package_field_not_blank(cls, v):
        """Reject a blank package field: it reads as "any" but would match nothing."""
        if v is not None and not v.strip():
            raise ValueError(
                "package_name, package_version and package_path must be omitted "
                "rather than left blank"
            )
        return v

    @field_validator("symbol")
    @classmethod
    def validate_symbol(cls, v):
        """Reject a symbol that is not a dotted qualified name.

        A malformed name could never equal a parsed definition's name, so the
        entry would silently match nothing.
        """
        if v is not None and not is_valid_symbol(v):
            raise ValueError(
                f"symbol must be a dotted qualified name such as "
                f"'MyClass.my_method' (identifiers joined by '.', no spaces, "
                f"wildcards or call syntax): {v!r}"
            )
        return v

    @field_validator("expiration")
    @classmethod
    def validate_expiration_date(cls, v):
        """Validate that expiration date is in YYYY-MM-DD format.

        Past dates are accepted; use is_expired to check whether the
        suppression has expired at runtime.
        """
        if v is not None:
            try:
                datetime.strptime(v, "%Y-%m-%d")
            except ValueError:
                raise ValueError(f"Invalid expiration date format. Use YYYY-MM-DD: {v}")
        return v

    @property
    def id(self) -> str:
        """Stable identifier derived from ``path|rule_id|line_start|line_end``.

        Unspecified rule_id is rendered as ``*``. When ``line_end`` is None,
        ``line_start`` is reused to match how suppressions are indexed elsewhere
        in the codebase.

        A suppression that sets any package field gets a fifth part,
        ``name@version@path`` with ``*`` for each unset piece, so two entries
        that differ only by package do not share an id. Entries without
        package fields keep the four-part id they always had.

        A suppression that sets ``symbol`` gets that fifth part (``*@*@*``
        when no package field is set) and the symbol as a sixth. Entries
        without a symbol keep the id they had before the field existed.
        """
        return suppression_id(self.model_dump())

    @property
    def is_expired(self) -> bool:
        """Return True if this suppression has a past expiration date."""
        if not self.expiration:
            return False
        try:
            expiration_date = datetime.strptime(self.expiration, "%Y-%m-%d").date()
        except ValueError:
            return False
        return expiration_date <= date.today()

    @property
    def days_until_expiry(self) -> Optional[int]:
        """Days from today until expiration; None if no expiration is set.

        A negative value indicates the suppression has already expired.
        """
        if not self.expiration:
            return None
        try:
            expiration_date = datetime.strptime(self.expiration, "%Y-%m-%d").date()
        except ValueError:
            return None
        return (expiration_date - date.today()).days

    def matches(
        self,
        finding: "FlatVulnerability",
        symbol_resolver: "SymbolResolver | None" = None,
    ) -> bool:
        """Return True if ``finding`` is covered by this suppression rule.

        Checks rule_id (exact or glob), path (supports ``**``), optional line
        range overlap, the optional package fields, and the optional symbol.
        Expired suppressions never match.

        Each package field that is set must match the finding's corresponding
        field. A finding that does not carry that field does not match: the
        scanner could not say which package it is about, so a package-scoped
        suppression must not assume it is the one intended.

        A symbol is checked last, so a file is parsed only for a finding that
        every other field already matches. It needs ``symbol_resolver``, which
        reads the scanned source; without one a symbol-scoped suppression
        matches nothing, and so does a finding with no line number or in a
        file the resolver cannot parse.
        """
        if self.is_expired:
            return False

        if self.rule_id:
            if finding.rule_id is None:
                return False
            # Case-insensitive glob match for OS portability
            if not fnmatch.fnmatch(finding.rule_id.lower(), self.rule_id.lower()):
                return False

        if not _path_pattern_matches(finding.file_path, self.path):
            return False

        if not self._line_range_matches(finding):
            return False

        if not self._package_matches(finding):
            return False

        if self.symbol is not None:
            if symbol_resolver is None:
                return False
            return symbol_resolver.contains(
                finding.file_path, finding.line_start, finding.line_end, self.symbol
            )

        return True

    def _package_matches(self, finding: "FlatVulnerability") -> bool:
        """Return True if every package field set here matches ``finding``."""
        for pattern, value in (
            (self.package_name, finding.package_name),
            (self.package_version, finding.package_version),
        ):
            if pattern is None:
                continue
            if value is None or not fnmatch.fnmatch(value.lower(), pattern.lower()):
                return False
        if self.package_path is not None:
            if finding.package_path is None:
                return False
            if not _path_pattern_matches(finding.package_path, self.package_path):
                return False
        return True

    def _line_range_matches(self, finding: "FlatVulnerability") -> bool:
        """Return True if ``finding``'s line range overlaps with this suppression."""
        if self.line_start is None and self.line_end is None:
            return True

        if finding.line_start is None:
            return False

        finding_end = (
            finding.line_end if finding.line_end is not None else finding.line_start
        )

        if self.line_start is not None and self.line_end is None:
            return finding_end >= self.line_start

        if self.line_start is None and self.line_end is not None:
            return finding_end <= self.line_end

        finding_start = finding.line_start
        return (finding_start <= (self.line_end or 0)) and (
            finding_end >= (self.line_start or 0)
        )

id property

Stable identifier derived from path|rule_id|line_start|line_end.

Unspecified rule_id is rendered as *. When line_end is None, line_start is reused to match how suppressions are indexed elsewhere in the codebase.

A suppression that sets any package field gets a fifth part, name@version@path with * for each unset piece, so two entries that differ only by package do not share an id. Entries without package fields keep the four-part id they always had.

A suppression that sets symbol gets that fifth part (*@*@* when no package field is set) and the symbol as a sixth. Entries without a symbol keep the id they had before the field existed.

is_expired property

Return True if this suppression has a past expiration date.

days_until_expiry property

Days from today until expiration; None if no expiration is set.

A negative value indicates the suppression has already expired.

validate_line_range(v, values) classmethod

Validate that line_end is greater than or equal to line_start if both are provided.

Source code in automated_security_helper/models/core.py

@field_validator("line_end")
@classmethod
def validate_line_range(cls, v, values):
    """Validate that line_end is greater than or equal to line_start if both are provided."""
    if (
        v is not None
        and hasattr(values, "data")
        and values.data is not None
        and values.data.get("line_start") is not None
        and v < values.data["line_start"]
    ):
        raise ValueError("line_end must be greater than or equal to line_start")
    return v

validate_package_field_not_blank(v) classmethod

Reject a blank package field: it reads as "any" but would match nothing.

Source code in automated_security_helper/models/core.py

@field_validator("package_name", "package_version", "package_path")
@classmethod
def validate_package_field_not_blank(cls, v):
    """Reject a blank package field: it reads as "any" but would match nothing."""
    if v is not None and not v.strip():
        raise ValueError(
            "package_name, package_version and package_path must be omitted "
            "rather than left blank"
        )
    return v

validate_symbol(v) classmethod

Reject a symbol that is not a dotted qualified name.

A malformed name could never equal a parsed definition's name, so the entry would silently match nothing.

Source code in automated_security_helper/models/core.py

@field_validator("symbol")
@classmethod
def validate_symbol(cls, v):
    """Reject a symbol that is not a dotted qualified name.

    A malformed name could never equal a parsed definition's name, so the
    entry would silently match nothing.
    """
    if v is not None and not is_valid_symbol(v):
        raise ValueError(
            f"symbol must be a dotted qualified name such as "
            f"'MyClass.my_method' (identifiers joined by '.', no spaces, "
            f"wildcards or call syntax): {v!r}"
        )
    return v

validate_expiration_date(v) classmethod

Validate that expiration date is in YYYY-MM-DD format.

Past dates are accepted; use is_expired to check whether the suppression has expired at runtime.

Source code in automated_security_helper/models/core.py

@field_validator("expiration")
@classmethod
def validate_expiration_date(cls, v):
    """Validate that expiration date is in YYYY-MM-DD format.

    Past dates are accepted; use is_expired to check whether the
    suppression has expired at runtime.
    """
    if v is not None:
        try:
            datetime.strptime(v, "%Y-%m-%d")
        except ValueError:
            raise ValueError(f"Invalid expiration date format. Use YYYY-MM-DD: {v}")
    return v

matches(finding, symbol_resolver=None)

Return True if finding is covered by this suppression rule.

Checks rule_id (exact or glob), path (supports **), optional line range overlap, the optional package fields, and the optional symbol. Expired suppressions never match.

Each package field that is set must match the finding's corresponding field. A finding that does not carry that field does not match: the scanner could not say which package it is about, so a package-scoped suppression must not assume it is the one intended.

A symbol is checked last, so a file is parsed only for a finding that every other field already matches. It needs symbol_resolver, which reads the scanned source; without one a symbol-scoped suppression matches nothing, and so does a finding with no line number or in a file the resolver cannot parse.

Source code in automated_security_helper/models/core.py

def matches(
    self,
    finding: "FlatVulnerability",
    symbol_resolver: "SymbolResolver | None" = None,
) -> bool:
    """Return True if ``finding`` is covered by this suppression rule.

    Checks rule_id (exact or glob), path (supports ``**``), optional line
    range overlap, the optional package fields, and the optional symbol.
    Expired suppressions never match.

    Each package field that is set must match the finding's corresponding
    field. A finding that does not carry that field does not match: the
    scanner could not say which package it is about, so a package-scoped
    suppression must not assume it is the one intended.

    A symbol is checked last, so a file is parsed only for a finding that
    every other field already matches. It needs ``symbol_resolver``, which
    reads the scanned source; without one a symbol-scoped suppression
    matches nothing, and so does a finding with no line number or in a
    file the resolver cannot parse.
    """
    if self.is_expired:
        return False

    if self.rule_id:
        if finding.rule_id is None:
            return False
        # Case-insensitive glob match for OS portability
        if not fnmatch.fnmatch(finding.rule_id.lower(), self.rule_id.lower()):
            return False

    if not _path_pattern_matches(finding.file_path, self.path):
        return False

    if not self._line_range_matches(finding):
        return False

    if not self._package_matches(finding):
        return False

    if self.symbol is not None:
        if symbol_resolver is None:
            return False
        return symbol_resolver.contains(
            finding.file_path, finding.line_start, finding.line_end, self.symbol
        )

    return True

suppression_id(suppression)

Identifier for a suppression given as a dict of its fields.

Shared by AshSuppression.id and the config linter, which reads raw YAML dicts, so the unused-suppressions report and the linter's fixer agree.

Source code in automated_security_helper/models/core.py

def suppression_id(suppression: Dict[str, Any]) -> str:
    """Identifier for a suppression given as a dict of its fields.

    Shared by ``AshSuppression.id`` and the config linter, which reads raw YAML
    dicts, so the unused-suppressions report and the linter's fixer agree.
    """
    line_start = suppression.get("line_start")
    line_end = suppression.get("line_end")
    line_end_val = line_end if line_end is not None else line_start
    parts = [
        suppression.get("path") or "",
        suppression.get("rule_id") or "*",
        str(line_start) if line_start is not None else "*",
        str(line_end_val) if line_end_val is not None else "*",
    ]
    symbol = suppression.get("symbol")
    if symbol or any(suppression.get(f) for f in PACKAGE_SUPPRESSION_FIELDS):
        parts.append(
            "@".join(suppression.get(f) or "*" for f in PACKAGE_SUPPRESSION_FIELDS)
        )
    # A symbol is a sixth part, and its presence always brings the fifth, so an
    # id's part count says which fields it carries and a symbol can never be
    # read as a package.
    if symbol:
        parts.append(symbol)
    return "|".join(parts)