hunt

func AllStrategies ¶

func AllStrategies() []schemas.HuntStrategy

AllStrategies returns the table's keys in order — `list(_STRATEGY_RUNNERS)`.

func AssembleFinding ¶

func AssembleFinding(
	location schemas.VulnLocation,
	enriched schemas.EnrichedFinding,
	findingType, strategy string,
) schemas.RawFinding

AssembleFinding ports _scan_enrich.py assemble_finding:

snippet_line_count = max(1, location.code_snippet.count("\n") + 1)
data_flow = None
summary = enriched.data_flow_summary.strip()
if summary:
    data_flow = [DataFlowStep(file_path=location.file_path, line=location.start_line,
                              component=strategy, operation=summary)]
return RawFinding(..., end_line=location.start_line + snippet_line_count - 1, ...)

Python parity notes:

• END LINE. `count("\n") + 1` counts LINES, so a snippet with no newline is one line and end_line == start_line; a snippet with a TRAILING newline ("one\ntwo\n") counts as three, so end_line overshoots by one. That is reproduced, not corrected. The `max(1, ...)` can never bite (count is never negative) and is kept only to mirror the source.
• DATA FLOW. An empty-after-strip summary leaves data_flow at None, which model_dump()s as `null` — hence a nil slice here, not an empty one. The step's `component` is the STRATEGY, not a code component, and its `line` is the location's start_line.
• `strip()` vs strings.TrimSpace: Python strips every character whose str.isspace() is true, which includes U+001C..U+001F (the file/group/ record/unit separators); Go's unicode.IsSpace does not. Both strip the ASCII whitespace, NEL and NBSP that real LLM output contains.
• CWE NAME. Python passes `cwe_name=enriched.cwe_id` — the ID, not a name. Deliberate duplication, reproduced.
• The finding's `id` and `fingerprint` are pydantic `default_factory=lambda: str(uuid4())`, so both are FRESH RANDOM UUIDs here; the fingerprint is overwritten with a real content fingerprint later, by dedup. schemas.NewRawFinding mints both the same way.

func EnrichLocation ¶

func EnrichLocation(
	ctx context.Context,
	app appx.Harnesser,
	location schemas.VulnLocation,
	findingType, strategy, reconContext, repoPath string,
) (schemas.EnrichedFinding, error)

EnrichLocation ports _scan_enrich.py enrich_location.

Python parity: the temp-dir prefix embeds the strategy — `tempfile.mkdtemp(prefix=f"secaf-hunt-enrich-{strategy}-")` — so a strategy containing a path separator would change where the scratch dir lands. Go's os.MkdirTemp REJECTS a pattern containing a separator (it returns ErrPatternHasSeparator) where Python's mkdtemp would happily build a nested path; every strategy this port passes is a bare HuntStrategy value, so the two agree in practice. The error is returned rather than swallowed.

func EnrichLocationsParallel ¶

func EnrichLocationsParallel(
	ctx context.Context,
	app appx.Harnesser,
	locations []schemas.VulnLocation,
	findingType, strategy, reconContext, repoPath string,
	maxConcurrent int,
) ([]schemas.EnrichedFinding, error)

EnrichLocationsParallel ports _scan_enrich.py enrich_locations_parallel:

if not locations: return []
semaphore = asyncio.Semaphore(max(1, max_concurrent))
async def _run(location): async with semaphore: return await enrich_location(...)
return await asyncio.gather(*[_run(location) for location in locations])

Concurrency parity:

• the bound is `max(1, max_concurrent)`, so a zero or negative max_concurrent still admits one enrichment at a time rather than deadlocking;
• results are index-aligned with locations. asyncio.gather preserves the input order regardless of completion order, so the port writes into a pre-sized slice rather than appending;
• errgroup.Group is used WITHOUT WithContext so a failing enrichment does not cancel its siblings — matching gather(return_exceptions=False), which never cancels the other awaitables either. Wait() returns the first error by completion time, the same one gather surfaces.
• DIFFERENCE: Wait() blocks until every goroutine has finished, while `await gather(...)` resumes the caller as soon as the first exception fires and leaves the rest running detached. The returned value is identical; only the moment of return differs, and every caller's next act on error is to propagate it.

