reasoners

package
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 14, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Overview

Package reasoners ports src/sec_af/reasoners/{__init__,recon,hunt,prove}.py — the thin adapter layer between the reasoner surface the pipeline `.call`s and the agent functions under internal/agents.

Python builds that surface with a module-level router: 

router = AgentRouter(tags=["security", "audit", "red-team"])   # __init__.py

@router.reasoner()
async def run_architecture_mapper(repo_path: str) -> dict[str, Any]:
    router.note("Architecture mapper starting", tags=["recon", "architecture"])
    result = await _run_architecture_mapper(router, repo_path)
    return result.model_dump()

Every adapter does the same four things, and this package reproduces them one for one:

  1. emit the reasoner's "X starting" note with its exact tags (a few carry none — see NameRunLogicBugsHunter and NameRunCWEExpansion);
  2. materialize the untyped request body into the pydantic models the agent function expects (afx.Bind over a typed input struct whose json tags and defaults are the Python signature, plus the required-field checks pydantic performs — validate.go);
  3. call the agent function;
  4. return `result.model_dump()` as a map (afx.ToMap).

Input validation

Step 2 is not the first thing a request meets. The Python SDK runs `Agent._validate_handler_input` over the body BEFORE the decorated function is entered, answering 422 for a null on a required parameter and COERCING scalars (`"50"` -> 50, `5` -> "5", `"yes"` -> true). handler_input.go ports that layer and RegisterAll wraps every handler in it; internal/secaf/audit runs the same layer over the audit entry's own request. A refusal is an *InputError carrying 422.

Registration

RegisterAll puts the 33 router reasoners into a Registry in the canonical DESIGN.md §3 order and returns its own ordered bookkeeping, which the parity test compares with the registry's. The sec-af node mounted the same handlers on an SDK router served to the AgentField control plane; inside codeaf the registry is read by the in-process Caller (internal/secaf/audit), and the router's tags, which only labelled the node for the control plane, are gone.

Input schemas

Python publishes a JSON Schema per reasoner, derived from the decorated function's signature. The exact schemas the Python node registered are committed as testdata/python_input_schemas.json and replayed verbatim through InputSchema — including Python's own derivation quirks — and the registry holds each reasoner's schema beside its handler. input_schemas.go documents them and how the capture was made.

A SECOND capture, testdata/python_input_types.json, holds the raw `(annotation, default)` pairs the same signatures produce — the input `_validate_handler_input` runs on. The two are derived by different SDK code paths from the same source, and handler_input_test.go asserts they agree.

What is NOT here

The `*_phase` bodies (the four reasoners that were the control-plane DAG) live in internal/phases; this package only binds their inputs and threads the node id. `run_cwe_expansion` likewise delegates to phases.RunCWEExpansion. Keeping the split means a phase can be tested without the registration layer and vice versa.

Index

Constants

View Source 
const (
	NameRunArchitectureMapper      = "run_architecture_mapper"
	NameRunDependencyAuditor       = "run_dependency_auditor"
	NameRunConfigScanner           = "run_config_scanner"
	NameRunDataFlowMapper          = "run_data_flow_mapper"
	NameRunSecurityContextProfiler = "run_security_context_profiler"

	NameRunInjectionHunter     = "run_injection_hunter"
	NameRunDosHunter           = "run_dos_hunter"
	NameRunSSRFHunter          = "run_ssrf_hunter"
	NameRunAuthHunter          = "run_auth_hunter"
	NameRunXSSHunter           = "run_xss_hunter"
	NameRunCryptoHunter        = "run_crypto_hunter"
	NameRunBusinessLogicHunter = "run_business_logic_hunter"
	// NameRunLogicBugsHunter is an ALIAS reasoner: its Python body forwards to
	// run_business_logic_hunter (the reasoner, not the agent function), so it
	// emits no note of its own and its DAG child is indistinguishable from a
	// direct run_business_logic_hunter call.
	NameRunLogicBugsHunter     = "run_logic_bugs_hunter"
	NameRunDataExposureHunter  = "run_data_exposure_hunter"
	NameRunSupplyChainHunter   = "run_supply_chain_hunter"
	NameRunConfigSecretsHunter = "run_config_secrets_hunter"
	NameRunAPISecurityHunter   = "run_api_security_hunter"
	NameRunDeduplicator        = "run_deduplicator"

	NameRunDepReachability      = "run_dep_reachability"
	NameRunVerifier             = "run_verifier"
	NameRunTracer               = "run_tracer"
	NameRunSanitizationAnalyzer = "run_sanitization_analyzer"
	NameRunExploitHypothesizer  = "run_exploit_hypothesizer"
	NameRunVerdictAgent         = "run_verdict_agent"
	NameRunRemediation          = "run_remediation"
	NameRunRemediationAgent     = "run_remediation_agent"
	NameRunDastVerifier         = "run_dast_verifier"
	NameRunCrossServiceAnalyzer = "run_cross_service_analyzer"

	// NameRunCWEExpansion is REGISTERED but never `.call`ed: hunt_phase invokes
	// expand_cwes_for_hunt in process (an `.ai()` call), so this reasoner
	// contributes no DAG node. See DESIGN.md §3.
	NameRunCWEExpansion = "run_cwe_expansion"

	NameReconPhase       = "recon_phase"
	NameHuntPhase        = "hunt_phase"
	NameProvePhase       = "prove_phase"
	NameRemediationPhase = "remediation_phase"
)

