Skip to content

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

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)

True when observed is non-empty and none of its scanners reached a verdict.

Non-empty is load-bearing and is not the same assertion. An empty scanner set means the scan phase recorded nothing, which is reachable from a legitimate --phases convert run and is refused at the CI boundary instead (see assert_scanners_completed.py, which fails a results file reporting no scanners at all). Folding the two together here would turn a phase-limited run into an error.

Source code in automated_security_helper/interactions/run_ash_scan.py
def no_scanner_ran(observed: List[tuple[str, str]]) -> bool:
    """True when *observed* is non-empty and none of its scanners reached a verdict.

    Non-empty is load-bearing and is not the same assertion. An empty scanner set
    means the scan phase recorded nothing, which is reachable from a legitimate
    ``--phases convert`` run and is refused at the CI boundary instead (see
    ``assert_scanners_completed.py``, which fails a results file reporting no
    scanners at all). Folding the two together here would turn a phase-limited run
    into an error.
    """
    if not observed:
        return False
    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 []

    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((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((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(
            (
                metric.scanner_name,
                f"{metric.status} ({failed} of {attempted} targets unevaluated)",
            )
        )
    return listed

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)

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

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, *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,
    *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,
    )

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

    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.
        _incomplete = (
            incomplete_scanners(results)
            if _resolve_fail_on_incomplete_scanners(
                results, opts, config_fail_on_incomplete_scanners
            )
            else []
        )
        if _incomplete:
            # "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)} selected scanner(s) did not evaluate "
                "everything they were given[/bold red]"
            )
            for _name, _status in _incomplete:
                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]"
            )
        else:
            print(
                "[bold red]ERROR (1) Exiting due to exception during ASH scan[/bold red]"
            )

    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

    @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("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.
        """
        line_end_val = (
            self.line_end if self.line_end is not None else self.line_start
        )
        parts = [
            self.path,
            self.rule_id or "*",
            str(self.line_start) if self.line_start is not None else "*",
            str(line_end_val) if line_end_val is not None else "*",
        ]
        return "|".join(parts)

    @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") -> bool:
        """Return True if ``finding`` is covered by this suppression rule.

        Checks rule_id (exact or glob), path (supports ``**``), and optional
        line range overlap. Expired suppressions never match.
        """
        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

        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.

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

Return True if finding is covered by this suppression rule.

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

Source code in automated_security_helper/models/core.py
def matches(self, finding: "FlatVulnerability") -> bool:
    """Return True if ``finding`` is covered by this suppression rule.

    Checks rule_id (exact or glob), path (supports ``**``), and optional
    line range overlap. Expired suppressions never match.
    """
    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

    return True