prove

func ApplyMetadata ¶

func ApplyMetadata(finding schemas.VerifiedFinding) schemas.VerifiedFinding

ApplyMetadata ports `_apply_metadata`:

finding.severity = apply_cwe_severity_floor(finding.cwe_id, finding.severity)
finding.compliance = get_compliance_mappings(finding.cwe_id)
finding.exploitability_score = compute_exploitability_score(finding)
finding.sarif_security_severity = finding.exploitability_score
if not finding.sarif_rule_id:
    finding.sarif_rule_id = f"sec-af/{finding.finding_type.value}/{cwe_slug}"
return finding

ORDER IS LOAD-BEARING: the severity floor is applied BEFORE the exploitability score is computed, so a floored severity feeds the score. The rule-id backfill runs last and only when the id is falsy (empty string).

Python mutates the model in place and returns it; Go takes and returns a value. Every call site rebuilds its list from the return value, so the two are observably identical.

`get_compliance_mappings(finding.cwe_id)` is called with the default `frameworks=None`, which the Go signature spells as a nil slice.

func AssembleVerifiedFinding ¶

func AssembleVerifiedFinding(
	finding schemas.RawFinding,
	dataFlowTrace schemas.DataFlowTrace,
	sanitization schemas.SanitizationResult,
	exploit schemas.ExploitHypothesis,
	verdictDecision schemas.VerdictDecision,
) schemas.VerifiedFinding

AssembleVerifiedFinding ports assembler.py assemble_verified_finding.

Python parity notes:

• `bypass_possible=bool(sanitization.bypass_method)` is Python truthiness on an `str | None`: None AND the empty string are both False. The field is `bool | None` in the schema but assembler always supplies a real bool, so the Go pointer is always non-nil here.
• `data_flow_trace` and `data_flow_evidence.steps` are the SAME list object in Python; Go copies the slice header into both, which is equivalent for everything downstream (nobody mutates it).
• `tags=[]`, `exploitability_score=0.0` and `sarif_security_severity=0.0` are literal here — `_apply_metadata` overwrites the two scores afterwards.
• `related_locations` / `compliance` are not passed, so they take pydantic's `default_factory=list`; Go seeds them explicitly so model_dump() parity holds (`[]`, not `null`).

func ChainBuilderPrompt ¶

func ChainBuilderPrompt(potentialChains []schemas.PotentialChain, findings []schemas.VerifiedFinding, depth string) string

ChainBuilderPrompt builds the exact prompt RunChainBuilder sends. There is no trailing CONTEXT block — chain_builder.py sends the substituted template alone. Exported for the golden test.

func CrossServicePrompt ¶

func CrossServicePrompt(services []string, findingsSummary, repoPath, depth string) string

CrossServicePrompt builds the exact prompt RunCrossServiceAnalyzer sends. Exported for the golden test.

func DastPrompt ¶

func DastPrompt(finding schemas.RawFinding, exploitPayload, repoPath, depth string) string

DastPrompt builds the exact prompt RunDastVerifier sends. Exported for the golden test.

func DepReachabilityPrompt ¶

func DepReachabilityPrompt(finding map[string]any, repoPath, depth string) string

DepReachabilityPrompt builds the exact prompt RunDepReachability sends. Exported for the golden test.

func ExploitPrompt ¶

func ExploitPrompt(
	finding schemas.RawFinding,
	dataFlowTrace schemas.DataFlowTrace,
	sanitization schemas.SanitizationResult,
	repoPath, depth string,
) string

ExploitPrompt builds the exact prompt RunExploitHypothesizer sends. Exported for the golden test.

func Fallback ¶

func Fallback(finding schemas.RawFinding, reason string, dropReason, originalVerdict *string) schemas.VerifiedFinding

Fallback ports verifier.py `fallback`:

def fallback(finding, reason, *, drop_reason=None, original_verdict=None) -> VerifiedFinding:
    rationale = f"Verification incomplete: {reason}"
    if original_verdict:
        rationale = f"{rationale} (original verdict: {original_verdict})"
    tags = ["low_confidence"] if drop_reason else []
    return VerifiedFinding(... verdict=INCONCLUSIVE, evidence_level=STATIC_MATCH ...)

This is the demotion path: a finding whose verification could not complete stays in the report as INCONCLUSIVE / STATIC_MATCH with a zero score, rather than being dropped silently. reasoners/phases.py and orchestrator.py both call it, and tests/test_prove_phase_demotion.py asserts its observable shape (verdict "inconclusive", drop_reason preserved, "low_confidence" in tags).

SIGNATURE PARITY. Python's two keyword-only arguments default to None, and both defaults are load-bearing:

• `drop_reason` drives BOTH the drop_reason field and the "low_confidence" tag — passing None yields an EMPTY tag list, not a tag list containing "".
• `original_verdict` is appended to the rationale only when TRUTHY, so an empty string behaves like None.