The 33 router reasoner names, in DESIGN.md §3 order. The order is the Python import order of reasoners/__init__.py (recon -> hunt -> prove -> phases) and, within each module, the order the `@router.reasoner()` decorators appear.

View Source 
const DefaultMaxFilesWithoutSignal = 30

DefaultMaxFilesWithoutSignal ports the `max_files_without_signal: int = 30` default every hunter reasoner declares.

View Source 
const NameAudit = "audit"

NameAudit is the ONE externally driven reasoner. Python registered it with `@app.reasoner()` on the Agent, not on the AgentRouter, so it is not part of the registry; internal/secaf/audit owns the entry, and the name is still how its input specification is looked up (ValidateHandlerInput).

Variables

View Source 
var Names = []string{
	NameRunArchitectureMapper,
	NameRunDependencyAuditor,
	NameRunConfigScanner,
	NameRunDataFlowMapper,
	NameRunSecurityContextProfiler,
	NameRunInjectionHunter,
	NameRunDosHunter,
	NameRunSSRFHunter,
	NameRunAuthHunter,
	NameRunXSSHunter,
	NameRunCryptoHunter,
	NameRunBusinessLogicHunter,
	NameRunLogicBugsHunter,
	NameRunDataExposureHunter,
	NameRunSupplyChainHunter,
	NameRunConfigSecretsHunter,
	NameRunAPISecurityHunter,
	NameRunDeduplicator,
	NameRunDepReachability,
	NameRunVerifier,
	NameRunTracer,
	NameRunSanitizationAnalyzer,
	NameRunExploitHypothesizer,
	NameRunVerdictAgent,
	NameRunRemediation,
	NameRunRemediationAgent,
	NameRunDastVerifier,
	NameRunCrossServiceAnalyzer,
	NameRunCWEExpansion,
	NameReconPhase,
	NameHuntPhase,
	NameProvePhase,
	NameRemediationPhase,
}

Names is the canonical ordered list of the 33 router reasoner names — DESIGN.md §3's list verbatim. RegisterAll registers exactly these, in exactly this order, and returns the same slice content as its bookkeeping.

Functions

func CoerceVerifierFinding 

func CoerceVerifierFinding(finding map[string]any) (schemas.RawFinding, error)

CoerceVerifierFinding ports `_coerce_verifier_finding(finding)` (reasoners/prove.py:28): 

try:
    return RawFinding.model_validate(finding)
except Exception:
    view = FindingForVerifier.model_validate(finding)
    return RawFinding(
        id=view.id,
        hunter_strategy="phase_boundary_projection",
        title=view.title,
        description=view.data_flow_summary or view.title,
        finding_type=FindingType.SAST,
        cwe_id=view.cwe_id,
        cwe_name=view.cwe_id,
        file_path=view.file_path,
        start_line=view.start_line,
        end_line=view.end_line,
        function_name=view.function_name,
        code_snippet=view.code_snippet,
        estimated_severity=apply_cwe_severity_floor(view.cwe_id, Severity.MEDIUM),
        confidence=Confidence.MEDIUM,
        fingerprint=view.id,
    )

This is the reasoner boundary that makes prove_phase work: prove_phase sends `finding.for_verifier().model_dump()`, a ten-field FindingForVerifier projection, which cannot satisfy RawFinding's twelve required fields — so the fallback branch is the LIVE path, not an edge case. See validate.go for why the required-field check has to be real.

