API reference#

pw_fortifier: Tools and frameworks for defensive development

Pipeline framework#

pw_fortifier.pipeline_stage#

Defines the PipelineStage base class for the pw_fortifier pipeline.

class pw_fortifier.pipeline_stage.BasicPipelineStage(name: str | None = None)#

Basic interface representing a stage in a data processing pipeline.

__init__(name: str | None = None) → None#

async configure(args: Namespace) → None#

Configures the stage.

Parameters:: args – Command-line arguments namespace.

property dst_repo: WritableGitWorkspace | None#

Returns the destination Git repository workspace.

Returns:: WritableGitWorkspace instance or None.

property name: str#

Returns the stage’s name, derived from the class name.

Returns:: The stage name string.

abstract async run() → None#: Runs the main processing loop.

property skip_setup: bool#: Whether to skip the setup phase in run().

property src_repo: ReadOnlyGitWorkspace | None#

Returns the source Git repository workspace.

Returns:: ReadOnlyGitWorkspace instance or None.

class pw_fortifier.pipeline_stage.PipelineConsumerMixin#

A consumer mixin that receives paths from an input queue.

__init__() → None#: Initializes the consumer with an input queue.

property input_queue: Queue[AsyncPath | None]#

Returns the input queue of this consumer.

Returns:: The input Queue instance.

class pw_fortifier.pipeline_stage.PipelineConsumerStage(name: str | None = None)#

Pipeline stage that consumes items.

__init__(name: str | None = None) → None#: Initializes the pipeline stage with default queues and folders.

async configure(args: Namespace) → None#

Configures the stage’s name and directories.

Parameters:: args – The command-line arguments containing configuration.

async run() → None#: Runs the stage’s main processing loop.

class pw_fortifier.pipeline_stage.PipelineDemux(name: str | None = None)#

Demuxes input from multiple producers to a single output.

__init__(name: str | None = None) → None#: Initializes the demux with empty fanin.

add_stage(prev_stage: PipelineProducerMixin) → None#

Adds a previous stage to the demux.

Parameters:: prev_stage – The producer stage to add.

property fanin: list[PipelineProducerMixin]#

Returns the list of previous stages fanning in.

Returns:: List of producer stages.

class pw_fortifier.pipeline_stage.PipelineMux(name: str | None = None)#

Muxes input from a single consumer to multiple consumers.

__init__(name: str | None = None) → None#: Initializes the mux with an empty fanout mapping.

add_stage(target: str, next_stage: PipelineConsumerMixin) → None#

Maps a target basename to a next stage.

Parameters:

target – Basename of the file to match.
next_stage – Consumer stage to forward the file to.

property strict_matching: bool#: Whether to raise an error if a target stage is not found.

property targets: list[str]#

Returns the registered target names.

Returns:: List of target names string.

class pw_fortifier.pipeline_stage.PipelineProducerMixin#

A producer mixin that sends paths to a connected consumer.

__init__() → None#: Initializes the producer with no connected output queue.

async close() → None#: Closes the producer by sending the None sentinel.

connect(next_stage: PipelineConsumerMixin) → None#

Connects this producer to a consumer.

Parameters:: next_stage – The consumer stage to connect to.

send(path: AsyncPath) → None#

Sends a path to the connected consumer.

Parameters:: path – The AsyncPath to send downstream.

class pw_fortifier.pipeline_stage.PipelineProducerStage(name: str | None = None)#

Pipeline stage that produces items.

__init__(name: str | None = None) → None#: Initializes the pipeline stage with default queues and folders.

async configure(args: Namespace) → None#

Configures the stage’s name and directories.

Parameters:: args – The command-line arguments containing configuration.

async run() → None#: Runs the stage’s main processing loop.

class pw_fortifier.pipeline_stage.PipelineSink#: Pipeline consumer that acts as a sink, discarding received items.

class pw_fortifier.pipeline_stage.PipelineStage(name: str | None = None)#

Base class for all stages in the defect finding and fixing pipeline.

__init__(name: str | None = None) → None#: Initializes the pipeline stage with default queues and folders.

async configure(args: Namespace) → None#

Configures the stage’s name and directories.

Parameters:: args – The command-line arguments containing configuration.

async run() → None#: Runs the stage’s main processing loop.

class pw_fortifier.pipeline_stage.PipelineStageStub(should_forward: bool = True, should_fail: bool = False)#

Stub PipelineStage that allows choosing whether to forward or drop.

__init__(should_forward: bool = True, should_fail: bool = False) → None#

Initializes the stub with configuration options.

Parameters:

should_forward – If True, forward outputs to next stage.
should_fail – If True, raise error during processing.

pw_fortifier.code_editor#

Defines the CodeEditor base class for pipeline stages modifying code.

class pw_fortifier.code_editor.CodeEditor(name: str = 'code_editor')#

Base class for pipeline stages that generate and upload code edits.

__init__(name: str = 'code_editor') → None#: Initializes the pipeline stage with default queues and folders.

async configure(args: Namespace) → None#

Configures the stage with command-line arguments.

Parameters:: args – Command-line arguments namespace object.

pw_fortifier.scanner#

Defines the generic Scanner base class for the pw_fortifier pipeline.

class pw_fortifier.scanner.Scanner(name: str)#

Generic scanner that orchestrates a pipeline of stages.

__init__(name: str) → None#

Initializes the Scanner with a program name.

Parameters:: name – Name of the scanner program.

async run(*cli_args) → None#

Creates and runs the pipeline stages concurrently.

Parameters:: *cli_args – Command-line arguments to parse.

async pw_fortifier.scanner.configure_stage_for_test(stage: BasicPipelineStage, **kwargs) → None#

Configures a pipeline stage with mock settings for testing.

Parameters:

stage – The pipeline stage to configure.
**kwargs – Configuration arguments such as working_dir.

pw_fortifier.collector#

Defines the generic Collector stage.

class pw_fortifier.collector.Collector#

Final stage in the pipeline that collects and summarizes outputs.

__init__() → None#: Initializes the collector.

async configure(args: Namespace) → None#

Configures the stage.

Parameters:: args – Command-line arguments namespace object.

pw_fortifier.deduplicator#

Defines the Deduplicator base class for filtering duplicate issues.

class pw_fortifier.deduplicator.Deduplicator#

Base class for stages that filter out duplicate defect findings.

ISSUE_TYPE#: alias of Issue

__init__() → None#: Initializes the pipeline stage with default queues and folders.

async configure(args: Namespace) → None#

Configures the stage and duplicates directory.

Parameters:: args – Command-line arguments namespace object.

property issue_tracker: IssueTracker#: The issue tracker client.

class pw_fortifier.deduplicator.DeduplicatorStub(is_duplicate_val: bool = False)#

A stub implementation of Deduplicator for testing.

ISSUE_TYPE#: alias of Issue

__init__(is_duplicate_val: bool = False) → None#

Initializes stub with configurable duplicate check return value.

Parameters:: is_duplicate_val – If True, duplicate check returns stubbed ID.

pw_fortifier.emitter#

Defines the Emitter pipeline stage for producing paths to analyze.

class pw_fortifier.emitter.Emitter(enumerator: PathEnumerator | None = None)#

First stage in the pipeline that enumerates files to be scanned.

__init__(enumerator: PathEnumerator | None = None) → None#

Initializes the emitter with an optional path enumerator.

Parameters:: enumerator – Optional PathEnumerator instance to wrap.

async configure(args: Namespace) → None#

Configures the stage.

Parameters:: args – Command-line arguments namespace object.

property enumerator: PathEnumerator#

Returns the path enumerator.

Returns:: The PathEnumerator instance.

pw_fortifier.triager#

Defines the Triager base class for assessing issue severity.

class pw_fortifier.triager.Triager(name: str | None = None)#

Base class for stages that assess the severity and details of issues.

ISSUE_TYPE#: alias of Issue

class pw_fortifier.triager.TriagerStub(stub_issue: Issue)#

A stub implementation of Triager for testing.

ISSUE_TYPE#: alias of Issue

__init__(stub_issue: Issue) → None#

Initializes the stub with a mock Issue to apply.

Parameters:: stub_issue – Mock Issue instance containing values to apply.

pw_fortifier.path_enumerator#

Defines the PathEnumerator types for enumerating paths in a project.

class pw_fortifier.path_enumerator.FakePathEnumerator(files: list[Path], src_repo: ReadOnlyGitWorkspace | None = None)#

A fake PathEnumerator that returns predefined paths.

__init__( files: list[Path], src_repo: ReadOnlyGitWorkspace | None = None, ) → None#

Initializes the fake enumerator with a list of files.

Parameters:

files – List of Path objects to return.
src_repo – Optional ReadOnlyGitWorkspace instance.

match(pattern: str | PathLike[str]) → None#

Adds a target pattern to match.

Parameters:: pattern – Pattern string or Path object.

class pw_fortifier.path_enumerator.PathEnumerator(src_repo: ReadOnlyGitWorkspace, include_files: bool = True, include_dirs: bool = False)#

Enumerates files in a Git repository matching one or more patterns.

__init__( src_repo: ReadOnlyGitWorkspace, include_files: bool = True, include_dirs: bool = False, ) → None#

Initializes the enumerator with a project repository workspace.

Parameters:

src_repo – The ReadOnlyGitWorkspace of the project repository.
include_files – If True, enumerated paths will include files.
include_dirs – If True, enumerated paths will include directories.

match(pattern: str | PathLike[str]) → None#

Adds the given pattern to those that will be yielded.

Parameters:: pattern – Root-relative path, may contain wildcards.

property src_repo: ReadOnlyGitWorkspace#

Returns the project repository workspace.

Returns:: The ReadOnlyGitWorkspace instance.

pw_fortifier.async_path#

Defines AsyncPath, an asynchronous wrapper around pathlib.Path.

class pw_fortifier.async_path.AsyncPath(*args: str | Path | AsyncPath)#

An asynchronous wrapper around pathlib.Path using asyncio executors.

AsyncPath wraps an underlying pathlib.Path instance and forwards operations:

Pure path operations: Pure path properties (name, stem, suffix, parent, parents, parts, etc.) and pure path methods (resolve, with_name, with_suffix, relative_to, / operator, etc.) operate synchronously and return wrapped AsyncPath instances where applicable.
Asynchronous I/O operations: File system operations (read_text, write_text, read_bytes, write_bytes, exists, is_file, is_dir, stat, mkdir, unlink, rename, etc.) are executed on an asyncio executor thread and must be awaited.
Iterables / generators: Directory iteration and globbing methods (iterdir, glob, rglob, walk) return awaitable async iterables that can be consumed with async for or awaited directly into a list.
PathLike & ordering: AsyncPath implements os.PathLike[str] via os.fspath and supports rich comparisons and sorting.

Basic use examples (corresponding to pathlib.Path basic use):

Importing the main class:

>>> from pw_fortifier.async_path import AsyncPath

Listing subdirectories:

>>> p = AsyncPath('.')
>>> [x async for x in p.iterdir() if await x.is_dir()]
[AsyncPath(PosixPath('docs')), AsyncPath(PosixPath('dist')), ...]
# Or by awaiting directly:
>>> [x for x in await p.iterdir() if await x.is_dir()]

Listing Python source files in this directory tree:

>>> await p.glob('**/*.py')
[AsyncPath(PosixPath('setup.py')), ...]
# Or using async for:
>>> [x async for x in p.glob('**/*.py')]

