Implementation guide for defect scanners#

This guide walks step-by-step through implementing a custom security defect scanner using pw_fortifier.

Overview#

Building a defect scanner consists of the following steps:

Create an IssueTracker to access Buganizer.
Implement a CodeAnalyzer to scan files for defects.
Implement a Critic to validate findings and filter false positives.
Implement a Deduplicator to prevent duplicate bugs.
Implement a Triager to assess defect severity and assign owners.
Implement a DefectSummarizer to format issue titles.
Implement a PocAndFixGenerator to produce fixes.
Wire all stages into a DefectScanner executable.
Define Build targets to run the scanner.

Note

The Pigweed team maintains implementations for the pipeline stages described below that you can quickly adapt for your project. If you are trying to build scanning tools for a Google project, reach out to us and we can help you get set up quickly!

Pipeline architecture#

The diagram below shows how defect scanning pipeline stages connect. Orange nodes represent implementer-provided components; light blue nodes represent library-provided components.

        flowchart TD
   subgraph Discovery ["Discovery & Generation"]
      Emitter --> CodeAnalyzer
      CodeAnalyzer --> Critic
      Critic --> Deduplicator
   end

   subgraph Tracking ["Triage & Tracking"]
      IssueTracker[("IssueTracker")]
      Deduplicator --> Triager
      Triager --> IssueWriter
      IssueWriter --> IssueDemux["Issue Demux"]
      IssueReader --> IssueDemux
      Deduplicator -.- IssueTracker
      IssueWriter -.- IssueTracker
      IssueReader -.- IssueTracker
   end

   subgraph Action ["Action & Output"]
      IssueDemux --> PocAndFixGenerator
      PocAndFixGenerator --> Collector
   end

   classDef implementer fill:#ffe0b2,stroke:#f57c00,stroke-width:2px;
   classDef library fill:#e1f5fe,stroke:#0288d1,stroke-width:1px;

   class CodeAnalyzer,Critic,Deduplicator implementer;
   class Triager,IssueTracker,PocAndFixGenerator implementer;
   class Emitter,IssueWriter,IssueDemux,IssueReader,Collector library;

1. IssueTracker#

The first step is to implement an IssueTracker to manage communication between the scanner and Buganizer.

Create a new file and define a subclass of IssueTracker:

from pw_fortifier.issue_tracker import IssueTracker

class MyIssueTracker(IssueTracker):

    def __init__(self) -> None:
        super().__init__()

Set your project’s Buganizer component ID (required):
```
self.component_id = 1234567
```
Set the primary hotlist ID used for deduplication and tracking (required):
```
self.primary_hotlist_id = 7654321
```
Set additional hotlists to attach to filed issues (optional):
```
self.extra_hotlist_ids = [9876543]
```
Set the default assignee email address if no owner is found (required):
```
self.default_assignee = 'sheriff@example.com'
```
Set default CC email addresses (optional):
```
self.ccs = ['team-alerts@example.com']
```

Implement create() to file new issues. This is largely a passthrough to Buganizer APIs:

async def create(self, issue: Issue) -> Issue:
    bug_id = await my_buganizer_api.create(
        component_id=self.component_id,
        title=issue.title,
        description=issue.description,
        assignee=issue.assignee or self.default_assignee,
        ccs=self.ccs,
        hotlists=self.hotlist_ids,
    )
    return issue._replace(issue_id=bug_id)

Implement read() to fetch an existing issue by ID:

async def read(self, issue_id: int) -> Issue:
    record = await my_buganizer_api.query(issue=issue_id)
    return Issue(
        issue_id=record.id,
        title=record.title,
        description=record.description,
        assignee=record.assignee,
    )

Implement read_hotlist() to stream issues matching a hotlist ID:

async def read_hotlist(self, hotlist_id: int) -> AsyncIterator[Issue]:
    records = await my_buganizer_api.query(hotlist=hotlist_id)
    for record in records:
        yield Issue(
            issue_id=record.id,
            title=record.title,
            description=record.description,
            assignee=record.assignee,
        )

For a complete working example, see pw_fortifier/py/pw_fortifier/demo_issue_tracker.py.

2. CodeAnalyzer#

The code analyzer stage inspects individual source files for vulnerabilities and writes findings to report files.

Subclass CodeAnalyzer:

from typing import AsyncIterator

from pw_fortifier.async_path import AsyncPath
from pw_fortifier.code_analyzer import CodeAnalyzer


class MyCodeAnalyzer(CodeAnalyzer):

    def _analyze(
        self, input_file: AsyncPath, out_path: AsyncPath
    ) -> AsyncIterator[AsyncPath]:

In the analysis logic, invoke your agent or scanner to inspect the input file:

async def _impl():
    findings = await my_analyzer_agent.scan(input_file)