Go spells "or None" as a nil *string; StrPtr (shared.go) is the call-site helper. Passing (nil, nil) is Python's `fallback(finding, reason)`.

func PrioritySort ¶

func PrioritySort(findings []schemas.RawFinding) []schemas.RawFinding

PrioritySort ports `_priority_sort`:

return sorted(findings,
              key=lambda f: (_SEVERITY_RANK.get(f.estimated_severity, 0),
                             _CONFIDENCE_RANK.get(f.confidence, 0)),
              reverse=True)

Python parity: `sorted(..., reverse=True)` returns a NEW list and is STABLE — reversing does not reverse ties, so findings with the same (severity, confidence) keep their input order. sort.SliceStable over a copy with a strict greater-than comparator is exactly that; sort.Slice would be free to permute ties.

func RunChainBuilder ¶

func RunChainBuilder(
	ctx context.Context,
	app appx.Harnesser,
	repoPath string,
	potentialChains []schemas.PotentialChain,
	findings []schemas.VerifiedFinding,
	depth string,
) ([]schemas.VerifiedFinding, error)

RunChainBuilder ports chain_builder.py run_chain_builder.

if not potential_chains or not findings:
    return findings
prompt = _build_prompt(...)
try:
    result = await app.harness(prompt=prompt, cwd=repo_path)
    payload = _parse_payload(result)
except Exception:
    payload = None
if payload is None:
    return findings
findings_by_id = {f.id: f for f in findings}
for chain in payload["chains"]:
    _apply_validated_chain(findings_by_id, chain)
return list(findings_by_id.values())

Python parity notes:

• the harness call passes NO schema and cwd=repo_path (NOT a scratch temp dir and no project_dir) — the only prove agent that lets the coding agent work directly inside the repository.
• the try/except swallows EVERY exception from the harness call and from parsing, so a transport error, a provider failure and unparseable output are all "no chains" and the input findings come back untouched.
• `_apply_validated_chain` runs OUTSIDE that guard, so its AttributeError escapes — see ErrChainTagsNotASet. On that error the partially-mutated findings are returned alongside it so a caller that logs and continues sees exactly what Python's traceback would have left behind.

func RunCrossServiceAnalyzer ¶

func RunCrossServiceAnalyzer(
	ctx context.Context,
	app appx.Harnesser,
	repoPath string,
	services []string,
	findingsSummary string,
	depth string,
) (schemas.CrossServiceFinding, error)

RunCrossServiceAnalyzer ports cross_service.py run_cross_service_analyzer.

result = await app.harness(prompt=prompt, schema=CrossServiceFinding,
                           cwd=harness_cwd, project_dir=repo_path)
return extract_harness_result(result, CrossServiceFinding, "CrossServiceAnalyzer")

Same temp-dir contract as RunTracer. This agent is registered as the `run_cross_service_analyzer` reasoner but is NOT part of the audit DAG — the orchestrator never calls it (multi-repo scanning is driven externally).

func RunDastVerifier ¶

func RunDastVerifier(
	ctx context.Context,
	app appx.Harnesser,
	repoPath string,
	finding schemas.RawFinding,
	exploitPayload string,
	depth string,
) (schemas.DastVerificationResult, error)

RunDastVerifier ports dast_verifier.py run_dast_verifier.

result = await app.harness(prompt=prompt, schema=DastVerificationResult,
                           cwd=harness_cwd, project_dir=repo_path)
return extract_harness_result(result, DastVerificationResult, "DastVerifier")

Same temp-dir contract as RunTracer. The orchestrator reaches this through `_run_dast_verification`, gated on AuditInput.enable_dast.

func RunDepReachability ¶

func RunDepReachability(
	ctx context.Context,
	app appx.Harnesser,
	repoPath string,
	finding map[string]any,
	depth string,
) (schemas.ReachabilityProof, error)

RunDepReachability ports dep_reachability.py run_dep_reachability.

result = await app.harness(prompt=prompt, schema=ReachabilityProof,
                           cwd=harness_cwd, project_dir=repo_path)
return extract_harness_result(result, ReachabilityProof, "DependencyReachabilityAnalyzer")

Same temp-dir contract as RunTracer.

func RunExploitHypothesizer ¶

func RunExploitHypothesizer(
	ctx context.Context,
	app appx.Harnesser,
	repoPath string,
	finding schemas.RawFinding,
	dataFlowTrace schemas.DataFlowTrace,
	sanitization schemas.SanitizationResult,
	depth string,
) (schemas.ExploitHypothesis, error)

RunExploitHypothesizer ports exploit.py run_exploit_hypothesizer.

result = await app.harness(prompt=prompt, schema=ExploitHypothesis,
                           cwd=harness_cwd, project_dir=repo_path)