Navigating inside a directory tree:

>>> p = AsyncPath('/etc')
>>> q = p / 'init.d' / 'reboot'
>>> q
AsyncPath(PosixPath('/etc/init.d/reboot'))
>>> q.resolve()
AsyncPath(PosixPath('/etc/rc.d/init.d/halt'))

Querying path properties:

>>> await q.exists()
True
>>> await q.is_dir()
False

Reading file contents:

>>> await q.read_text()
'#!/bin/bash\n'

__init__(*args: str | Path | AsyncPath) → None#

Initializes AsyncPath from strings, Paths, or other AsyncPaths.

Parameters:: *args – Path components to join into a single path.

path: Path#: The underlying pathlib.Path instance.

pw_fortifier.code_snippet#

Defines the CodeSnippet data structure for representing code locations.

class pw_fortifier.code_snippet.CodeSnippet(file: str | PathLike[str], lines: tuple[int, int] | None = None)#

Represents a file or range of lines in a file containing code.

file: str | PathLike[str]#: Alias for field number 0

lines: tuple[int, int] | None#: Alias for field number 1

pw_fortifier.code_snippet.find_location( root_path: str | PathLike[str], file_path: str | PathLike[str], *args: str | Pattern, ) → CodeSnippet#

Creates a CodeSnippet for file_path with relative path and lines.

Parameters:

root_path – Path to the repository root.
file_path – Path to the target file (relative or absolute).
*args – Substrings or Patterns to search for in lines.

Returns:

CodeSnippet instance.

pw_fortifier.issue#

Defines the Issue base class for pipeline findings.

class pw_fortifier.issue.Issue( *, issue_id: int | None = None, title: str | None = None, description: str | None = None, comments: list[str] = <factory>, priority: int = 4, severity: int = 4, assignee: str | None = None, cl_num: int | None = None, location: ~pw_fortifier.code_snippet.CodeSnippet | None = None, )#

A structured report representing a finding in the codebase.

__init__( *, issue_id: int | None = None, title: str | None = None, description: str | None = None, comments: list[str] = <factory>, priority: int = 4, severity: int = 4, assignee: str | None = None, cl_num: int | None = None, location: ~pw_fortifier.code_snippet.CodeSnippet | None = None, ) → None#

classmethod from_issue(issue: Issue) → _IssueT | None#: Converts an untyped tracker Issue into this Issue subtype.

async classmethod load(path: AsyncPath) → _IssueT#

Loads and reconstructs an Issue from a JSON file.

Parameters:: path – The file path to load the JSON report from.
Returns:: An Issue instance populated with the loaded data.

async save(path: AsyncPath) → None#

Serializes and saves the Issue to a JSON file.

Parameters:: path – The file path to write the JSON report to.

pw_fortifier.issue_tracker#

Defines IssueReader base class for reading issue reports.

class pw_fortifier.issue_tracker.IssueReader(issue_tracker: IssueTracker)#

Base class for stages that read issue reports from an issue tracker.

ISSUE_TYPE#: alias of Issue

__init__(issue_tracker: IssueTracker) → None#: Initializes the pipeline stage with default queues and folders.

async configure(args: Namespace) → None#

Configures the stage with issue IDs and hotlist IDs from args.

Parameters:: args – Command-line arguments namespace object.

class pw_fortifier.issue_tracker.IssueTracker#

Abstract base class for interacting with an issue tracker.

ISSUE_TYPE#: alias of Issue

__init__() → None#

property ccs: list[str]#: List of default CC email addresses.

property component_id: int#: The Buganizer component ID.

abstract async create(issue: Issue) → Issue#

Creates a new issue report in Buganizer.

Parameters:: issue – The Issue finding to file.
Returns:: The created Issue with populated issue ID.

property default_assignee: str | None#: Default assignee email address if no owner is found.

property extra_hotlist_ids: list[int]#: Additional hotlist IDs attached upon filing.

property hotlist_ids: list[int]#: All configured hotlist IDs (primary + extra).

property primary_hotlist_id: int | None#: Primary hotlist ID for deduplication and tracking.

abstract async read(issue_id: int) → Issue#

Reads an issue by ID from Buganizer.

Parameters:: issue_id – The ID of the issue to read.
Returns:: The loaded Issue.

abstract read_hotlist(hotlist_id: int) → AsyncIterator[Issue]#

Yields issues associated with a Buganizer hotlist ID.

Parameters:: hotlist_id – The hotlist ID to query.
Yields:: Issue instances matching the hotlist query.

class pw_fortifier.issue_tracker.IssueTrackerStub#

In-memory stub implementation of IssueTracker for testing.

__init__() → None#

add_issue(issue: Issue, hotlist_ids: list[int] | None = None) → None#

Adds an existing issue to the in-memory tracker.

Parameters:

issue – The issue to store.
hotlist_ids – Optional list of hotlist IDs associated with the issue.

async create(issue: Issue) → Issue#

Creates and stores a new issue with an auto-incremented ID.

Parameters:: issue – The issue to create.
Returns:: The created issue with assigned ID.

async read(issue_id: int) → Issue#

Reads an issue from the in-memory store.

Parameters:: issue_id – ID of the issue to read.
Returns:: The retrieved Issue instance.

async read_hotlist(hotlist_id: int) → AsyncIterator[Issue]#

Yields issues from the in-memory store matching the hotlist ID.

Parameters:: hotlist_id – Hotlist ID to search for.
Yields:: Matching Issue instances.

class pw_fortifier.issue_tracker.IssueWriter(issue_tracker: IssueTracker)#

Base class for stages that write issue reports.

__init__(issue_tracker: IssueTracker) → None#: Initializes the pipeline stage with default queues and folders.