Write the markdown findings to files under out_path and yield their paths:

for idx, finding in enumerate(findings):
    report_path = out_path / f'finding-{idx}.md'
    await report_path.write_text(finding.markdown)
    yield report_path

return _impl()

The base class automatically extracts referenced source files from the report and constructs Defect instances.

3. Critic#

The critic stage validates findings from the code analyzer to filter out false positives before filing.

Subclass Critic:

from pw_fortifier.critic import Critic
from pw_fortifier.defect import Defect


class MyCritic(Critic):

Implement _criticize() to validate the defect description using an agent or tool:

async def _criticize(self, issue: Defect) -> Defect | None:
    is_valid = await my_critic_agent.validate(issue.description)
    if not is_valid:
        return None
    return issue

Returning None drops the defect from the pipeline as a false positive.

4. Deduplicator#

The deduplicator stage checks incoming defect reports against existing historical issues to prevent duplicate filings.

Subclass Deduplicator and set ISSUE_TYPE = Defect:

from pw_fortifier.deduplicator import Deduplicator
from pw_fortifier.defect import Defect
from pw_fortifier.issue import Issue


class MyDeduplicator(Deduplicator):
    ISSUE_TYPE = Defect

Implement _is_duplicate() using self.issue_tracker to check existing Buganizer issues:

async def _is_duplicate(self, issue: Issue) -> int | None:
    assert isinstance(issue, Defect)
    duplicate_id = await my_dedup_agent.find_duplicate(
        issue, self.issue_tracker
    )
    return duplicate_id

Return the duplicate Buganizer issue ID if one exists, or None if the finding represents a new vulnerability.

5. Triager#

The triager stage assesses defect severity.

Subclass Triager and set ISSUE_TYPE = Defect:

from pw_fortifier.defect import Defect
from pw_fortifier.issue import Issue
from pw_fortifier.triager import Triager


class MyTriager(Triager):
    ISSUE_TYPE = Defect

Implement _triage() to set the defect severity in place:

async def _triage(self, issue: Issue) -> None:
    assert isinstance(issue, Defect)
    issue.severity = await my_triage_agent.assess_severity(issue)

Owner assignment is handled automatically by the base stage using CoreOwnerFinder if issue.assignee is not already populated.

6. DefectSummarizer#

The defect summarizer formats defect titles and vulnerability codes for Buganizer issues.

Subclass DefectSummarizer:

from pw_fortifier.defect import Defect
from pw_fortifier.defect_tracker import DefectSummarizer


class MyDefectSummarizer(DefectSummarizer):

Implement summarize() to return a tuple of (vuln_code, summary):

def summarize(self, defect: Defect) -> tuple[str, str]:
    vuln_code = 'UAF'
    summary = 'Use-after-free in buffer handling'
    return (vuln_code, summary)

7. PocAndFixGenerator#

The code editor stage generates a proof-of-concept unit test and a code fix for the defect.

Subclass PocAndFixGenerator:

from pw_fortifier.defect import Defect
from pw_fortifier.poc_and_fix_generator import PocAndFixGenerator


class MyPocAndFixGenerator(PocAndFixGenerator):

    def __init__(self) -> None:
        super().__init__('my_poc_and_fix_generator')

Implement _generate_poc() to create a failing test:

async def _generate_poc(self, defect: Defect) -> str | None:
    test_target = await my_fix_agent.create_poc(defect)
    return test_target

Implement _generate_fix() to apply the code fix:

async def _generate_fix(self, defect: Defect) -> bool:
    fix_ok = await my_fix_agent.create_fix(defect)
    return fix_ok

Implement _summarize() to construct the Git commit message:

async def _summarize(
    self, defect: Defect, diffs: list[str]
) -> tuple[str, str]:
    subject = f'Fix security defect in {defect.location.file}'
    body = 'Applies fix and adds regression unit test.'
    return (subject, body)

8. DefectScanner#

Assemble all stages into an executable scanner tool by subclassing DefectScanner.

Create a script (e.g. my_defect_scanner.py) and define the scanner skeleton:

import asyncio
import sys

from pw_fortifier.defect_scanner import DefectScanner

class MyDefectScanner(DefectScanner):

    def __init__(self) -> None:
        super().__init__('my_defect_scanner')

async def main() -> None:
    scanner = MyDefectScanner()
    await scanner.run(*sys.argv[1:])

if __name__ == '__main__':
    asyncio.run(main())

Set the repository URL to scan:

self.repo_url = 'sso://repo-host/my-project'

Wire up the code analyzer and critic:

self.code_analyzer = MyCodeAnalyzer()
self.critic = MyCritic()

Wire up the summarizer and issue tracker:

self.summarizer = MyDefectSummarizer()
self.issue_tracker = MyIssueTracker()