On error the returned slice is nil: Python's caller never reads the partial results either, because the exception propagates out of the `await`.

A cancelled ctx surfaces through semaphore.Acquire as ctx.Err(), which is the closest Go analogue of the CancelledError asyncio would raise inside the gathered coroutines.

func EnrichPrompt ¶

func EnrichPrompt(location schemas.VulnLocation, findingType, strategy, reconContext string) string

EnrichPrompt builds the Step 2 prompt for one location.

Ports the seven chained `.replace(...)` calls in enrich_location. Order is irrelevant to the output here (no substituted value contains another marker in practice) but is kept identical to Python's anyway.

func IsBusinessLogicHunterEnabled ¶

func IsBusinessLogicHunterEnabled(depth string) bool

IsBusinessLogicHunterEnabled ports business_logic.py is_business_logic_hunter_enabled:

profile = _normalize_depth(depth)
return profile in {DepthProfile.STANDARD, DepthProfile.THOROUGH}

Python parity: the normalization is the LENIENT one, so an unrecognised depth becomes STANDARD and the hunter runs. Only "quick" (in any case) disables it. business_logic.py declares its own `_normalize_depth` accepting `str | DepthProfile`; config.NormalizeDepth covers both arms because Go's DepthProfile IS a string type.

func IsLogicHunterEnabled ¶

func IsLogicHunterEnabled(depth string) bool

IsLogicHunterEnabled ports logic.py is_logic_hunter_enabled — a straight delegation to is_business_logic_hunter_enabled.

func QuickStrategies ¶

func QuickStrategies() []schemas.HuntStrategy

QuickStrategies returns a copy of _QUICK_STRATEGIES.

func RunAPISecurityHunter ¶

func RunAPISecurityHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunAPISecurityHunter ports api_security.py run_api_security_hunter:

async def run_api_security_hunter(app, repo_path, recon,
                                  max_files_without_signal: int = 30) -> HuntResult

func RunAuthHunter ¶

func RunAuthHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, depth string, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunAuthHunter ports auth.py run_auth_hunter.

func RunBusinessLogicHunter ¶

func RunBusinessLogicHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, depth string, maxFilesWithoutSignal int, depthPrompt string,
) (schemas.HuntResult, error)

RunBusinessLogicHunter ports business_logic.py run_business_logic_hunter:

async def run_business_logic_hunter(app, repo_path, recon_result, depth,
                                    max_files_without_signal: int = 30,
                                    depth_prompt: str = "") -> HuntResult

depthPrompt is the Go spelling of the `depth_prompt: str = ""` default; pass "" for it. __init__.py computes a depth_prompt but never manages to deliver it (package doc), so the live pipeline always passes "".

func RunConfigSecretsHunter ¶

func RunConfigSecretsHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunConfigSecretsHunter ports config_secrets.py run_config_secrets_hunter:

async def run_config_secrets_hunter(app, repo_path, recon,
                                    max_files_without_signal: int = 30) -> HuntResult

func RunCryptoHunter ¶

func RunCryptoHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunCryptoHunter ports crypto.py run_crypto_hunter:

async def run_crypto_hunter(app, repo_path, recon,
                            max_files_without_signal: int = 30) -> HuntResult

Note the third parameter is named `recon`, not `recon_result`, and there is no `depth` at all — which is exactly why __init__.py's argument cascade falls through to its POSITIONAL shape for this hunter and lands the depth string in max_files_without_signal. This exported entry point is the honest one, used by src/sec_af/reasoners/hunt.py.

func RunDataExposureHunter ¶

func RunDataExposureHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunDataExposureHunter ports data_exposure.py run_data_exposure_hunter:

async def run_data_exposure_hunter(app, repo_path, recon,
                                   max_files_without_signal: int = 30) -> HuntResult

func RunDosHunter ¶

func RunDosHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, depth string, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunDosHunter ports dos.py run_dos_hunter.

func RunHunt ¶

func RunHunt(
	ctx context.Context,
	app appx.Harnesser,
	repoPath string,
	recon schemas.ReconResult,
	depth string,
	maxConcurrentHunters int,
	earlyStopFileThreshold int,
	includePaths []string,
) (schemas.HuntResult, error)

RunHunt ports src/sec_af/agents/hunt/__init__.py run_hunt:

async def run_hunt(app, repo_path, recon_result, depth,
                   max_concurrent_hunters=4, early_stop_file_threshold=30,
                   include_paths=None) -> HuntResult

Sequence: normalize the depth leniently, select the strategies, run them all under a semaphore, flatten the per-hunter findings in STRATEGY order, hand everything to dedup, apply the include-paths filter, then overwrite the four counters and the duration.

Concurrency parity:

• A failed hunter contributes no findings, but codeaf records the failure separately so a resumed audit cannot mistake it for a completed search.
• Results are flattened in strategy order regardless of completion order, so the port writes into a pre-sized slice rather than appending from the goroutines.
• The semaphore is held for the WHOLE hunter, which includes that hunter's own five-way enrichment fan-out — the two limits multiply.

Python parity: earlyStopFileThreshold is accepted and unused. See the package doc; DefaultEarlyStopFileThreshold is the value every caller passes.

func RunHuntStreaming ¶

func RunHuntStreaming(
	ctx context.Context,
	app appx.Harnesser,
	repoPath string,
	recon schemas.ReconResult,
	findings chan<- []schemas.RawFinding,
	depth string,
	maxConcurrentHunters int,
	earlyStopFileThreshold int,
	includePaths []string,
) (schemas.HuntResult, error)

RunHuntStreaming ports run_hunt_streaming — run_hunt plus an incremental fingerprint dedup that publishes each hunter's NEW findings as soon as it finishes, so the PROVE phase can start before HUNT is done.

async def run_hunt_streaming(app, repo_path, recon_result, findings_queue, depth,
                             max_concurrent_hunters=4, early_stop_file_threshold=30,
                             include_paths=None) -> HuntResult

Channel semantics: Python's `asyncio.Queue[list[RawFinding] | None]` carries batches and is terminated by a None sentinel. The Go port sends the same batches on findings and CLOSES it in place of the sentinel — a closed channel is Go's end-of-stream, and it cannot be mistaken for a real (empty) batch the way a nil slice could. The channel is always closed, including when a hunter fails.

Two deliberate differences from Python, both forced by channels:

• Python's default Queue is UNBOUNDED, so `put` never blocks. A Go channel is not: pass a buffer of at least len(SelectStrategies(depth)) if the consumer is not draining concurrently, or the producers will stall.
• The send is guarded by ctx.Done() so a consumer that goes away cannot wedge the hunt. Python has no equivalent because it cannot block here.

The fingerprint pass differs from run_hunt's in a way that matters: it seeds `finding.fingerprint` with `f"{file_path}:{start_line}:{cwe_id}"` when the hunter left it empty, which is NOT the sha256 form dedup.ComputeFingerprint produces. Findings therefore reach DeduplicateAndCorrelate already fingerprinted, and that function's own seeding never fires for them.

func RunInjectionHunter ¶

func RunInjectionHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, depth string, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunInjectionHunter ports injection.py run_injection_hunter:

async def run_injection_hunter(app, repo_path, recon_result, depth,
                               max_files_without_signal: int = 30) -> HuntResult

func RunLogicHunter ¶

func RunLogicHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, depth string, maxFilesWithoutSignal int, depthPrompt string,
) (schemas.HuntResult, error)

RunLogicHunter ports logic.py run_logic_hunter:

async def run_logic_hunter(app, repo_path, recon, depth,
                           max_files_without_signal=30, depth_prompt="") -> HuntResult:
    return await run_business_logic_hunter(
        app=app, repo_path=repo_path, recon_result=recon,
        depth=depth, max_files_without_signal=max_files_without_signal,
        depth_prompt=depth_prompt)

Every argument is forwarded unchanged, so the prompt, the findings and the strategies_run value ("business_logic", never "logic") are byte-identical to RunBusinessLogicHunter's for the same inputs.

func RunSSRFHunter ¶

func RunSSRFHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, depth string, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunSSRFHunter ports ssrf.py run_ssrf_hunter.

func RunSupplyChainHunter ¶

func RunSupplyChainHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunSupplyChainHunter ports supply_chain.py run_supply_chain_hunter:

async def run_supply_chain_hunter(app, repo_path, recon,
                                  max_files_without_signal: int = 30) -> HuntResult

func RunXSSHunter ¶

func RunXSSHunter(
	ctx context.Context, app appx.Harnesser, repoPath string,
	recon schemas.ReconResult, depth string, maxFilesWithoutSignal int,
) (schemas.HuntResult, error)

RunXSSHunter ports xss.py run_xss_hunter.

func ScanLocations ¶

func ScanLocations(ctx context.Context, app appx.Harnesser, prompt, repoPath string) ([]schemas.VulnLocation, error)

ScanLocations ports _scan_enrich.py scan_locations:

async def scan_locations(app, prompt, repo_path) -> list[VulnLocation]:
    scan_template = SCAN_PROMPT_PATH.read_text(encoding="utf-8")
    scan_prompt = scan_template.replace("{{HUNTER_PROMPT}}", prompt)
    harness_cwd = tempfile.mkdtemp(prefix="secaf-hunt-scan-")
    try:
        result = await app.harness(prompt=scan_prompt, schema=ScanLocationsResult,
                                   cwd=harness_cwd, project_dir=repo_path)
        parsed = extract_harness_result(result, ScanLocationsResult, "Hunt location scanner")
        return parsed.locations
    finally:
        shutil.rmtree(harness_cwd, ignore_errors=True)

Python parity: `str.replace` replaces EVERY occurrence, so a template with two {{HUNTER_PROMPT}} markers would get both filled — strings.ReplaceAll, not strings.Replace(…, 1). The harness runs with Cwd on a private scratch dir and ProjectDir on 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 ScanPrompt ¶

func ScanPrompt(hunterPrompt string) string

ScanPrompt builds the Step 1 prompt: the shared scan_locations template with the hunter's own prompt substituted for {{HUNTER_PROMPT}}.

Extracted as a pure function so a golden test can compare it byte-for-byte against the Python builder (testdata/golden/scan_locations_prompt.txt).

func SelectStrategies ¶

func SelectStrategies(depth config.DepthProfile) []schemas.HuntStrategy

SelectStrategies ports _select_strategies:

if depth == DepthProfile.QUICK: return list(_QUICK_STRATEGIES)
return list(_STRATEGY_RUNNERS)

Anything that is not QUICK — including a depth the lenient normalizer already turned into STANDARD — runs the full eleven.

func ShouldRunCryptoHunter ¶

func ShouldRunCryptoHunter(recon schemas.ReconResult) bool

ShouldRunCryptoHunter ports crypto.py should_run_crypto_hunter:

return bool(recon.security_context.crypto_usage)

Note this checks the LIST, not the usage contexts: a CryptoUsage with a nil usage_context still opens the gate (and then contributes to neither candidate list).

func ShouldRunSupplyChainHunter ¶

func ShouldRunSupplyChainHunter(recon schemas.ReconResult) bool

ShouldRunSupplyChainHunter ports supply_chain.py should_run_supply_chain_hunter:

return recon.dependencies.direct_count > 0

Python parity: the gate reads direct_count only. A repository whose SBOM is populated but whose direct_count was never set still skips the hunter.