async configure(args: Namespace) → None#

Configures the stage.

Parameters:: args – Command-line arguments namespace object.

Security defect scanning#

pw_fortifier.defect_scanner#

Defines the DefectScanner for the pw_fortifier pipeline.

class pw_fortifier.defect_scanner.DefectCollector#

Collector implementation that formats and prints defects.

__init__() → None#: Initializes the defect collector columns.

class pw_fortifier.defect_scanner.DefectScanner(name: str)#

DefectScanner that parses command-line arguments and runs pipeline.

__init__(name: str) → None#

Initializes the DefectScanner.

Parameters:: name – Program name string.

property code_generator: PipelineStage#: The code generator pipeline stage.

property deduplicator: Deduplicator#: The deduplicator pipeline stage.

property issue_tracker: IssueTracker#: The issue tracker client.

property triager: Triager#: The triager pipeline stage.

pw_fortifier.defect#

Defines the Defect class for representing security/freshness defects.

class pw_fortifier.defect.Defect( *, issue_id: int | None = None, title: str | None = None, description: str | None = None, comments: list[str] = <factory>, priority: int = 4, severity: int = 4, assignee: str | None = None, cl_num: int | None = None, location: ~pw_fortifier.code_snippet.CodeSnippet | None = None, affected_files: list[str] = <factory>, )#

A structured report representing a defect found in the codebase.

__init__( *, issue_id: int | None = None, title: str | None = None, description: str | None = None, comments: list[str] = <factory>, priority: int = 4, severity: int = 4, assignee: str | None = None, cl_num: int | None = None, location: ~pw_fortifier.code_snippet.CodeSnippet | None = None, affected_files: list[str] = <factory>, ) → None#

property filename: str#: Convenience property returning the primary file path.

classmethod from_issue(issue: Issue) → Defect | None#

Converts a base Issue into a Defect.

Parameters:: issue – The Issue report fetched from the tracker.
Returns:: The converted Defect instance, or None if invalid.

pw_fortifier.code_analyzer#

Defines the CodeAnalyzer base class and helper types for scanning files.

class pw_fortifier.code_analyzer.CodeAnalyzer#

Base class for stages that analyze codebase files for defects.

__init__() → None#: Initializes the analyzer.

class pw_fortifier.code_analyzer.CodeAnalyzerStub#

A stub implementation of CodeAnalyzer for testing.

__init__() → None#: Initializes the stub.

pw_fortifier.critic#

Defines the Critic base class for validating defect findings.

class pw_fortifier.critic.Critic(name: str | None = None)#: Base class for stages that validate and challenge defect findings.

class pw_fortifier.critic.CriticStub(criticize_val: bool = True)#

A stub implementation of Critic for testing.

__init__(criticize_val: bool = True) → None#

Initializes the stub with a configurable validation return value.

Parameters:: criticize_val – If True, validate and return defect.

pw_fortifier.defect_tracker#

Defines issue tracker adapter classes for defect issues.

class pw_fortifier.defect_tracker.DefectReader(issue_tracker: IssueTracker)#

Stage that reads defect reports from an issue tracker.

ISSUE_TYPE#: alias of Defect

class pw_fortifier.defect_tracker.DefectSummarizer#

Abstract base class for defect summaries and vulnerability codes.

abstract summarize(defect: Defect) → tuple[str, str]#

Returns a vulnerability code and brief summary for a defect.

Parameters:: defect – The defect finding to summarize.
Returns:: A tuple of (vuln_code, summary).

class pw_fortifier.defect_tracker.DefectWriter(issue_tracker: IssueTracker, summarizer: DefectSummarizer)#

Stage that writes defect reports to an issue tracker.

__init__( issue_tracker: IssueTracker, summarizer: DefectSummarizer, ) → None#

Initializes the defect writer.

Parameters:

issue_tracker – Issue tracker client used to write issues.
summarizer – DefectSummarizer used to format defect titles.

make_title(defect: Defect) → str#

Generates a title string for a defect.

Parameters:: defect – The defect finding to create a title for.
Returns:: Formatted title string.

pw_fortifier.poc_and_fix_generator#

Defines the PocAndFixGenerator base class and stub.

class pw_fortifier.poc_and_fix_generator.PocAndFixGenerator(name: str = 'poc_and_fix_generator')#

Base class for stages generating Proof of Concept tests and fixes.

__init__(name: str = 'poc_and_fix_generator') → None#: Initializes the pipeline stage with default queues and folders.

class pw_fortifier.poc_and_fix_generator.PocAndFixGeneratorStub( poc_result: str | bool | None = '//pw_defect:poc_test', fix_result: bool = True, summary: tuple[str, str] = ('Commit Title', 'Commit Description'), name: str = 'poc_and_fix_generator_stub', )#

A stub implementation of PocAndFixGenerator for testing.

__init__( poc_result: str | bool | None = '//pw_defect:poc_test', fix_result: bool = True, summary: tuple[str, str] = ('Commit Title', 'Commit Description'), name: str = 'poc_and_fix_generator_stub', ) → None#: Initializes the pipeline stage with default queues and folders.

Dependency freshness scanning#

pw_fortifier.freshness_scanner#

Defines the FreshnessScanner for orchestrating package freshness scans.

class pw_fortifier.freshness_scanner.FreshnessScanner(name: str)#

Scanner that orchestrates third-party package freshness scans.

__init__(name: str) → None#

Initializes the FreshnessScanner with mux and demux stages.

Parameters:: name – Program name string.

property code_generator: PipelineStage#: The code generator pipeline stage.

property issue_tracker: IssueTracker#: The issue tracker client.

register( analyzer: PackageAnalyzer, updater: PackageUpdater | None = None, ) → None#