Python parity in the fallback:

  • `cwe_name` is set to the CWE **id**, not a name. Reproduced.
  • `description` uses `view.data_flow_summary or view.title` — Python truthiness, so an empty summary falls back to the title.
  • the reconstructed RawFinding's `fingerprint` is the view's id, and its `id` is the view's id too, so the two match (a fresh RawFinding would have minted two different uuid4s).
  • `related_files` is not passed and keeps `default_factory=list` -> []; `data_flow` keeps None. NewRawFinding() seeds exactly that (and then the explicit ID/Fingerprint assignments overwrite its minted uuid4s).
  • the `except Exception` is BROAD: any failure to build a RawFinding — not just a missing field — takes the fallback. A payload that is neither model still fails, on the FindingForVerifier validation.

func HandlerInputStatusError 

func HandlerInputStatusError(err error) error

HandlerInputStatusError maps a *HandlerInputError onto the *InputError that carries 422, the status Python's endpoint returned.

func HuntPhase 

func HuntPhase(ctx context.Context, app appx.App, nodeID string, in HuntPhaseInput) (map[string]any, error)

HuntPhase ports the `hunt_phase` reasoner (reasoners/phases.py:239).

Python parity: `ai_gate` is always None on the live path (app.py's `.call` omits it), so phases.NewJSONAIGate yields nil there. A direct caller that supplies one takes the `ai_gate is not None` branch and gets the "AI gate failed: ..." note — see HuntPhaseInput.

func InputSchema 

func InputSchema(name string) json.RawMessage

InputSchema returns the input schema the Python node publishes for the reasoner called name, as the registry stores it beside the handler.

It PANICS when the fixture has no entry for name. That is the point: every registration goes through here, so adding a reasoner to the Go port without regenerating the capture (or renaming one out of sync with Python) crashes at registration — loudly, at boot and in every test — instead of silently publishing the SDK's `{"type":"object","additionalProperties":true}` placeholder and letting the two nodes drift apart in discovery.

The returned slice is a copy, so a caller cannot mutate the shared fixture.

func InputSchemaNames 

func InputSchemaNames() []string

InputSchemaNames returns every reasoner id the fixture carries, sorted. It is the "what Python publishes" side of the registration parity assertions.

func ProvePhase 

func ProvePhase(ctx context.Context, app appx.App, nodeID string, in ProvePhaseInput) (map[string]any, error)

ProvePhase ports the `prove_phase` reasoner (reasoners/phases.py:381).

func ReconPhase 

func ReconPhase(ctx context.Context, app appx.App, nodeID string, in ReconPhaseInput) (map[string]any, error)

ReconPhase ports the `recon_phase` reasoner (reasoners/phases.py:152).

func RegisterAll 

func RegisterAll(reg *Registry, app appx.App) []string

RegisterAll registers every router reasoner on reg, in DESIGN.md §3 order, and returns the ordered names it registered.

app is the App every handler closes over — and therefore the App whose Call the four `*_phase` handlers make their nested `.call`s through. To keep those nested calls in process too, pass the same local-calling App the registry is answering for (audit.WithLocalCalls does exactly that).

nodeID is read ONCE, here, from phases.NodeID(): Python's module-level `NODE_ID = os.getenv("NODE_ID", "sec-af")`, now a constant. Every `*_phase` handler closes over that value, so the `.call` targets are fixed.

func RemediationPhase 

func RemediationPhase(ctx context.Context, app appx.App, nodeID string, in RemediationPhaseInput) (map[string]any, error)

RemediationPhase ports the `remediation_phase` reasoner (reasoners/phases.py:487).

func RunAPISecurityHunter 

func RunAPISecurityHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunAPISecurityHunter ports `run_api_security_hunter` (reasoners/hunt.py:256). Depth-less hunter — see runHunter.

func RunArchitectureMapper 

func RunArchitectureMapper(ctx context.Context, app appx.App, in RepoPathInput) (map[string]any, error)

RunArchitectureMapper ports `run_architecture_mapper(repo_path)` (reasoners/recon.py:16).

func RunAuthHunter 

func RunAuthHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunAuthHunter ports `run_auth_hunter` (reasoners/hunt.py:123).

func RunBusinessLogicHunter 

func RunBusinessLogicHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunBusinessLogicHunter ports `run_business_logic_hunter` (reasoners/hunt.py:174).

Python parity: the agent function's sixth parameter `depth_prompt: str = ""` is never supplied by _run_hunter, so it keeps its default.

func RunCWEExpansion 

func RunCWEExpansion(ctx context.Context, app appx.App, in CWEExpansionInput) (map[string]any, error)

RunCWEExpansion ports `run_cwe_expansion(recon_summary, strategies)` (reasoners/phases.py:139): 

additional = await expand_cwes_for_hunt(recon_summary, strategies)
return {"additional_cwes": additional}

Python parity: no note, and no error path — expand_cwes_for_hunt swallows every failure and returns []. It is registered but never `.call`ed (hunt_phase runs expand_cwes_for_hunt in process), so it draws no DAG node.

func RunConfigScanner 

func RunConfigScanner(ctx context.Context, app appx.App, in RepoPathInput) (map[string]any, error)

RunConfigScanner ports `run_config_scanner(repo_path)` (reasoners/recon.py:32).

func RunConfigSecretsHunter 

func RunConfigSecretsHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunConfigSecretsHunter ports `run_config_secrets_hunter` (reasoners/hunt.py:239). Depth-less hunter — see runHunter.

func RunCrossServiceAnalyzer 

func RunCrossServiceAnalyzer(ctx context.Context, app appx.App, in CrossServiceInput) (map[string]any, error)

RunCrossServiceAnalyzer ports `run_cross_service_analyzer(repo_path, services, findings_summary, depth)` (reasoners/prove.py:169).

func RunCryptoHunter 

func RunCryptoHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunCryptoHunter ports `run_crypto_hunter` (reasoners/hunt.py:157).

Python parity: run_crypto_hunter's signature has no `depth`, so _run_hunter reaches it through the third cascade shape — depth is DROPPED and max_files_without_signal falls back to 30. See runHunter.

func RunDastVerifier 

func RunDastVerifier(ctx context.Context, app appx.App, in DastVerifierInput) (map[string]any, error)

RunDastVerifier ports `run_dast_verifier(repo_path, finding, exploit_payload, depth)` (reasoners/prove.py:158).

func RunDataExposureHunter 

func RunDataExposureHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunDataExposureHunter ports `run_data_exposure_hunter` (reasoners/hunt.py:205). Depth-less hunter — see runHunter.

func RunDataFlowMapper 

func RunDataFlowMapper(ctx context.Context, app appx.App, in ArchitectureInput) (map[string]any, error)

RunDataFlowMapper ports `run_data_flow_mapper(repo_path, architecture)` (reasoners/recon.py:40): 

architecture_model = ArchitectureMap(**architecture)

ArchitectureMap has no required field of its OWN, so an empty (or absent) `architecture` binds to the pydantic defaults rather than raising — but each of its five lists holds a model that DOES have required fields (Module name/path/language, EntryPoint kind/identifier/file_path/line, APIEndpoint method/path/handler/file_path/line, ...), and pydantic validates those. `ArchitectureMap(**{"modules": [{"name": "x"}]})` raises 2 errors on the pinned interpreter, so this is bindArchitectureMap, not a bare afx.Bind: the reasoner is registered on the router and a control-plane caller can send that payload directly.

func RunDeduplicator 

func RunDeduplicator(ctx context.Context, app appx.App, in DeduplicatorInput) (map[string]any, error)

RunDeduplicator ports `run_deduplicator(findings, recon_context, repo_path)` (reasoners/hunt.py:273): 

raw_findings = [RawFinding(**f) for f in findings]
recon = ReconResult(**recon_context)
result = await _deduplicate_and_correlate(raw_findings, recon, router, repo_path)
return result.model_dump()

Python parity: `ReconResult(**recon_context)` here is the RAW constructor — it does NOT go through _recon_model, so an incomplete recon_context fails validation instead of being seeded. hunt_phase always forwards the caller's full recon dump to this reasoner (only the HUNTERS get the pruned projection), so the strict bind is the correct one.

func RunDepReachability 

func RunDepReachability(ctx context.Context, app appx.App, in FindingDepthInput) (map[string]any, error)

RunDepReachability ports `run_dep_reachability(repo_path, finding, depth)` (reasoners/prove.py:52).

Python parity: `finding` is forwarded to the agent as the RAW DICT — this is the one prove reasoner that does not build a model first, because agents/prove/dep_reachability.py renders the dict into its prompt directly.

func RunDependencyAuditor 

func RunDependencyAuditor(ctx context.Context, app appx.App, in RepoPathInput) (map[string]any, error)

RunDependencyAuditor ports `run_dependency_auditor(repo_path)` (reasoners/recon.py:24).

func RunDosHunter 

func RunDosHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunDosHunter ports `run_dos_hunter` (reasoners/hunt.py:89).

func RunExploitHypothesizer 

func RunExploitHypothesizer(ctx context.Context, app appx.App, in ExploitInput) (map[string]any, error)

RunExploitHypothesizer ports `run_exploit_hypothesizer(repo_path, finding, data_flow, sanitization, depth)` (reasoners/prove.py:91).

func RunInjectionHunter 

func RunInjectionHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunInjectionHunter ports `run_injection_hunter` (reasoners/hunt.py:72).

func RunLogicBugsHunter 

func RunLogicBugsHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunLogicBugsHunter ports `run_logic_bugs_hunter` (reasoners/hunt.py:191): 

return await run_business_logic_hunter(repo_path=..., recon_context=...,
                                       depth=..., max_files_without_signal=...)

Python parity, two points that are easy to miss:

  • it emits NO note of its own. The note the execution shows is the "Business logic hunter starting" one the delegate emits.
  • it delegates to the REASONER, not to agents/hunt/logic.py's run_logic_hunter. reasoners/hunt.py imports run_logic_hunter and never uses it (a dead import in the Python source), so this reasoner is behaviorally indistinguishable from run_business_logic_hunter.

func RunRemediation 

func RunRemediation(ctx context.Context, app appx.App, in RemediationInput) (map[string]any, error)

RunRemediation ports `run_remediation(repo_path, finding)` (reasoners/prove.py:131) — the reasoner remediation_phase fans out over. 

finding_model = VerifiedFinding(**finding)
result = await generate_remediation(router, repo_path, finding_model)
return RemediationSuggestion.model_validate(result).model_dump()

Python parity: the note is "Remediation agent starting" — the same string run_remediation_agent uses, tags included.

The trailing `RemediationSuggestion.model_validate(result)` re-validates a value that generate_remediation already returned as a RemediationSuggestion. It is a no-op round trip; GenerateRemediation returns the typed value here, so ToMap is the whole of it.

func RunRemediationAgent 

func RunRemediationAgent(ctx context.Context, app appx.App, in RemediationAgentInput) (map[string]any, error)

RunRemediationAgent ports `run_remediation_agent(repo_path, finding, verdict, rationale)` (reasoners/prove.py:145). `finding` is a RawFinding here, unlike run_remediation's VerifiedFinding.

func RunSSRFHunter 

func RunSSRFHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunSSRFHunter ports `run_ssrf_hunter` (reasoners/hunt.py:106).

func RunSanitizationAnalyzer 

func RunSanitizationAnalyzer(ctx context.Context, app appx.App, in SanitizationInput) (map[string]any, error)

RunSanitizationAnalyzer ports `run_sanitization_analyzer(repo_path, finding, data_flow, depth)` (reasoners/prove.py:77).

func RunSecurityContextProfiler 

func RunSecurityContextProfiler(ctx context.Context, app appx.App, in ArchitectureInput) (map[string]any, error)

RunSecurityContextProfiler ports `run_security_context_profiler(repo_path, architecture)` (reasoners/recon.py:49). Same `ArchitectureMap(**architecture)` bind, same nested validation — see RunDataFlowMapper.

func RunSupplyChainHunter 

func RunSupplyChainHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunSupplyChainHunter ports `run_supply_chain_hunter` (reasoners/hunt.py:222). Depth-less hunter — see runHunter.

func RunTracer 

func RunTracer(ctx context.Context, app appx.App, in FindingDepthInput) (map[string]any, error)

RunTracer ports `run_tracer(repo_path, finding, depth)` (reasoners/prove.py:69).

Python parity: `RawFinding(**finding)` here is the STRICT constructor, not _coerce_verifier_finding — a FindingForVerifier projection sent to this reasoner raises rather than being adapted.

func RunVerdictAgent 

func RunVerdictAgent(ctx context.Context, app appx.App, in VerdictInput) (map[string]any, error)

RunVerdictAgent ports `run_verdict_agent(finding, data_flow, sanitization, exploit)` (reasoners/prove.py:109).

Python parity: the reasoner has no repo_path parameter and passes the literal "." to the agent, which never reads it.

func RunVerifier 

func RunVerifier(ctx context.Context, app appx.App, in FindingDepthInput) (map[string]any, error)

RunVerifier ports `run_verifier(repo_path, finding, depth)` (reasoners/prove.py:60) — the reasoner prove_phase fans out over.

func RunXSSHunter 

func RunXSSHunter(ctx context.Context, app appx.App, in HunterInput) (map[string]any, error)

RunXSSHunter ports `run_xss_hunter` (reasoners/hunt.py:140).

func ValidateHandlerInput 

func ValidateHandlerInput(name string, data map[string]any) (map[string]any, error)

ValidateHandlerInput is `_validate_handler_input(data, input_types)` for the reasoner called name: it returns the validated, coerced keyword map to bind, or a *HandlerInputError.

Like Python's, the returned map contains ONLY declared parameters — an undeclared key in the body is dropped rather than forwarded.

It PANICS on a name the capture does not know, for the same reason InputSchema does: every registration goes through here, so adding a reasoner without regenerating the capture fails at boot and in every test instead of silently shipping an unvalidated handler.

Types

type ArchitectureInput 

type ArchitectureInput struct {
	RepoPath     string         `json:"repo_path"`
	Architecture map[string]any `json:"architecture"`
}

ArchitectureInput ports the two deep-recon reasoner signatures: 

run_data_flow_mapper(repo_path: str, architecture: dict[str, Any])
run_security_context_profiler(repo_path: str, architecture: dict[str, Any])

`architecture` becomes `ArchitectureMap(**architecture)`. ArchitectureMap has no required field of its own, but its five lists hold models that do, so the bind goes through phases.BindArchitectureMap rather than a bare afx.Bind.

type CWEExpansionInput 

type CWEExpansionInput struct {
	ReconSummary string   `json:"recon_summary"`
	Strategies   []string `json:"strategies"`
}

CWEExpansionInput ports: 

run_cwe_expansion(recon_summary: str, strategies: list[str])

type CrossServiceInput 

type CrossServiceInput struct {
	RepoPath        string   `json:"repo_path"`
	Services        []string `json:"services"`
	FindingsSummary string   `json:"findings_summary"`
	Depth           string   `json:"depth"`
}

CrossServiceInput ports: 

run_cross_service_analyzer(repo_path: str, services: list[str],
                           findings_summary: str, depth: str)

type DastVerifierInput 

type DastVerifierInput struct {
	RepoPath       string         `json:"repo_path"`
	Finding        map[string]any `json:"finding"`
	ExploitPayload string         `json:"exploit_payload"`
	Depth          string         `json:"depth"`
}

DastVerifierInput ports: 

run_dast_verifier(repo_path: str, finding: dict[str, Any], exploit_payload: str, depth: str)

type DeduplicatorInput 

type DeduplicatorInput struct {
	Findings     []map[string]any `json:"findings"`
	ReconContext map[string]any   `json:"recon_context"`
	RepoPath     string           `json:"repo_path"`
}

DeduplicatorInput ports: 

run_deduplicator(findings: list[dict[str, Any]], recon_context: dict[str, Any], repo_path: str)

Field ORDER follows the Python signature (findings first), which is also the order the JSON keys are documented in; it has no effect on binding.

type ExploitInput 

type ExploitInput struct {
	RepoPath     string         `json:"repo_path"`
	Finding      map[string]any `json:"finding"`
	DataFlow     map[string]any `json:"data_flow"`
	Sanitization map[string]any `json:"sanitization"`
	Depth        string         `json:"depth"`
}

ExploitInput ports: 

run_exploit_hypothesizer(repo_path: str, finding: dict[str, Any],
                         data_flow: dict[str, Any],
                         sanitization: dict[str, Any], depth: str)

type FindingDepthInput 

type FindingDepthInput struct {
	RepoPath string         `json:"repo_path"`
	Finding  map[string]any `json:"finding"`
	Depth    string         `json:"depth"`
}

FindingDepthInput ports the three signatures shaped `(repo_path, finding, depth)`: 

run_dep_reachability(repo_path: str, finding: dict[str, Any], depth: str)
run_verifier(repo_path: str, finding: dict[str, Any], depth: str)
run_tracer(repo_path: str, finding: dict[str, Any], depth: str)

type Handler 

type Handler func(ctx context.Context, input map[string]any) (any, error)

Handler is one reasoner: the untyped keyword map a `.call` carries in, the reasoner's result out. It is the shape the SDK's agent.HandlerFunc had, kept so every handler below moved across unchanged.

func ValidateHandler 

func ValidateHandler(name string, h Handler) Handler

ValidateHandler wraps h with `_validate_handler_input` for the reasoner called name — the check the Python SDK ran on every request body BEFORE the handler function was entered (agent.py:3120-3134).

Exported because internal/secaf/audit runs the same layer over the audit entry's own request.

Python answered a failure with `JSONResponse(422, {"detail": safe_message})` from the endpoint, so the handler never ran. Here the check runs as the first thing inside the handler and reports the same status through an *InputError.

type HandlerInputError 

type HandlerInputError struct{ Message string }

HandlerInputError is the Go shape of `_HandlerInputError`, the ValueError subclass `_validate_handler_input` raises. Its message is the SDK-constructed `safe_message` Python puts in the 422 body verbatim.

func (*HandlerInputError) Error 

func (e *HandlerInputError) Error() string

type HuntPhaseInput 

type HuntPhaseInput struct {
	RepoPath               string          `json:"repo_path"`
	ReconContext           map[string]any  `json:"recon_context"`
	Depth                  string          `json:"depth"`
	AIGate                 json.RawMessage `json:"ai_gate"`
	MaxConcurrentHunters   int             `json:"max_concurrent_hunters"`
	EarlyStopFileThreshold int             `json:"early_stop_file_threshold"`
}

HuntPhaseInput ports: 

hunt_phase(repo_path: str, recon_context: dict[str, Any], depth: str = "standard",
           ai_gate: Any | None = None, max_concurrent_hunters: int = 4,
           early_stop_file_threshold: int = 30)

`ai_gate` is kept as a RAW JSON value. It is an object parameter that app.py's `.call` into hunt_phase never passes, but hunt_phase is a REGISTERED reasoner, so a control-plane caller can send one — and Python binds it (the SDK passes an `Any`-hinted parameter through untouched), takes the `ai_gate is not None` branch, raises AttributeError on `ai_gate.select_strategy(...)` and emits 

AI gate failed: 'dict' object has no attribute 'select_strategy', using default strategies

with tags ["hunt","ai_gate","error"] before falling back to the default strategies. phases.NewJSONAIGate turns the raw value into exactly that: nil (Python's None) for an absent key or `null`, and otherwise a gate that fails with the CPython message for that JSON type. Raw bytes rather than `any` because encoding/json decodes every number to float64, which would report "float" where CPython reports "int".

func NewHuntPhaseInput 

func NewHuntPhaseInput() HuntPhaseInput

NewHuntPhaseInput returns the Python keyword defaults.

func (*HuntPhaseInput) UnmarshalJSON 

func (h *HuntPhaseInput) UnmarshalJSON(b []byte) error

UnmarshalJSON seeds hunt_phase's three keyword defaults before decoding.

type HunterInput 

type HunterInput struct {
	RepoPath              string         `json:"repo_path"`
	ReconContext          map[string]any `json:"recon_context"`
	Depth                 string         `json:"depth"`
	MaxFilesWithoutSignal int            `json:"max_files_without_signal"`
}

HunterInput is the signature all twelve hunter reasoners share: 

run_<strategy>_hunter(repo_path: str, recon_context: dict[str, Any],
                      depth: str, max_files_without_signal: int = 30)

`depth` has NO default here (it is positional-or-keyword without one), so a request that OMITS it binds the empty string — which every downstream _normalize_depth turns into "standard".

This is the ONE rule of handler_input.go's port of the Python SDK's `_validate_handler_input` that is deliberately not reproduced: Python answers `422 Missing required field: depth` before the handler body runs (agent.py:1169-1171), because it can see that the key was absent. The Go input structs cannot — an absent scalar is indistinguishable from a zero one — so the request proceeds. Everything else that layer does (the null-on-required rejection, the int/float/str/bool coercions, the dict/list shape checks) IS reproduced; see handler_input.go.

func NewHunterInput 

func NewHunterInput() HunterInput

NewHunterInput returns the Python keyword defaults.

func (*HunterInput) UnmarshalJSON 

func (h *HunterInput) UnmarshalJSON(b []byte) error

UnmarshalJSON seeds max_files_without_signal=30 before decoding.

type InputError 

type InputError struct {
	StatusCode int
	Message    string
}

InputError is a request the reasoner refused before running: the input did not bind. StatusCode keeps the HTTP status the node answered with — 422 for the SDK-level validation Python's endpoint performs, 400 for the audit entry's own bad-input family — because that number is how a caller told "your input was wrong" from "the audit failed", and the codeaf side may want to keep drawing the same line.

func (*InputError) Error 

func (e *InputError) Error() string

type ProvePhaseInput 

type ProvePhaseInput struct {
	RepoPath             string         `json:"repo_path"`
	HuntResult           map[string]any `json:"hunt_result"`
	Depth                string         `json:"depth"`
	MaxProvers           *int           `json:"max_provers"`
	MaxConcurrentProvers int            `json:"max_concurrent_provers"`
}

ProvePhaseInput ports: 

prove_phase(repo_path: str, hunt_result: dict[str, Any], depth: str = "standard",
            max_provers: int | None = None, max_concurrent_provers: int = 3)

max_provers is a pointer: nil is Python's None (use the depth's cap), and a present 0 caps the prover fan-out at zero.

func NewProvePhaseInput 

func NewProvePhaseInput() ProvePhaseInput

NewProvePhaseInput returns the Python keyword defaults.

func (*ProvePhaseInput) UnmarshalJSON 

func (p *ProvePhaseInput) UnmarshalJSON(b []byte) error

UnmarshalJSON seeds prove_phase's keyword defaults before decoding.

type ReconPhaseInput 

type ReconPhaseInput struct {
	RepoPath string `json:"repo_path"`
	Depth    string `json:"depth"`
}

ReconPhaseInput ports: 

recon_phase(repo_path: str, depth: str = "standard")

func NewReconPhaseInput 

func NewReconPhaseInput() ReconPhaseInput

NewReconPhaseInput returns the Python keyword defaults.

func (*ReconPhaseInput) UnmarshalJSON 

func (r *ReconPhaseInput) UnmarshalJSON(b []byte) error

UnmarshalJSON seeds depth="standard" before decoding.

type Registry 

type Registry struct {
	// contains filtered or unexported fields
}

Registry is the reasoner table: name -> handler, plus the registration order and the input schema each name was registered with.

It is built once by RegisterAll and only read afterwards, so it needs no lock; a caller that registers concurrently with lookups is misusing it.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry returns an empty registry.

func (*Registry) InputSchema 

func (r *Registry) InputSchema(name string) (json.RawMessage, bool)

InputSchema returns a copy of the input schema name was registered with, and false for a name the registry does not carry.

func (*Registry) Lookup 

func (r *Registry) Lookup(name string) (Handler, bool)

Lookup returns the handler registered under name.

func (*Registry) Names 

func (r *Registry) Names() []string

Names returns a copy of the registered names, in registration order.

func (*Registry) Register 

func (r *Registry) Register(name string, h Handler, schema json.RawMessage)

Register adds one handler under name.

It PANICS on a name registered twice: the SDK's reasoner table was a map where the second registration silently won, and a collision is a wiring bug that must fail at construction, not at the first `.call` that reaches the wrong handler.

type RemediationAgentInput 

type RemediationAgentInput struct {
	RepoPath  string         `json:"repo_path"`
	Finding   map[string]any `json:"finding"`
	Verdict   string         `json:"verdict"`
	Rationale string         `json:"rationale"`
}

RemediationAgentInput ports: 

run_remediation_agent(repo_path: str, finding: dict[str, Any], verdict: str, rationale: str)

`finding` is a RawFinding here. Nothing in the pipeline calls this reasoner; it is part of the registered surface.

type RemediationInput 

type RemediationInput struct {
	RepoPath string         `json:"repo_path"`
	Finding  map[string]any `json:"finding"`
}

RemediationInput ports: 

run_remediation(repo_path: str, finding: dict[str, Any])

`finding` is a VerifiedFinding here (not a RawFinding) — this is the reasoner remediation_phase calls.

type RemediationPhaseInput 

type RemediationPhaseInput struct {
	RepoPath                  string           `json:"repo_path"`
	VerifiedFindings          []map[string]any `json:"verified_findings"`
	MaxConcurrentRemediations int              `json:"max_concurrent_remediations"`
}

RemediationPhaseInput ports: 

remediation_phase(repo_path: str, verified_findings: list[dict[str, Any]],
                  max_concurrent_remediations: int = 3)

func NewRemediationPhaseInput 

func NewRemediationPhaseInput() RemediationPhaseInput

NewRemediationPhaseInput returns the Python keyword defaults.

func (*RemediationPhaseInput) UnmarshalJSON 

func (r *RemediationPhaseInput) UnmarshalJSON(b []byte) error

UnmarshalJSON seeds max_concurrent_remediations=3 before decoding.

type RepoPathInput 

type RepoPathInput struct {
	RepoPath string `json:"repo_path"`
}

RepoPathInput is the signature shared by the three recon reasoners that take only a repository path: 

run_architecture_mapper(repo_path: str)
run_dependency_auditor(repo_path: str)
run_config_scanner(repo_path: str)

type SanitizationInput 

type SanitizationInput struct {
	RepoPath string         `json:"repo_path"`
	Finding  map[string]any `json:"finding"`
	DataFlow map[string]any `json:"data_flow"`
	Depth    string         `json:"depth"`
}

SanitizationInput ports: 

run_sanitization_analyzer(repo_path: str, finding: dict[str, Any],
                          data_flow: dict[str, Any], depth: str)

type VerdictInput 

type VerdictInput struct {
	Finding      map[string]any `json:"finding"`
	DataFlow     map[string]any `json:"data_flow"`
	Sanitization map[string]any `json:"sanitization"`
	Exploit      map[string]any `json:"exploit"`
}

VerdictInput ports: 

run_verdict_agent(finding: dict[str, Any], data_flow: dict[str, Any],
                  sanitization: dict[str, Any], exploit: dict[str, Any])

There is no repo_path parameter: the adapter passes the literal "." to run_verdict_agent, which never reads it.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.