return extract_harness_result(result, ExploitHypothesis, "ExploitHypothesizer")

Same temp-dir contract as RunTracer.

func RunProve ¶

func RunProve(
	ctx context.Context,
	app HarnessAIer,
	repoPath string,
	huntResult schemas.HuntResult,
	depth string,
	maxConcurrentProvers int,
) ([]schemas.VerifiedFinding, error)

RunProve ports `run_prove`.

profile = _normalize_depth(depth)
prioritized = _priority_sort(hunt_result.findings)
verified = await _run_parallel_verification(app, repo_path, prioritized, profile.value, max_concurrent_provers)
verified = [_apply_metadata(f) for f in verified]
if hunt_result.chains:
    verified = await run_chain_builder(app=..., repo_path=..., potential_chains=hunt_result.chains,
                                       findings=verified, depth=profile.value)
    verified = [_apply_metadata(f) for f in verified]
verified.sort(key=lambda f: (f.exploitability_score, f.evidence_level), reverse=True)
return verified

Python parity notes:

• the depth string handed to the sub-agents is the NORMALIZED profile value, so an unrecognised depth reaches the prompts as "standard";
• `_apply_metadata` runs TWICE when chains are present — the second pass re-scores the findings the chain builder tagged with a chain_id, which is what turns a chained finding's exploitability score into the 2x chain-bonus form;
• run_chain_builder's AttributeError (see ErrChainTagsNotASet) propagates out of run_prove, exactly as it does in Python. The partially-mutated findings are returned alongside the error.

func RunProveStreaming ¶

func RunProveStreaming(
	ctx context.Context,
	app HarnessAIer,
	repoPath string,
	findingsQueue <-chan []schemas.RawFinding,
	depth string,
	maxConcurrentProvers int,
	proverCap int,
) []schemas.VerifiedFinding

RunProveStreaming ports `run_prove_streaming` — the incremental variant the orchestrator drives while HUNT is still producing findings.

profile = _normalize_depth(depth)
semaphore = asyncio.Semaphore(max(1, max_concurrent_provers))
while True:
    batch = await findings_queue.get()
    if batch is None: break
    for finding in batch:
        if proved_count >= prover_cap: break
        pending_tasks.append(asyncio.create_task(_verify_one(finding)))
        proved_count += 1
    if proved_count >= prover_cap:
        while True:                        # drain to the sentinel
            remaining = await findings_queue.get()
            if remaining is None: break
        break
if pending_tasks:
    results = await asyncio.gather(*pending_tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, BaseException): continue
        verified.append(_apply_metadata(result))
verified.sort(key=lambda f: (f.exploitability_score, f.evidence_level), reverse=True)

The Go channel stands in for the asyncio.Queue; a NIL batch is the `None` sentinel. An EMPTY-but-non-nil batch is a real (empty) batch and does NOT terminate the loop, matching Python where `[]` is not None.

Python parity notes:

• tasks START as soon as they are created (asyncio.create_task schedules immediately), bounded only by the semaphore — so verification of batch N overlaps arrival of batch N+1. The Go port launches a goroutine per finding at the same moment for the same reason.
• the semaphore here is `max(1, max_concurrent_provers)` — NOT clamped to the finding count as in _run_parallel_verification, because the total is not known up front.
• once the cap is hit the producer is drained to its sentinel rather than abandoned, which is what keeps the producing side from blocking forever on a full queue.
• `_verify_one` swallows every exception into a Fallback, so the `isinstance(result, BaseException): continue` filter never actually drops a finding. Ported as a nil-error check for the same reason as runParallelVerification.
• the results keep TASK CREATION order before the final sort, which is the queue arrival order.

Unlike RunProve there is no chain-builder pass here, so no error can escape; the signature returns only the findings.

func RunSanitizationAnalyzer ¶

func RunSanitizationAnalyzer(
	ctx context.Context,
	app appx.Harnesser,
	repoPath string,
	finding schemas.RawFinding,
	dataFlowTrace schemas.DataFlowTrace,
	depth string,
) (schemas.SanitizationResult, error)

RunSanitizationAnalyzer ports sanitization.py run_sanitization_analyzer.

result = await app.harness(prompt=prompt, schema=SanitizationResult,
                           cwd=harness_cwd, project_dir=repo_path)
return extract_harness_result(result, SanitizationResult, "SanitizationAnalyzer")

Same temp-dir contract as RunTracer.

func RunTracer ¶

func RunTracer(ctx context.Context, app appx.Harnesser, repoPath string, finding schemas.RawFinding, depth string) (schemas.DataFlowTrace, error)

RunTracer ports tracer.py run_tracer.