Registers the given analyzer and optional package updater.

Parameters:

analyzer – Produces zero or more freshness scan results.
updater – Optional PackageUpdater to register with RollGenerator.

pw_fortifier.freshness_result#

Defines the FreshnessResult and PackageVersion classes.

class pw_fortifier.freshness_result.FreshnessResult( *, issue_id: int | None = None, title: str | None = None, description: str | None = None, comments: list[str] = <factory>, priority: int = 4, severity: int = 4, assignee: str | None = None, cl_num: int | None = None, location: ~pw_fortifier.code_snippet.CodeSnippet | None = None, package: str, pkg_type: str, current: ~pw_fortifier.freshness_result.PackageVersion, earliest: ~pw_fortifier.freshness_result.PackageVersion, tier: int, )#

The result of scanning a single package dependency for updates.

__init__( *, issue_id: int | None = None, title: str | None = None, description: str | None = None, comments: list[str] = <factory>, priority: int = 4, severity: int = 4, assignee: str | None = None, cl_num: int | None = None, location: ~pw_fortifier.code_snippet.CodeSnippet | None = None, package: str, pkg_type: str, current: ~pw_fortifier.freshness_result.PackageVersion, earliest: ~pw_fortifier.freshness_result.PackageVersion, tier: int, ) → None#

classmethod from_issue(issue: Issue) → FreshnessResult | None#

Converts an Issue from the tracker into a FreshnessResult.

Parameters:: issue – The Issue report fetched from the tracker.
Returns:: The converted FreshnessResult instance, or None if invalid.

property source: str#: Convenience property returning the manifest file path.

class pw_fortifier.freshness_result.PackageVersion(version: str, timestamp: date)#

Represents a specific version of a package with its release timestamp.

timestamp: date#: The release date of this version.

version: str#

abc’).

