diff --git a/pkg/executors/imagescan.go b/pkg/executors/imagescan.go new file mode 100644 index 0000000..55d6c39 --- /dev/null +++ b/pkg/executors/imagescan.go @@ -0,0 +1,319 @@ +package executors + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "github.com/lyzrai/flow/pkg/engine" + "github.com/lyzrai/flow/pkg/models" +) + +// ImageScanExecutor scans an OCI image (already pushed to a registry) for +// CVEs, malware, embedded secrets, and base-image vulnerabilities. Pairs +// with a Build node upstream — ImageScan reads `__build.image` from input +// items by default — but can also scan a hand-specified ref. +// +// This is distinct from the SAST node: SAST scans *source*, ImageScan +// scans the *built artifact*. They catch different classes of bugs (a +// vulnerable transitive dep that only appears in the final layer; a +// secret baked into a layer; a malicious base image). +// +// Tools (sibling docker containers): +// - trivy — `trivy image ` — vulns + secrets + misconfig in image +// - grype — `grype ` — Anchore's CVE scanner +// - custom — your container, your command. We mount nothing; it's your +// tool's job to pull whatever it needs. +// +// Parameters: +// - tool trivy | grype | custom (default: trivy) +// - imageRef string (optional; defaults to upstream __build.image) +// - severityThreshold LOW | MEDIUM | HIGH | CRITICAL (default: HIGH) +// - failOnFinding bool (default: true) +// - timeoutSeconds number (default: 600) +// - registryUsername string for private source registries +// - registryPassword string +// - insecure bool accept plain-HTTP / self-signed (default: true; matches local registry) +// - dockerNetwork string override compose network (default: langship-restate_default) +// - custom only: +// image string scanner image +// command string shell command (image ref is exported as $IMAGE_REF) +type ImageScanExecutor struct{} + +func (e *ImageScanExecutor) Execute(ctx context.Context, node models.NodeDef, inputs [][]models.Item, _ *engine.ExecutionContext) (map[int][]models.Item, error) { + logger := engine.NodeLoggerFromContext(ctx) + + // --- resolve image ref --- + ref := strParam(node.Parameters, "imageRef", "") + if ref == "" { + ref = imageFromBuildOutput(inputs) + } + if ref == "" { + return nil, errors.New("imageScan: no imageRef (set explicitly or wire a Build node upstream)") + } + + tool := strings.ToLower(strParam(node.Parameters, "tool", "trivy")) + threshold := strings.ToUpper(strParam(node.Parameters, "severityThreshold", "HIGH")) + failOnFinding := boolParam(node.Parameters, "failOnFinding", true) + timeoutSec := intParam(node.Parameters, "timeoutSeconds", 600) + if timeoutSec < 30 { + timeoutSec = 30 + } + if timeoutSec > 3600 { + timeoutSec = 3600 + } + insecure := boolParam(node.Parameters, "insecure", true) + username := strParam(node.Parameters, "registryUsername", "") + password := strParam(node.Parameters, "registryPassword", "") + dockerNetwork := strParam(node.Parameters, "dockerNetwork", "langship-restate_default") + + hardCtx, cancel := context.WithTimeout(ctx, time.Duration(timeoutSec)*time.Second) + defer cancel() + + // --- registry-network ref rewrite --- + // Build pushes to `registry:5000/...` (the compose service name) but the + // flow process running on the host stamps `__build.image` with that same + // ref. From the host, `registry` doesn't resolve — but we run the scanner + // inside the compose network where it does. So we leave the ref intact + // and run docker on the same network. If the ref is a localhost/127.x + // address, swap to `registry:5000` so the scanner-in-network can reach it. + scanRef := registryNetworkRef(ref) + + logger.Log(fmt.Sprintf("[imagescan:%s] scanning %s", tool, scanRef)) + + // --- optional auth via mounted docker config --- + var configMount []string + if username != "" || password != "" { + dir, cleanup, err := writeDockerConfig(scanRef, username, password) + if err != nil { + return nil, fmt.Errorf("imageScan: write docker config: %w", err) + } + defer cleanup() + configMount = []string{"-v", dir + ":/root/.docker:ro"} + } + + var ( + findings []sastFinding + toolErr error + ) + switch tool { + case "trivy": + findings, toolErr = runTrivyImage(hardCtx, scanRef, threshold, insecure, configMount, dockerNetwork, logger) + case "grype": + findings, toolErr = runGrype(hardCtx, scanRef, insecure, configMount, dockerNetwork, logger) + case "custom": + findings, toolErr = runCustomImage(hardCtx, node.Parameters, scanRef, dockerNetwork, logger) + default: + return nil, fmt.Errorf("imageScan: unknown tool %q", tool) + } + if toolErr != nil { + return nil, fmt.Errorf("imageScan (%s): %w", tool, toolErr) + } + + counts := map[string]int{} + for _, f := range findings { + counts[strings.ToUpper(f.Severity)]++ + } + logger.Log(fmt.Sprintf("[imagescan:%s] %d finding(s) %v", tool, len(findings), counts)) + + out := map[string]any{ + "tool": tool, + "imageRef": ref, // original ref users care about + "scanRef": scanRef, // rewritten ref the scanner used + "severityThreshold": threshold, + "counts": counts, + "finding_count": len(findings), + "findings": findings, + "finished_at": time.Now().UTC(), + } + + items := make([]models.Item, 0) + for _, in := range inputs { + for _, it := range in { + ci := copyItem(it) + ci["__imageScan"] = out + items = append(items, ci) + } + } + if len(items) == 0 { + items = append(items, models.Item{"__imageScan": out}) + } + + if failOnFinding && exceedsThreshold(findings, threshold) { + return nil, fmt.Errorf("imageScan (%s) failed: severity threshold %s exceeded (%v)", + tool, threshold, counts) + } + return map[int][]models.Item{0: items}, nil +} + +// --- trivy image --------------------------------------------------------- + +func runTrivyImage(ctx context.Context, ref, threshold string, insecure bool, configMount []string, network string, logger engine.NodeLogger) ([]sastFinding, error) { + args := []string{"run", "--rm", "--network", network} + args = append(args, configMount...) + args = append(args, + "aquasec/trivy:latest", + "image", "--quiet", + "--format", "json", + "--severity", severityChainAtOrAbove(threshold), + "--scanners", "vuln,secret", + ) + if insecure { + args = append(args, "--insecure") + } + args = append(args, ref) + + out, err := dockerRunCapture(ctx, args, logger) + if err != nil && len(out) == 0 { + return nil, err + } + return parseTrivy(out) // same JSON shape as `trivy fs` +} + +// --- grype --------------------------------------------------------------- + +func runGrype(ctx context.Context, ref string, insecure bool, configMount []string, network string, logger engine.NodeLogger) ([]sastFinding, error) { + args := []string{"run", "--rm", "--network", network} + args = append(args, configMount...) + if insecure { + // grype trusts the docker-config "insecureRegistries" too, but the + // quickest knob is its env var. + args = append(args, "-e", "GRYPE_REGISTRY_INSECURE_USE_HTTP=true") + args = append(args, "-e", "GRYPE_REGISTRY_INSECURE_SKIP_TLS_VERIFY=true") + } + args = append(args, + "anchore/grype:latest", + ref, + "-o", "json", + ) + out, err := dockerRunCapture(ctx, args, logger) + if err != nil && len(out) == 0 { + return nil, err + } + return parseGrype(out) +} + +type grypeReport struct { + Matches []struct { + Vulnerability struct { + ID string `json:"id"` + Severity string `json:"severity"` + } `json:"vulnerability"` + Artifact struct { + Name string `json:"name"` + Version string `json:"version"` + } `json:"artifact"` + } `json:"matches"` +} + +func parseGrype(raw []byte) ([]sastFinding, error) { + if i := bytes.IndexByte(raw, '{'); i > 0 { + raw = raw[i:] + } + var r grypeReport + if err := json.Unmarshal(raw, &r); err != nil { + return nil, fmt.Errorf("parse grype json: %w", err) + } + out := make([]sastFinding, 0, len(r.Matches)) + for _, m := range r.Matches { + out = append(out, sastFinding{ + Tool: "grype", + Severity: strings.ToUpper(m.Vulnerability.Severity), + RuleID: m.Vulnerability.ID, + Message: fmt.Sprintf("%s in %s@%s", + m.Vulnerability.ID, m.Artifact.Name, m.Artifact.Version), + }) + } + return out, nil +} + +// --- custom -------------------------------------------------------------- + +func runCustomImage(ctx context.Context, p map[string]any, ref, network string, logger engine.NodeLogger) ([]sastFinding, error) { + image := strings.TrimSpace(strParam(p, "image", "")) + command := strings.TrimSpace(strParam(p, "command", "")) + if image == "" || command == "" { + return nil, errors.New("custom imageScan requires image and command") + } + args := []string{ + "run", "--rm", "--network", network, + "-e", "IMAGE_REF=" + ref, + image, + "/bin/sh", "-c", command, + } + out, err := dockerRunCapture(ctx, args, logger) + if err != nil { + return nil, fmt.Errorf("custom scanner exit: %w (last: %s)", err, oneLineSummary(string(out))) + } + return []sastFinding{{ + Tool: "custom", + Severity: "UNKNOWN", + Message: lastLines(string(out), 20), + }}, nil +} + +// --- helpers ------------------------------------------------------------- + +// registryNetworkRef rewrites localhost-flavoured registry refs so the +// scanner-in-network can reach the bundled `registry:5000`. Anything else +// passes through. +func registryNetworkRef(ref string) string { + for _, host := range []string{"localhost:5000", "127.0.0.1:5000", "host.docker.internal:5000"} { + if strings.HasPrefix(ref, host+"/") { + return "registry:5000" + ref[len(host):] + } + } + return ref +} + +// writeDockerConfig drops a docker config.json into a temp dir so the +// scanner container picks up creds when mounted at /root/.docker. The host +// of the registry is parsed off the image ref. Returns (dir, cleanup, err). +func writeDockerConfig(ref, username, password string) (string, func(), error) { + host := registryHostFromRef(ref) + if host == "" { + return "", func() {}, errors.New("could not parse registry host from image ref") + } + dir, err := os.MkdirTemp("", "flow-imagescan-") + if err != nil { + return "", func() {}, err + } + cleanup := func() { _ = os.RemoveAll(dir) } + + // docker-config "auths" expects the credential as base64(user:pass). + authB64 := base64Encode(username + ":" + password) + cfg := fmt.Sprintf(`{"auths":{%q:{"auth":%q}}}`, host, authB64) + if err := os.WriteFile(filepath.Join(dir, "config.json"), []byte(cfg), 0o600); err != nil { + cleanup() + return "", func() {}, err + } + return dir, cleanup, nil +} + +// registryHostFromRef returns the registry hostname (and optional port) +// from an OCI ref. "registry:5000/owner/img:tag" → "registry:5000". +func registryHostFromRef(ref string) string { + i := strings.Index(ref, "/") + if i < 0 { + return "" + } + head := ref[:i] + // A bare path with no host (e.g. "library/alpine:latest") shouldn't get + // here — Docker treats those as docker.io. + if !strings.Contains(head, ".") && !strings.Contains(head, ":") && head != "localhost" { + return "" + } + return head +} + +// base64Encode wraps stdlib for symmetry with the docker-config writer. +func base64Encode(s string) string { + return base64.StdEncoding.EncodeToString([]byte(s)) +} diff --git a/pkg/executors/push.go b/pkg/executors/push.go index 63cbfdd..8dd54f9 100644 --- a/pkg/executors/push.go +++ b/pkg/executors/push.go @@ -81,6 +81,12 @@ func (e *PushExecutor) Execute(ctx context.Context, node models.NodeDef, inputs if srcImage == "" { return nil, errors.New("push: no source image (set srcImage or wire a Build node upstream)") } + // Build pushes to `registry:5000` (the compose service name), but Push + // runs in the flow process on the host where `registry` doesn't + // resolve. Rewrite to `localhost:5000` so crane.Pull can reach the + // published host port. Symmetric to the rewrite ImageScan does in the + // other direction. + srcImage = pushSourceHostRef(srcImage) if _, err := name.ParseReference(srcImage); err != nil { return nil, fmt.Errorf("push: invalid src %q: %w", srcImage, err) } @@ -313,6 +319,18 @@ func defaultName(explicit, registry string, idx int) string { return host } +// pushSourceHostRef rewrites a compose-internal registry hostname to its +// host-published equivalent so the host-running flow process can actually +// reach the registry. Only the most common pair we set up +// (`registry:5000` → `localhost:5000`) is rewritten; everything else +// passes through unchanged. +func pushSourceHostRef(ref string) string { + if strings.HasPrefix(ref, "registry:5000/") { + return "localhost:5000" + ref[len("registry:5000"):] + } + return ref +} + func anyToBool(v any, def bool) bool { switch x := v.(type) { case bool: diff --git a/pkg/executors/registry.go b/pkg/executors/registry.go index 7008a34..2e5003c 100644 --- a/pkg/executors/registry.go +++ b/pkg/executors/registry.go @@ -56,9 +56,11 @@ func RegisterAll(deps ...RegistryDeps) { // Human-in-the-loop Register("flow-nodes-base.waitForApproval", &ApprovalExecutor{}) - // CI/CD primitives — Build + Push are real; the rest are stubs for now. + // CI/CD primitives — Build + Push + SAST are real; the rest are stubs. Register("flow-nodes-base.build", &BuildExecutor{Agents: d.Agents}) Register("flow-nodes-base.push", &PushExecutor{}) + Register("flow-nodes-base.sast", &SastExecutor{Agents: d.Agents}) + Register("flow-nodes-base.imageScan", &ImageScanExecutor{}) Register("flow-nodes-base.test", &TestExecutor{}) Register("flow-nodes-base.eval", &EvalExecutor{}) Register("flow-nodes-base.policy", &PolicyExecutor{}) diff --git a/pkg/executors/sast.go b/pkg/executors/sast.go new file mode 100644 index 0000000..8626496 --- /dev/null +++ b/pkg/executors/sast.go @@ -0,0 +1,629 @@ +package executors + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os/exec" + "strings" + "time" + + "github.com/lyzrai/flow/pkg/engine" + "github.com/lyzrai/flow/pkg/models" + "github.com/lyzrai/flow/pkg/storage" +) + +// SastExecutor runs static-analysis security scanning against the agent's +// source repo. The `tool` parameter selects which scanner runs; each tool +// is just a sibling docker container we shell out to (`docker run --rm`), +// except `sonar` which uploads to SonarCloud and polls for the quality +// gate verdict. +// +// Trigger payload requirements (same as Build): +// - agentId string (provided by dispatchAgent / webhook) +// - commit string commit SHA to check out (optional) +// - ref string ref override +// +// Common parameters: +// - tool trivy | semgrep | gitleaks | sonar | custom (default: trivy) +// - severityThreshold LOW | MEDIUM | HIGH | CRITICAL (default: HIGH) +// - failOnFinding bool (default: true) +// - timeoutSeconds number (default: 600) +// +// Tool-specific parameters: +// - trivy / semgrep / gitleaks no extra config — sane defaults +// - sonar: +// sonarHost default https://sonarcloud.io +// organization required +// projectKey required +// sonarToken required (SONAR_TOKEN) +// branchName optional, defaults to the agent ref +// - custom: +// image OCI image to run (required) +// command shell command inside the container (required) +type SastExecutor struct { + Agents storage.AgentStore +} + +func (e *SastExecutor) Execute(ctx context.Context, node models.NodeDef, inputs [][]models.Item, _ *engine.ExecutionContext) (map[int][]models.Item, error) { + if e.Agents == nil { + return nil, errors.New("sast: AgentStore not configured") + } + + logger := engine.NodeLoggerFromContext(ctx) + + trigger := firstItem(inputs) + agentID, _ := trigger["agentId"].(string) + if agentID == "" { + return nil, errors.New("sast: trigger payload missing agentId") + } + a, err := e.Agents.Get(ctx, agentID) + if err != nil { + return nil, fmt.Errorf("sast: load agent %q: %w", agentID, err) + } + + tool := strings.ToLower(strParam(node.Parameters, "tool", "trivy")) + threshold := strings.ToUpper(strParam(node.Parameters, "severityThreshold", "HIGH")) + failOnFinding := boolParam(node.Parameters, "failOnFinding", true) + timeoutSec := intParam(node.Parameters, "timeoutSeconds", 600) + if timeoutSec < 30 { + timeoutSec = 30 + } + if timeoutSec > 3600 { + timeoutSec = 3600 + } + + commitSHA, _ := trigger["commit"].(string) + ref := stripRefsHeads(strFirst(strFromAny(trigger["ref"]), a.Ref, "main")) + + cloneDir, cleanup, err := cloneRepo(ctx, a, ref, commitSHA, time.Duration(timeoutSec)*time.Second) + if err != nil { + return nil, err + } + defer cleanup() + + hardCtx, cancel := context.WithTimeout(ctx, time.Duration(timeoutSec)*time.Second) + defer cancel() + + logger.Log(fmt.Sprintf("[sast:%s] running against %s @ %s", tool, a.Name, ref)) + + var ( + findings []sastFinding + summary map[string]any + toolErr error + ) + switch tool { + case "trivy": + findings, toolErr = runTrivy(hardCtx, cloneDir, threshold, logger) + case "semgrep": + findings, toolErr = runSemgrep(hardCtx, cloneDir, logger) + case "gitleaks": + findings, toolErr = runGitleaks(hardCtx, cloneDir, logger) + case "sonar": + summary, toolErr = runSonarCloud(hardCtx, node.Parameters, a, cloneDir, ref, commitSHA, logger) + case "custom": + findings, toolErr = runCustom(hardCtx, node.Parameters, cloneDir, logger) + default: + return nil, fmt.Errorf("sast: unknown tool %q", tool) + } + + if toolErr != nil { + return nil, fmt.Errorf("sast (%s): %w", tool, toolErr) + } + + // Counts by severity. + counts := map[string]int{} + for _, f := range findings { + counts[strings.ToUpper(f.Severity)]++ + } + + out := map[string]any{ + "tool": tool, + "agent_id": agentID, + "agent_name": a.Name, + "ref": ref, + "commit": commitSHA, + "severityThreshold": threshold, + "counts": counts, + "finding_count": len(findings), + "finished_at": time.Now().UTC(), + } + if findings != nil { + out["findings"] = findings + } + if summary != nil { + // Sonar mode replaces per-finding output with a server-side + // quality-gate summary. + for k, v := range summary { + out[k] = v + } + } + + logger.Log(fmt.Sprintf("[sast:%s] done — %d finding(s)", tool, len(findings))) + + // Decide pass/fail. + failed := false + if failOnFinding { + if tool == "sonar" { + // Sonar passes/fails on its quality-gate verdict. + if status, _ := summary["qualityGate"].(string); strings.ToUpper(status) == "ERROR" { + failed = true + } + } else if exceedsThreshold(findings, threshold) { + failed = true + } + } + + items := make([]models.Item, 0) + for _, in := range inputs { + for _, it := range in { + ci := copyItem(it) + ci["__sast"] = out + items = append(items, ci) + } + } + if len(items) == 0 { + items = append(items, models.Item{"__sast": out}) + } + + if failed { + return nil, fmt.Errorf("sast (%s) failed: severity threshold %s exceeded (%v)", + tool, threshold, counts) + } + return map[int][]models.Item{0: items}, nil +} + +// sastFinding is the normalized shape we emit on output items. +type sastFinding struct { + Tool string `json:"tool"` + Severity string `json:"severity"` + RuleID string `json:"ruleId,omitempty"` + File string `json:"file,omitempty"` + Line int `json:"line,omitempty"` + Message string `json:"message"` +} + +// --- trivy ---------------------------------------------------------------- + +// trivy filesystem scan: vulns + secrets + IaC misconfig in one shot. +// We scan the host clone path by mounting it read-only into the trivy +// container. Output is JSON; we parse the few fields we render. +func runTrivy(ctx context.Context, dir, threshold string, logger engine.NodeLogger) ([]sastFinding, error) { + // Restrict scan to threshold + above so trivy doesn't dump 5000 LOW + // findings. Trivy understands a comma-separated severity list. + sev := severityChainAtOrAbove(threshold) + args := []string{ + "run", "--rm", + "-v", dir + ":/src:ro", + "aquasec/trivy:latest", + "fs", "--quiet", + "--format", "json", + "--severity", sev, + "--scanners", "vuln,secret,misconfig", + "/src", + } + out, err := dockerRunCapture(ctx, args, logger) + if err != nil && len(out) == 0 { + return nil, err + } + return parseTrivy(out) +} + +// trivyReport is the minimal shape of `trivy fs --format json`. +type trivyReport struct { + Results []struct { + Target string `json:"Target"` + Class string `json:"Class"` + Vulnerabilities []struct { + VulnerabilityID string `json:"VulnerabilityID"` + PkgName string `json:"PkgName"` + InstalledVersion string `json:"InstalledVersion"` + Severity string `json:"Severity"` + Title string `json:"Title"` + } `json:"Vulnerabilities,omitempty"` + Secrets []struct { + RuleID string `json:"RuleID"` + Severity string `json:"Severity"` + Title string `json:"Title"` + StartLine int `json:"StartLine"` + } `json:"Secrets,omitempty"` + Misconfigurations []struct { + ID string `json:"ID"` + Severity string `json:"Severity"` + Title string `json:"Title"` + } `json:"Misconfigurations,omitempty"` + } `json:"Results"` +} + +func parseTrivy(raw []byte) ([]sastFinding, error) { + // Trivy may emit logs on stderr that bleed into combined output; find the + // first '{' to start parsing JSON. + if i := bytes.IndexByte(raw, '{'); i > 0 { + raw = raw[i:] + } + var r trivyReport + if err := json.Unmarshal(raw, &r); err != nil { + return nil, fmt.Errorf("parse trivy json: %w", err) + } + var findings []sastFinding + for _, res := range r.Results { + for _, v := range res.Vulnerabilities { + findings = append(findings, sastFinding{ + Tool: "trivy", + Severity: v.Severity, + RuleID: v.VulnerabilityID, + File: res.Target, + Message: fmt.Sprintf("%s in %s@%s — %s", + v.VulnerabilityID, v.PkgName, v.InstalledVersion, v.Title), + }) + } + for _, s := range res.Secrets { + findings = append(findings, sastFinding{ + Tool: "trivy", + Severity: s.Severity, + RuleID: s.RuleID, + File: res.Target, + Line: s.StartLine, + Message: s.Title, + }) + } + for _, m := range res.Misconfigurations { + findings = append(findings, sastFinding{ + Tool: "trivy", + Severity: m.Severity, + RuleID: m.ID, + File: res.Target, + Message: m.Title, + }) + } + } + return findings, nil +} + +// --- semgrep -------------------------------------------------------------- + +func runSemgrep(ctx context.Context, dir string, logger engine.NodeLogger) ([]sastFinding, error) { + args := []string{ + "run", "--rm", + "-v", dir + ":/src:ro", + "-w", "/src", + "returntocorp/semgrep:latest", + "semgrep", "scan", + "--config", "auto", // pulls Semgrep's curated registry rules + "--json", "--quiet", + } + out, err := dockerRunCapture(ctx, args, logger) + if err != nil && len(out) == 0 { + return nil, err + } + return parseSemgrep(out) +} + +type semgrepReport struct { + Results []struct { + CheckID string `json:"check_id"` + Path string `json:"path"` + Start struct { + Line int `json:"line"` + } `json:"start"` + Extra struct { + Severity string `json:"severity"` + Message string `json:"message"` + } `json:"extra"` + } `json:"results"` +} + +func parseSemgrep(raw []byte) ([]sastFinding, error) { + if i := bytes.IndexByte(raw, '{'); i > 0 { + raw = raw[i:] + } + var r semgrepReport + if err := json.Unmarshal(raw, &r); err != nil { + return nil, fmt.Errorf("parse semgrep json: %w", err) + } + out := make([]sastFinding, 0, len(r.Results)) + for _, x := range r.Results { + out = append(out, sastFinding{ + Tool: "semgrep", + Severity: normalizeSemgrepSev(x.Extra.Severity), + RuleID: x.CheckID, + File: x.Path, + Line: x.Start.Line, + Message: x.Extra.Message, + }) + } + return out, nil +} + +// Semgrep uses ERROR/WARNING/INFO; map to our LOW/MEDIUM/HIGH/CRITICAL. +func normalizeSemgrepSev(s string) string { + switch strings.ToUpper(s) { + case "ERROR": + return "HIGH" + case "WARNING": + return "MEDIUM" + case "INFO": + return "LOW" + } + return strings.ToUpper(s) +} + +// --- gitleaks ------------------------------------------------------------- + +func runGitleaks(ctx context.Context, dir string, logger engine.NodeLogger) ([]sastFinding, error) { + args := []string{ + "run", "--rm", + "-v", dir + ":/src:ro", + "zricethezav/gitleaks:latest", + "detect", "--source=/src", + "--no-git", // we're scanning the working tree, not git history + "--report-format=json", "--report-path=/dev/stdout", + "--no-banner", + } + out, err := dockerRunCapture(ctx, args, logger) + if err != nil && len(out) == 0 { + return nil, err + } + return parseGitleaks(out) +} + +type gitleaksFinding struct { + RuleID string `json:"RuleID"` + Description string `json:"Description"` + File string `json:"File"` + StartLine int `json:"StartLine"` + Match string `json:"Match"` +} + +func parseGitleaks(raw []byte) ([]sastFinding, error) { + // gitleaks --report-path=/dev/stdout emits a JSON array. + if i := bytes.IndexByte(raw, '['); i > 0 { + raw = raw[i:] + } + var arr []gitleaksFinding + if err := json.Unmarshal(raw, &arr); err != nil { + // gitleaks prints "no leaks found" sometimes; treat parse failure + // without a leading '[' as zero findings. + return nil, nil + } + out := make([]sastFinding, 0, len(arr)) + for _, g := range arr { + out = append(out, sastFinding{ + Tool: "gitleaks", + Severity: "HIGH", // any leaked secret is high severity + RuleID: g.RuleID, + File: g.File, + Line: g.StartLine, + Message: g.Description, + }) + } + return out, nil +} + +// --- sonar (SonarCloud) --------------------------------------------------- + +// runSonarCloud uploads the workspace to SonarCloud via the official scanner +// container and polls the v2 quality-gate API for the verdict. Returns a +// summary that includes the quality gate status (OK | WARN | ERROR), the +// dashboard URL, and the underlying analysis ID for traceability. +func runSonarCloud(ctx context.Context, p map[string]any, a *storage.Agent, cloneDir, ref, commit string, logger engine.NodeLogger) (map[string]any, error) { + host := strParam(p, "sonarHost", "https://sonarcloud.io") + org := strings.TrimSpace(strParam(p, "organization", "")) + projectKey := strings.TrimSpace(strParam(p, "projectKey", "")) + token := strings.TrimSpace(strParam(p, "sonarToken", "")) + branch := strFirst(strParam(p, "branchName", ""), ref, "main") + + if org == "" || projectKey == "" || token == "" { + return nil, errors.New("sonar requires organization, projectKey, and sonarToken") + } + + args := []string{ + "run", "--rm", + "-e", "SONAR_HOST_URL=" + host, + "-e", "SONAR_TOKEN=" + token, + "-v", cloneDir + ":/usr/src:ro", + "-w", "/usr/src", + "sonarsource/sonar-scanner-cli:latest", + "-Dsonar.organization=" + org, + "-Dsonar.projectKey=" + projectKey, + "-Dsonar.sources=.", + "-Dsonar.branch.name=" + branch, + } + if commit != "" { + args = append(args, "-Dsonar.scm.revision="+commit) + } + if _, err := dockerRunCapture(ctx, args, logger); err != nil { + return nil, fmt.Errorf("sonar-scanner: %w", err) + } + + // Poll for the quality gate verdict — analysis is async server-side. + logger.Log("[sast:sonar] waiting for quality gate verdict…") + gate, err := pollSonarGate(ctx, host, org, projectKey, branch, token) + if err != nil { + return nil, fmt.Errorf("sonar quality gate: %w", err) + } + logger.Log(fmt.Sprintf("[sast:sonar] quality gate: %s", gate)) + + dashboardURL := fmt.Sprintf("%s/project/overview?id=%s", + strings.TrimRight(host, "/"), url.QueryEscape(projectKey)) + return map[string]any{ + "qualityGate": gate, + "dashboardUrl": dashboardURL, + "projectKey": projectKey, + "branchName": branch, + }, nil +} + +// pollSonarGate polls /api/qualitygates/project_status until SonarCloud +// returns a non-NONE / non-PENDING verdict. Bounded by the parent ctx +// (the executor's hard timeout). +func pollSonarGate(ctx context.Context, host, org, projectKey, branch, token string) (string, error) { + endpoint := fmt.Sprintf("%s/api/qualitygates/project_status?projectKey=%s&branch=%s", + strings.TrimRight(host, "/"), + url.QueryEscape(projectKey), + url.QueryEscape(branch)) + + deadline := time.NewTicker(5 * time.Second) + defer deadline.Stop() + + httpc := &http.Client{Timeout: 15 * time.Second} + for { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return "", err + } + req.SetBasicAuth(token, "") // SonarCloud convention + req.Header.Set("Accept", "application/json") + resp, err := httpc.Do(req) + if err != nil { + return "", err + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("status %d: %s", resp.StatusCode, oneLineSummary(string(body))) + } + var r struct { + ProjectStatus struct { + Status string `json:"status"` // OK | WARN | ERROR | NONE + } `json:"projectStatus"` + } + if err := json.Unmarshal(body, &r); err != nil { + return "", err + } + switch strings.ToUpper(r.ProjectStatus.Status) { + case "OK", "WARN", "ERROR": + return r.ProjectStatus.Status, nil + } + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-deadline.C: + } + } +} + +// --- custom --------------------------------------------------------------- + +func runCustom(ctx context.Context, p map[string]any, dir string, logger engine.NodeLogger) ([]sastFinding, error) { + image := strings.TrimSpace(strParam(p, "image", "")) + command := strings.TrimSpace(strParam(p, "command", "")) + if image == "" || command == "" { + return nil, errors.New("custom tool requires image and command") + } + args := []string{ + "run", "--rm", + "-v", dir + ":/src:ro", + "-w", "/src", + image, + "/bin/sh", "-c", command, + } + out, err := dockerRunCapture(ctx, args, logger) + if err != nil { + return nil, fmt.Errorf("custom scanner exit: %w (last: %s)", err, oneLineSummary(string(out))) + } + // We don't parse arbitrary tool output — just emit a single finding + // of unknown severity carrying the tail. Users wiring a real tool can + // switch to `tool: trivy` etc., or post-process via downstream nodes. + return []sastFinding{{ + Tool: "custom", + Severity: "UNKNOWN", + Message: lastLines(string(out), 20), + }}, nil +} + +// --- helpers -------------------------------------------------------------- + +// dockerRunCapture spawns a `docker run …` subprocess, streams stdout/stderr +// through the NodeLogger so the UI sees live output, and returns the full +// stdout as bytes for downstream JSON parsing. +func dockerRunCapture(ctx context.Context, args []string, logger engine.NodeLogger) ([]byte, error) { + logger.Log("$ docker " + strings.Join(args, " ")) + + cmd := exec.CommandContext(ctx, "docker", args...) + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, err + } + stderr, err := cmd.StderrPipe() + if err != nil { + return nil, err + } + if err := cmd.Start(); err != nil { + return nil, err + } + + // Capture stdout (the JSON / report) into a buffer; stream stderr to + // the logger so users see it live. + var buf bytes.Buffer + doneOut := make(chan error, 1) + go func() { + _, e := io.Copy(&buf, io.TeeReader(stdout, &lineLogger{logger: logger, prefix: ""})) + doneOut <- e + }() + go func() { + _, _ = io.Copy(&lineLogger{logger: logger, prefix: ""}, stderr) + }() + <-doneOut + + werr := cmd.Wait() + return buf.Bytes(), werr +} + +// lineLogger is an io.Writer that splits incoming bytes on '\n' and forwards +// each non-empty line to a NodeLogger. Used so a docker subprocess's output +// streams live into the SSE feed. +type lineLogger struct { + logger engine.NodeLogger + prefix string + buf []byte +} + +func (l *lineLogger) Write(p []byte) (int, error) { + l.buf = append(l.buf, p...) + for { + i := bytes.IndexByte(l.buf, '\n') + if i < 0 { + return len(p), nil + } + line := strings.TrimRight(string(l.buf[:i]), "\r") + if line != "" { + l.logger.Log(l.prefix + line) + } + l.buf = l.buf[i+1:] + } +} + +// severityChainAtOrAbove returns a comma-list of severities at or above +// the given threshold, in trivy's expected casing. +func severityChainAtOrAbove(threshold string) string { + chain := []string{"LOW", "MEDIUM", "HIGH", "CRITICAL"} + t := strings.ToUpper(threshold) + for i, s := range chain { + if s == t { + return strings.Join(chain[i:], ",") + } + } + return "HIGH,CRITICAL" +} + +// exceedsThreshold returns true if any finding's severity is at or above +// the threshold. Used to gate the run when failOnFinding is true. +func exceedsThreshold(findings []sastFinding, threshold string) bool { + rank := map[string]int{"LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4} + tr := rank[strings.ToUpper(threshold)] + if tr == 0 { + tr = 3 // default HIGH + } + for _, f := range findings { + if rank[strings.ToUpper(f.Severity)] >= tr { + return true + } + } + return false +} diff --git a/web/app/executions/view/page.tsx b/web/app/executions/view/page.tsx index 5e44f2a..ec05a13 100644 --- a/web/app/executions/view/page.tsx +++ b/web/app/executions/view/page.tsx @@ -438,6 +438,9 @@ function NodeRow(props: { const buildSummary = extractBuildSummary(node.name, statusOutputs); const triggerSummary = extractTriggerSummary(node.name, statusOutputs); + const sastSummary = extractScanSummary(node.name, statusOutputs, "__sast"); + const imageScanSummary = extractScanSummary(node.name, statusOutputs, "__imageScan"); + const pushSummary = extractPushSummary(node.name, statusOutputs); return (
@@ -510,6 +513,41 @@ function NodeRow(props: {
)} + {sastSummary && ( + + )} + {imageScanSummary && ( + + )} + + {pushSummary && ( +
+
+ src: + {pushSummary.src} +
+ {pushSummary.copies.map((c, i) => ( +
+ + → {c.name || c.registry}:{" "} + + {c.imageRef} + {c.error ? ( + — {c.error} + ) : c.digest ? ( + + {" "} + ✓{" "} + + {c.digest.slice(0, 19)}… + + + ) : null} +
+ ))} +
+ )} + {/* Build log disclosure — visible for all node types that emitted lines */} {(liveLog?.length || isTerminal) && ( ; + findingCount: number; + findings: ScanFinding[]; +}; + +function extractScanSummary( + nodeName: string, + outputs: Record | undefined, + key: "__sast" | "__imageScan" +): ScanSummary | null { + if (!outputs) return null; + const port = ((outputs[nodeName] as Record) || {})["0"]; + if (!Array.isArray(port) || port.length === 0) return null; + const item = port[0] as Record; + const blob = item[key] as Record | undefined; + if (!blob) return null; + + const counts: Record = {}; + if (blob.counts && typeof blob.counts === "object") { + for (const [k, v] of Object.entries(blob.counts as Record)) { + if (typeof v === "number") counts[k.toUpperCase()] = v; + } + } + const findings = Array.isArray(blob.findings) ? (blob.findings as ScanFinding[]) : []; + return { + tool: typeof blob.tool === "string" ? blob.tool : undefined, + imageRef: typeof blob.imageRef === "string" ? blob.imageRef : undefined, + scanRef: typeof blob.scanRef === "string" ? blob.scanRef : undefined, + qualityGate: typeof blob.qualityGate === "string" ? blob.qualityGate : undefined, + dashboardUrl: typeof blob.dashboardUrl === "string" ? blob.dashboardUrl : undefined, + threshold: + typeof blob.severityThreshold === "string" + ? (blob.severityThreshold as string) + : undefined, + counts, + findingCount: + typeof blob.finding_count === "number" + ? (blob.finding_count as number) + : findings.length, + findings, + }; +} + +type PushSummary = { + src: string; + copies: { name?: string; registry?: string; imageRef: string; digest?: string; error?: string }[]; +}; + +function extractPushSummary( + nodeName: string, + outputs?: Record +): PushSummary | null { + if (!outputs) return null; + const port = ((outputs[nodeName] as Record) || {})["0"]; + if (!Array.isArray(port) || port.length === 0) return null; + const item = port[0] as Record; + const blob = item.__push as Record | undefined; + if (!blob) return null; + const copies = Array.isArray(blob.copies) + ? (blob.copies as Record[]).map((c) => ({ + name: typeof c.name === "string" ? c.name : undefined, + registry: typeof c.registry === "string" ? c.registry : undefined, + imageRef: typeof c.imageRef === "string" ? c.imageRef : "", + digest: typeof c.digest === "string" ? c.digest : undefined, + error: typeof c.error === "string" && c.error ? c.error : undefined, + })) + : []; + // Single-target legacy shape — promote dst to a one-entry copies array. + if (copies.length === 0 && typeof blob.dst === "string") { + copies.push({ + name: undefined, + registry: undefined, + imageRef: blob.dst as string, + digest: typeof blob.digest === "string" ? (blob.digest as string) : undefined, + error: undefined, + }); + } + return { + src: typeof blob.src === "string" ? (blob.src as string) : "", + copies, + }; +} + +const SEVERITY_ORDER = ["CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN"] as const; + +function severityClass(sev: string): string { + switch (sev.toUpperCase()) { + case "CRITICAL": + return "bg-rose-600/20 text-rose-700 dark:text-rose-400 border-rose-600/40"; + case "HIGH": + return "bg-orange-500/20 text-orange-700 dark:text-orange-400 border-orange-500/40"; + case "MEDIUM": + return "bg-amber-500/20 text-amber-700 dark:text-amber-400 border-amber-500/40"; + case "LOW": + return "bg-sky-500/20 text-sky-700 dark:text-sky-400 border-sky-500/40"; + default: + return "bg-muted text-muted-foreground border-border"; + } +} + +function ScanResults({ + summary, + kind, +}: { + summary: ScanSummary; + kind: "sast" | "imageScan"; +}) { + const [open, setOpen] = useState(false); + + const total = summary.findingCount; + const showFindings = summary.findings && summary.findings.length > 0; + const isSonar = summary.tool === "sonar"; + const headerLabel = kind === "sast" ? "SAST" : "Image scan"; + + return ( +
+
+
+ + {headerLabel} + + {summary.tool && ( + + {summary.tool} + + )} + {summary.threshold && ( + + threshold {summary.threshold} + + )} + {/* Severity counts pills */} + {SEVERITY_ORDER.map((sev) => { + const n = summary.counts[sev] || 0; + if (n === 0) return null; + return ( + + {sev.toLowerCase()} {n} + + ); + })} + {total === 0 && !isSonar && ( + + clean + + )} + {isSonar && summary.qualityGate && ( + + gate {summary.qualityGate} + + )} +
+ {showFindings && ( + + )} + {summary.dashboardUrl && ( + + open dashboard ↗ + + )} +
+ + {open && showFindings && ( +
+
+ + + + + + + + + + + {summary.findings.slice(0, 100).map((f, i) => ( + + + + + + + ))} + +
severityrulewheremessage
+ + {(f.severity ?? "?").toLowerCase()} + + + {f.ruleId ?? ""} + + {f.file ? `${f.file}${f.line ? ":" + f.line : ""}` : ""} + {f.message ?? ""}
+ {summary.findings.length > 100 && ( +
+ + {summary.findings.length - 100} more (truncated) +
+ )} +
+
+ )} +
+ ); +} diff --git a/web/components/canvas/node-form.tsx b/web/components/canvas/node-form.tsx index 36aace9..306b036 100644 --- a/web/components/canvas/node-form.tsx +++ b/web/components/canvas/node-form.tsx @@ -31,6 +31,10 @@ export function NodeForm({ node, onChange }: NodeFormProps) { return ; case "flow-nodes-base.waitForApproval": return ; + case "flow-nodes-base.sast": + return ; + case "flow-nodes-base.imageScan": + return ; case "flow-nodes-base.push": return ; case "flow-nodes-base.deploy": @@ -50,6 +54,8 @@ export function hasTypedForm(type: string): boolean { return [ "flow-nodes-base.trigger", "flow-nodes-base.build", + "flow-nodes-base.sast", + "flow-nodes-base.imageScan", "flow-nodes-base.push", "flow-nodes-base.test", "flow-nodes-base.eval", @@ -483,6 +489,388 @@ function ApprovalForm({ node, onChange }: NodeFormProps) { ); } +function SastForm({ node, onChange }: NodeFormProps) { + const tool = getString(node, "tool", "trivy"); + const threshold = getString(node, "severityThreshold", "HIGH"); + const failOnFinding = getBool(node, "failOnFinding", true); + const timeout = getNumber(node, "timeoutSeconds", 600); + + // sonar + const sonarHost = getString(node, "sonarHost", "https://sonarcloud.io"); + const organization = getString(node, "organization", ""); + const projectKey = getString(node, "projectKey", ""); + const sonarToken = getString(node, "sonarToken", ""); + const branchName = getString(node, "branchName", ""); + + // custom + const image = getString(node, "image", ""); + const command = getString(node, "command", ""); + + return ( +
+
+ + +
+ +
+
+ + +
+
+ + + onChange(setParam(node, "timeoutSeconds", Number(e.target.value))) + } + /> +
+
+ +
+ + onChange(setParam(node, "failOnFinding", e.target.checked)) + } + className="size-3.5" + /> + +
+ + {tool === "sonar" && ( +
+
+ SonarCloud +
+
+ + + onChange(setParam(node, "sonarHost", e.target.value)) + } + placeholder="https://sonarcloud.io" + className="font-mono text-xs" + /> +
+
+
+ + + onChange(setParam(node, "organization", e.target.value)) + } + placeholder="my-org" + className="font-mono text-xs" + /> +
+
+ + + onChange(setParam(node, "projectKey", e.target.value)) + } + placeholder="my-org_my-agent" + className="font-mono text-xs" + /> +
+
+
+ + + onChange(setParam(node, "sonarToken", e.target.value)) + } + placeholder="SONAR_TOKEN (User → My Account → Security)" + autoComplete="new-password" + className="font-mono text-xs" + /> +
+
+ + + onChange(setParam(node, "branchName", e.target.value)) + } + placeholder="(uses agent ref by default)" + className="font-mono text-xs" + /> +
+

+ Quality-gate failure flips the run to failed. We + also link the dashboard URL on the run page. +

+
+ )} + + {tool === "custom" && ( +
+
+ Custom scanner +
+
+ + + onChange(setParam(node, "image", e.target.value)) + } + placeholder="ghcr.io/owner/scanner:latest" + className="font-mono text-xs" + /> +
+
+ +