async def run_tracer(app, repo_path, finding, depth) -> DataFlowTrace:
    prompt = _build_prompt(...) + "\n\nCONTEXT:\n..."
    harness_cwd = tempfile.mkdtemp(prefix="secaf-prove-tracer-")
    try:
        result = await app.harness(prompt=prompt, schema=DataFlowTrace,
                                   cwd=harness_cwd, project_dir=repo_path)
        return extract_harness_result(result, DataFlowTrace, "DataFlowTracer")
    finally:
        shutil.rmtree(harness_cwd, ignore_errors=True)

Python parity: the harness runs with Cwd set to a PRIVATE scratch directory and ProjectDir set to the repository, so the coding agent explores the repo but writes its JSON output outside it. `shutil.rmtree(..., ignore_errors=True)` maps to a deferred os.RemoveAll whose error is deliberately dropped.

func RunVerdictAgent ¶

func RunVerdictAgent(
	ctx context.Context,
	app appx.AIer,
	repoPath string,
	finding schemas.RawFinding,
	dataFlow schemas.DataFlowTrace,
	sanitization schemas.SanitizationResult,
	exploit schemas.ExploitHypothesis,
) (schemas.VerdictDecision, error)

RunVerdictAgent ports verdict.py run_verdict_agent.

result = await app.ai(user=prompt, schema=VerdictDecision)
return _extract_ai_result(result, "VerdictAgent")

Python parity notes:

• `repo_path` is accepted and IGNORED (the comment in verdict.py explains why: a pure judgment task needs no file access). The parameter is kept so the Go call sites read like the Python ones and so reasoners/phases.py's `repo_path="."` has somewhere to go.
• `system=` is not passed, so aix.Structured gets the empty system prompt and adds no system message — Python's `system=None`.
• `_extract_ai_result`'s three duck-typed branches (already a VerdictDecision / a dict / `.parsed`) all collapse into aix.Structured's single "unmarshal the response text into T", because the Go SDK's ai.Response only ever carries text. A response that will not parse yields an error whose message names VerdictDecision, which is what the Python TypeError does.

func RunVerifier ¶

func RunVerifier(ctx context.Context, app HarnessAIer, repoPath string, finding schemas.RawFinding, depth string) (schemas.VerifiedFinding, error)

RunVerifier ports verifier.py run_verifier — the four-stage in-process chain.

seed_trace = DataFlowTrace(source=f"{file}:{line}", sink=function or file,
                           steps=[...] if finding.data_flow else [], sink_reached=False)
results = await asyncio.gather(run_tracer(...), run_sanitization_analyzer(...),
                               return_exceptions=True)
data_flow_trace = seed_trace if tracer failed else tracer_result
sanitization = SanitizationResult(found=False, sufficient=None, bypass_method=None)
               if sanitization failed else sanitization_result
exploit = await run_exploit_hypothesizer(...)     # errors PROPAGATE
verdict = await run_verdict_agent(...)            # errors PROPAGATE
verified = assemble_verified_finding(...)
# backfill sarif_rule_id and reproduction_steps

Concurrency parity: stages 1 and 2 run CONCURRENTLY with `return_exceptions=True`, so a failure in one does not cancel or fail the other — each has its own documented fallback value. Go uses a WaitGroup with two indexed result slots rather than errgroup precisely because no error must escape. Stages 3 and 4 are sequential awaits whose errors DO propagate; the caller (_run_parallel_verification) turns them into a demoted Fallback.

Python parity on the sanitization seed: the note that the seed trace's `sink_reached` is False even when the hunter reported a flow is deliberate — only the tracer may claim the sink is reached.

func SanitizationPrompt ¶

func SanitizationPrompt(finding schemas.RawFinding, trace schemas.DataFlowTrace, repoPath, depth string) string

SanitizationPrompt builds the exact prompt RunSanitizationAnalyzer sends. Exported for the golden test.

func StrPtr ¶

func StrPtr(s string) *string

StrPtr returns a pointer to s.

Several ported functions take `*string` parameters because their Python counterparts declare keyword-only arguments defaulting to None (Fallback's dropReason / originalVerdict). Go has no literal address-of for a constant, so callers need this one-liner; it lives here rather than in each caller.

func TracerPrompt ¶

func TracerPrompt(finding schemas.RawFinding, repoPath, depth string) string

TracerPrompt builds the exact prompt RunTracer sends, including the CONTEXT block Python appends after substitution. Exported for the golden test.

func VerdictPrompt ¶

func VerdictPrompt(
	finding schemas.RawFinding,
	dataFlow schemas.DataFlowTrace,
	sanitization schemas.SanitizationResult,
	exploit schemas.ExploitHypothesis,
) string

VerdictPrompt builds the exact prompt RunVerdictAgent sends. Unlike every other prove agent there is NO trailing CONTEXT block — verdict.py sends the substituted template alone. Exported for the golden test.