Type:: The version string (e.g., ‘1.2.3’, ‘git_revision

pw_fortifier.freshness_result.get_display_version(version: str | PackageVersion) → str#

Returns a human-readable display string for a version.

Strips prefixes like ‘git_revisions:’, ‘git_revision:’, ‘g3-revision:’, and ‘version:’, and abbreviates 40-character commit hashes to 9 characters.

Parameters:: version – PackageVersion or raw version string.
Returns:: Human-readable abbreviated version string.

pw_fortifier.freshness_result.get_tier_description(tier: int) → str#

Returns a descriptive string for a dependency classification tier.

Parameters:: tier – Integer tier level (0-3).
Returns:: Human-readable description string of the tier.

pw_fortifier.freshness_result_tracker#

Defines issue tracker adapter classes for freshness result issues.

class pw_fortifier.freshness_result_tracker.FreshnessResultReader(issue_tracker: IssueTracker)#

Stage that reads freshness result reports from an issue tracker.

ISSUE_TYPE#: alias of FreshnessResult

class pw_fortifier.freshness_result_tracker.FreshnessResultWriter(issue_tracker: IssueTracker)#

Stage that writes freshness result reports to an issue tracker.

classmethod make_description(result: FreshnessResult) → str#

Generates an issue description for the stale package finding.

Parameters:: result – The FreshnessResult finding.
Returns:: Formatted description string describing the freshness finding.

classmethod make_title(result: FreshnessResult) → str#

Generates an issue title for the stale package finding.

Parameters:: result – The FreshnessResult finding.
Returns:: Formatted title string describing the stale version and package.

pw_fortifier.freshness_collector#

Defines the FreshnessResultCollector for freshness scan results.

class pw_fortifier.freshness_collector.FreshnessResultCollector#

Collector that formats and prints freshness results.

__init__() → None#: Initializes column specifications for freshness results.

pw_fortifier.freshness_deduplicator#

Defines FreshnessDeduplicator for deduplicating freshness results.

class pw_fortifier.freshness_deduplicator.FreshnessDeduplicator#

Deduplicator implementation for assessing freshness results.

ISSUE_TYPE#: alias of FreshnessResult

__init__() → None#: Initializes the freshness deduplicator.

property issue_tracker: IssueTracker#: The issue tracker client.

pw_fortifier.freshness_triager#

Defines the FreshnessTriager class for assessing freshness results.

class pw_fortifier.freshness_triager.FreshnessTriager(name: str | None = None)#

Triager implementation for assessing freshness results.

ISSUE_TYPE#: alias of FreshnessResult

pw_fortifier.package_analyzer#

Utility to scan packages for freshness.

class pw_fortifier.package_analyzer.PackageAnalyzer#

Base class for package analyzers.

__init__() → None#: Initializes the package analyzer.

clean_version(ver_str: str) → str#

Extracts a clean X.Y.Z version from a semver range.

Parameters:: ver_str – Raw version string to clean.
Returns:: Cleaned version string.

find_lowest( tier: int, current: PackageVersion, versions: list[PackageVersion], ) → PackageVersion#

Returns the lowest version that is >= current and is fresh.

Parameters:

tier – Classification tier integer.
current – Currently used PackageVersion.
versions – List of available PackageVersion candidates.

Returns:

Lowest fresh PackageVersion.

major_version(ver_str: str) → str | None#

Extracts the major version from a version string.

Parameters:: ver_str – Raw version string.
Returns:: Major version string or None.

parse_semver(ver_str: str) → SemVer | None#: Parses a version string into a SemVer named tuple.

property root: str#

Returns the root directory.

The configure method must be called before getting this property.

Returns:: Root directory path string.

property skip_setup: bool#: Whether to skip the setup phase in run().

class pw_fortifier.package_analyzer.PackageAnalyzerStub#

Stub package analyzer for testing registry.

__init__() → None#: Initializes fake scanner.

pw_fortifier.package_analyzer.find_lowest( tier: int, current: PackageVersion, versions: list[PackageVersion], loose_semver: bool = False, ) → PackageVersion#

Returns the lowest version that is >= current and is fresh.

Parameters:

tier – Classification tier integer.
current – Currently used PackageVersion.
versions – List of available PackageVersion candidates.
loose_semver – True if pre-release labels should be treated as extra labels.

Returns:

The lowest fresh PackageVersion.

pw_fortifier.package_updater#

Defines the PackageUpdater base class and stub.

class pw_fortifier.package_updater.PackageUpdater#

Abstract base class for package-type specific roll updaters.

abstract async update( dst_repo: WritableGitWorkspace, result: FreshnessResult, ) → bool#

Updates a dependency in dst_repo based on freshness result.

Parameters:

dst_repo – The writable git workspace to update.
result – The freshness result finding.

Returns:

True if update succeeded and passed presubmit; False otherwise.

class pw_fortifier.package_updater.PackageUpdaterStub(update_return_value: bool = True)#

Stub package updater for testing.

__init__(update_return_value: bool = True) → None#

async update( dst_repo: WritableGitWorkspace, result: FreshnessResult, ) → bool#: Simulates package update.

pw_fortifier.roll_generator#

Defines the RollGenerator base class and stub.

class pw_fortifier.roll_generator.RollGenerator(name: str = 'roll_generator')#

Base class for stages that generate dependency roll commits.

__init__(name: str = 'roll_generator') → None#: Initializes the pipeline stage with default queues and folders.

register(updater: PackageUpdater) → None#

Registers a package updater by package type or target filename.

Parameters:: updater – PackageUpdater instance to register.

class pw_fortifier.roll_generator.RollGeneratorStub(generate_return_value: bool = True, name: str = 'roll_generator_stub')#

A stub implementation of RollGenerator for testing.

__init__(generate_return_value: bool = True, name: str = 'roll_generator_stub') → None#: Initializes the pipeline stage with default queues and folders.

Package analyzers#

pw_fortifier.bazel_cipd#

Package scanner for CIPD repositories in Bazel.

class pw_fortifier.bazel_cipd.BazelCipdAnalyzer#

Scans CIPD repositories in MODULE.bazel for freshness.

__init__() → None#: Initializes the package analyzer.

async configure(args: Namespace) → None#: Configures the analyzer and determines if setup can be skipped.

pw_fortifier.bazel_dep#

Package scanner for Bazel dependencies.

class pw_fortifier.bazel_dep.BazelDepAnalyzer#

Scans Bazel dependencies in MODULE.bazel for freshness.

__init__() → None#: Initializes the package analyzer.

async configure(args: Namespace) → None#: Configures the analyzer and determines if setup can be skipped.

pw_fortifier.bazel_maven#

Package scanner for Maven dependencies in Bazel (Bzlmod).

class pw_fortifier.bazel_maven.BazelMavenAnalyzer#

Scans Maven dependencies in MODULE.bazel for freshness.

__init__() → None#: Initializes the package analyzer.

async configure(args: Namespace) → None#: Configures the analyzer and determines if setup can be skipped.

pw_fortifier.cargo#

Package scanner for cargo packages.

class pw_fortifier.cargo.CargoAnalyzer#

Scans cargo packages for freshness.

__init__() → None#: Initializes the package analyzer.

pw_fortifier.cipd_setup#

Package scanner for CIPD setup JSON files in pw_env_setup.

class pw_fortifier.cipd_setup.CipdSetupAnalyzer#

Scans CIPD setup JSON files in pw_env_setup for freshness.

__init__() → None#: Initializes the package analyzer.

async configure(args: Namespace) → None#: Configures the analyzer and determines if setup can be skipped.

pw_fortifier.copybara#

Package scanner for Copybara packages.

class pw_fortifier.copybara.CopybaraAnalyzer#: Scans Copybara packages for freshness.

pw_fortifier.go_mod#

Package scanner for Go modules in go.mod.

class pw_fortifier.go_mod.GoModAnalyzer#

Scans Go modules in go.mod for freshness.

async configure(args: Namespace) → None#: Configures the analyzer and determines if setup can be skipped.

pw_fortifier.npm#

Package scanner for npm packages.

class pw_fortifier.npm.NpmAnalyzer#

Scans npm packages for freshness.

__init__() → None#: Initializes the package analyzer.

add_on_device_module(module: str) → None#

Adds a module to be treated as TIER0_ON_DEVICE.

Parameters:: module – The module name to treat as tier 0 on-device.

pw_fortifier.pip#

Package scanner for pip packages.

class pw_fortifier.pip.PipAnalyzer#

Scans pip packages for freshness.

__init__() → None#: Initializes the package analyzer.

Utilities#

pw_fortifier.bazelisk_utils#

Base package scanner for Bazel-based dependencies.

class pw_fortifier.bazelisk_utils.BazelRepo( canonical_name: str, rule_name: str | None, location: CodeSnippet | None, attributes: list[dict], assignee: str | None = None, )#

Represents a Bazel repository and its metadata from MODULE.bazel.

assignee: str | None#: Optional legacy assignee field.

attributes: list[dict]#: List of attribute dictionaries parsed from show_repo.

canonical_name: str#: Canonical repository name string.

get_attr_str(name: str) → str#

Finds an attribute and returns stringValue or empty string.

Parameters:: name – Name of the attribute to find.
Returns:: The stringValue if found, or empty string.

get_attr_str_dict(name: str) → dict[str, str]#

Finds an attribute and returns stringDictValue as a dict.

Parameters:: name – Name of the attribute to find.
Returns:: The dictionary of attribute keys and values.

get_attr_str_list(name: str) → list[str]#

Finds an attribute and returns stringListValue or empty list.

Parameters:: name – Name of the attribute to find.
Returns:: The stringListValue if found, or an empty list.

static load( module_bazel_path: str | PathLike[str], ) → AsyncIterator[BazelRepo]#

Loads repository mappings and attributes from MODULE.bazel.

Parameters:: module_bazel_path – Path to the MODULE.bazel file to parse.
Yields:: BazelRepo instances for each discovered external repository.

location: CodeSnippet | None#: Optional location code snippet.

rule_name: str | None#: Optional repository rule name string.

async pw_fortifier.bazelisk_utils.run_bazelisk(args: list[str], cwd: str | PathLike[str]) → CompletedProcess#

Runs a bazelisk command asynchronously in an executor.

Parameters:

args – List of command-line arguments for the bazelisk command.
cwd – Working directory where the command is executed.

Returns:

A subprocess.CompletedProcess instance containing execution results.

pw_fortifier.build_utils#

Build and test utility functions for pw_fortifier.

async pw_fortifier.build_utils.run_presubmit(cwd: str | PathLike[str] = '.') → bool#

Runs presubmit checks using bazelisk.

Parameters:: cwd – Working directory where the command is executed.
Returns:: True only if the presubmit command succeeds; False otherwise.

async pw_fortifier.build_utils.run_unit_tests(cwd: str | PathLike[str] = '.', target: str | None = None) → AsyncIterator[str]#

Runs unit tests using bazelisk and yields names of failed tests.

Parameters:

cwd – Working directory where the command is executed.
target – Optional Bazel target to test. Defaults to ‘//…’.

Yields:

Names of failed test targets.

pw_fortifier.cipd_utils#

Base package scanner for CIPD-based dependencies.

class pw_fortifier.cipd_utils.CipdPackageSet(rel_path: str, pkg_type: str)#

Helper class to collect CIPD packages and produce FreshnessResults.

__init__(rel_path: str, pkg_type: str) → None#

async add( pkg: str, version: str, tier: int, platforms: list[str] | None = None, location: CodeSnippet | None = None, ) → None#

Adds a CIPD package entry to be analyzed.

Parameters:

pkg – CIPD package path or template string.
version – Target version string.
tier – Dependency tier integer.
platforms – Optional list of explicit host platform strings.
location – Optional CodeSnippet location for finding.

async generate_results( scanned_packages: set[str], ) → AsyncIterator[FreshnessResult]#

Generates FreshnessResult instances for gathered packages.

Parameters:: scanned_packages – Set of package keys to track duplicate packages.
Yields:: FreshnessResult objects for each gathered package.

pw_fortifier.find_core_owners#

Utility to find core owners (or any owners) for code snippets.

This utility analyzes git history to find the team members who have most recently and frequently modified specific files or line ranges. It filters out large-scale changes (LSCs) and restrict the search by default to core team members listed in the root OWNERS file.

class pw_fortifier.find_core_owners.CoreOwnerFinder(repo: ReadOnlyGitWorkspace)#

Finds the core assignee for a set of code snippets.

__init__(repo: ReadOnlyGitWorkspace)#

add( file: str | PathLike[str] | CodeSnippet, lines: tuple[int, int] | None = None, ) → None#

Adds a file or file range to be examined.

Parameters:

file – Path to the file or a CodeSnippet instance.
lines – Optional inclusive line range tuple (start, end).

core_members() → set[str]#

Gets core team members from the OWNERS file in the repo root.

Returns:: Set of email addresses of core team members.

find(any_owner: bool = False) → str | None#

Finds the core team member who most modified the added snippets.

Parameters:: any_owner – Whether to allow any author or only core team members.
Returns:: The chosen assignee email address, or None.

pw_fortifier.find_core_owners.find_owners(root_path: str | PathLike[str], file_path: str | PathLike[str], *args: str | Pattern) → str | None#

Finds assignee of first line containing args or matching regexes.

Parameters:

root_path – Path to the repository root.
file_path – Path to the target file relative to root.
*args – Substrings or Patterns to search for in lines.

Returns:

Assignee email string or None if not found.

pw_fortifier.find_core_owners.main(argv: list[str] | None = None) → int#

Finds core owners for code snippets.

Parameters:: argv – Optional list of command-line arguments.
Returns:: Exit code (0 on success).

pw_fortifier.find_core_owners.parse_owners_file(path: str | PathLike[str]) → list[str]#

Parses an OWNERS file and returns the list of emails found.

Parameters:: path – Path to the OWNERS file.
Returns:: List of email address strings parsed from the file.

pw_fortifier.git_utils#

Defines Git workspace helpers for managing local Git repositories.

class pw_fortifier.git_utils.BasicGitWorkspace( project_dir: Path | str | PathLike[str], repo_url: str | None = None, tmp_dir: TemporaryDirectory | None = None, )#

Base class for Git workspaces.

__init__( project_dir: Path | str | PathLike[str], repo_url: str | None = None, tmp_dir: TemporaryDirectory | None = None, ) → None#

Initializes the workspace for an already cloned repository.

Parameters:

project_dir – The local directory of the git repository.
repo_url – Optional remote repository URL.
tmp_dir – Optional TemporaryDirectory instance that owns project_dir and should be cleaned up on deletion.

property project_dir: Path#

Returns the project directory path.

Returns:: The project directory Path instance.

property repo_url: str | None#

Returns the remote repository URL, if known.

Returns:: The remote repository URL string or None.

async run_git(*args, **kwargs) → CompletedProcess#

Runs a git command in the context of the workspace directory.

Parameters:

*args – Arguments to pass to git.
**kwargs – Keyword arguments to pass to subprocess.run.

Returns:

The CompletedProcess result from running the git command.

run_git_sync(*args, **kwargs) → CompletedProcess#

Runs a git command synchronously in the workspace directory.

Parameters:

*args – Arguments to pass to git.
**kwargs – Keyword arguments to pass to subprocess.run.

Returns:

The CompletedProcess result from running the git command.

class pw_fortifier.git_utils.GitBranch(workspace: BasicGitWorkspace, name: str = 'tmp', keep: bool = False)#

Context-managed Git branch for staging, committing, and pushing.

__init__(workspace: BasicGitWorkspace, name: str = 'tmp', keep: bool = False) → None#

async add(path: str | Path | PathLike[str] = '.') → None#

Stages file or directory changes in the branch.

Parameters:: path – File or directory path to stage (defaults to ‘.’).

async commit(commit_msg: str, amend: bool = False) → None#

Commits pending changes to the local workspace branch.

Parameters:

commit_msg – The commit message.
amend – If True and branch is not pristine, amends previous commit.

async commit_msg() → list[str]#

Returns the commit message lines of the current HEAD commit.

Returns:: A list of newline-delimited output lines from git log.

async diff(staged: bool = False) → list[str]#

Returns the diff lines of the working tree or staging area.

Parameters:: staged – If True, returns diff of staged changes. If False, returns diff of unstaged working tree changes.
Returns:: A list of diff output lines.

property keep: bool#

Returns True if the branch should be preserved on context exit.

Returns:: Boolean indicating whether to keep the branch on exit.

property name: str#

Returns the branch name.

Returns:: The branch name string.

property pristine: bool#

Returns True if no commits have been made on this branch yet.

Returns:: Boolean indicating whether the branch is pristine.

async push() → int | None#

Pushes committed changes to Gerrit, returning the CL number.

Returns:: The Gerrit CL number if successfully parsed; None otherwise.

async reset(staged: bool = False) → None#

Resets the branch working tree and optionally staged changes.

Parameters:: staged – If True, removes staged changes as well, returning the branch to a pristine state. If False, removes working tree and untracked files, leaving only staged changes.

async setup() → GitBranch#: Captures active branch and checks out temporary branch.

async teardown() → None#: Resets, switches back, and deletes branch if keep is False.

class pw_fortifier.git_utils.ReadOnlyGitWorkspace( project_dir: Path | str | PathLike[str], repo_url: str | None = None, tmp_dir: TemporaryDirectory | None = None, )#

Local Git repo interface for examining files and revision history.

blame(file: str | PathLike[str], commit: str, lines: tuple[int, int] | None = None) → list[str]#

Runs git blame on a file at a specific commit.

Parameters:

file – Path to the file to blame.
commit – The commit hash to blame at.
lines – Optional (start, end) line range to blame.

Returns:

A list of blame output lines.

Asynchronously clones a repository.

Parameters:

src_url – URL of the repository to clone.
dst_dir – Optional destination directory for the clone.
timestamp – Optional timestamp for shallow-since clones.
no_checkout – If True, skip checking out working tree.
git_filter – Optional git filter argument (e.g. blob:none).
depth – Optional commit depth for shallow clone.

Returns:

A new workspace instance for the cloned repo.

Synchronously clones a repository.

Parameters:

src_url – URL of the repository to clone.
dst_dir – Optional destination directory for the clone.
timestamp – Optional timestamp for shallow-since clones.
no_checkout – If True, skip checking out working tree.
git_filter – Optional git filter argument (e.g. blob:none).
depth – Optional commit depth for shallow clone.

Returns:

A new ReadOnlyGitWorkspace instance for the cloned repo.

Synchronously queries repository history for versions.

Parameters:

num – Optional maximum number of log entries.
pattern – Optional grep pattern to filter commit messages.
scope – Optional path scope to filter log entries.

Yields:

Tuples of (commit_hash, commit_date).

log_revisions(file: str | PathLike[str], limit: int = 20) → list[str]#

Runs git log to get revisions for a file.

Parameters:

file – Path to the file.
limit – Maximum number of revisions to return.

Returns:

A list of log output lines (one per revision).

show_names(commit: str) → list[str]#

Runs git show to get files changed in a commit.

Parameters:: commit – The commit hash.
Returns:: A list of file paths changed in the commit.

class pw_fortifier.git_utils.WritableGitWorkspace( project_dir: Path | str | PathLike[str], repo_url: str | None = None, tmp_dir: TemporaryDirectory | None = None, )#

Local Git workspace with operations for modifying and pushing changes.

branch(name: str = 'tmp', keep: bool = False) → GitBranch#

Returns a context-managed GitBranch object for creating a branch.

Parameters:

name – Branch name to checkout (defaults to ‘tmp’).
keep – If True, skips cleanup on context exit.

Returns:

A GitBranch instance.

pw_fortifier.git_utils.get_git_repo_root(path: Path | str | PathLike[str] | None = None) → Path | None#

Gets the root directory of the git repository.

Parameters:: path – Optional path to a directory within the repository. Defaults to BUILD_WORKSPACE_DIRECTORY, BUILD_WORKING_DIRECTORY, or the current working directory (outside of test environments).
Returns:: The resolved Path to the repository root, or None if not in a repository.

pw_fortifier.git_utils.get_git_repo_url(path: Path | str | PathLike[str] | None = None, remote: str = 'origin') → str#

Gets the remote repository URL for a git repository.

Parameters:

path – Optional path to the repository directory. Defaults to the invocation directory or current working directory.
remote – Remote name to query (defaults to ‘origin’).

Returns:

The remote repository URL string.

pw_fortifier.git_utils.make_git_commit_msg(title: str, description: str, issue_id: int | None = None) → Iterator[str]#

Formats a git commit message from title, description, and issue ID.

Parameters:

title – The commit subject line.
description – The commit body description.
issue_id – Optional issue/bug tracker ID.

Yields:

Lines of the formatted git commit message.