Wire up the deduplicator, triager, and code generator:

self.deduplicator = MyDeduplicator()
self.triager = MyTriager()
self.code_generator = MyPocAndFixGenerator()

9. Build targets#

Define a pw_py_binary target in BUILD.bazel to run your scanner:

Load pw_py_binary:

load("//pw_build:python.bzl", "pw_py_binary")

Define the binary target and include //pw_fortifier/py:pw_fortifier in deps:

pw_py_binary(
    name = "my_defect_scanner",
    srcs = [
        "my_defect_scanner.py",
    ],
    deps = [
        "//pw_fortifier/py:pw_fortifier",
        ":my_security_stages_lib",
        ":my_issue_tracker_lib",
    ],
)

Summary#

Implementation checklist:

IssueTracker
- [ ] Subclass IssueTracker.
- [ ] Set component_id, primary_hotlist_id, and default_assignee.
- [ ] (Optional) Set extra_hotlist_ids and ccs.
- [ ] Implement create() to file bugs in Buganizer.
- [ ] Implement read() to query bugs by ID.
- [ ] Implement read_hotlist() to stream bugs by hotlist.
CodeAnalyzer
- [ ] Subclass CodeAnalyzer.
- [ ] Implement _analyze() to inspect files and yield markdown reports.
Critic
- [ ] Subclass Critic.
- [ ] Implement _criticize() to filter out false positives.
Deduplicator
- [ ] Subclass Deduplicator with ISSUE_TYPE = Defect.
- [ ] Implement _is_duplicate() using self.issue_tracker.
Triager
- [ ] Subclass Triager with ISSUE_TYPE = Defect.
- [ ] Implement _triage() to set defect severity.
DefectSummarizer
- [ ] Subclass DefectSummarizer.
- [ ] Implement summarize() returning vulnerability code and summary.
PocAndFixGenerator
- [ ] Subclass PocAndFixGenerator.
- [ ] Implement _generate_poc() to create a failing unit test.
- [ ] Implement _generate_fix() to resolve the defect.
- [ ] Implement _summarize() to generate commit messages.
DefectScanner
- [ ] Subclass DefectScanner and call super().__init__().
- [ ] Set repo_url.
- [ ] Wire up all pipeline stages in __init__.
Build targets
- [ ] Define pw_py_binary in BUILD.bazel.
- [ ] Add //pw_fortifier/py:pw_fortifier to deps.

Testing and debugging#

Run the scanner with:

# Local dry run across the repository (no bugs filed, no CLs uploaded)
$ bazelisk run //path/to:my_defect_scanner -- -s /path/to/my/project

# Scan specific files and output a CSV report
$ bazelisk run //path/to:my_defect_scanner -- \
    -s /path/to/my/project -f "pw_sync/**" -o defects.csv

# Run a full scan, filing bugs and uploading candidate fix CLs
$ bazelisk run //path/to:my_defect_scanner -- \
    -s /path/to/my/project -b -u

Execution flags and permissions#

Scanners operate with safe defaults that do not modify remote trackers or repositories unless explicitly permitted via command-line flags:

Findings are reported locally and not filed in Buganizer unless -b,–create-bugs is specified.
Code modifications and validation builds are skipped unless -e,–allow-edits is specified.
Local Git changes are kept locally and not uploaded to Gerrit unless -u,–allow-uploads is specified.

Run resumption#

If an invocation of the scanner is interrupted, the in-flight inputs will be saved as files under stage-specific subdirectories of the working directory. Implementers may examine these files to debug assertions and exceptions. They may also resume an interrupted run via the -r,–resume command line argument.

Retry configuration#

By default, consumers will try to process each input, and retry on request or subprocess failure up to the maximum number of times specified via the -m,–max-retries command line argument. When the maximum number of retries has been reached, the input will be moved to a stage-specific “error” subdirectory of the working directory, e.g. “triager_err”. By setting the maximum number of retries to zero, i.e. -m 0, implementers can cause the scanner pipeline to fail fast and then inspect the offending inputs to debug.

Input preservation#

Each pipeline consumer has a _preserve_inputs attribute.

When _preserve_inputs=False (the default for most intermediate stages), the consumer moves incoming files from the upstream stage’s output directory into its own <stage>_in/ directory and deletes (unlinks) them once processing succeeds.
When _preserve_inputs=True, the consumer processes input files directly in place in the upstream directory without moving or deleting them.

This attribute is enabled by default on initial analysis stages (such as CodeAnalyzer and PackageAnalyzer) so that source repository files are never moved or deleted.

Stage implementers can also enable _preserve_inputs=True on their stage and the subsequent stage to debug transformations: the inputs to the stage remain in <prev_stage>_out/ and the outputs remain in <stage>_out/, allowing direct comparison of the input and output state.