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
    ignore_suppressions: bool = False
    min_severity: str = "low"
    changed_files_only: bool = False
    base_ref: str = "origin/main"
    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

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, ignore_suppressions=False, min_severity='low', changed_files_only=False, base_ref='origin/main', 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,
    ignore_suppressions: bool = False,
    min_severity: str = "low",
    changed_files_only: bool = False,
    base_ref: str = "origin/main",
    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,
        ignore_suppressions=ignore_suppressions,
        min_severity=min_severity,
        changed_files_only=changed_files_only,
        base_ref=base_ref,
        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)
    results: Optional[AshAggregatedResults]
    if opts.mode == RunMode.container:
        results = _run_container_mode(opts, logger, resolved_fail_on_findings=config_fail_on_findings)
    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)

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