From 8c222113c1b350720796821dd4a77b628b0b3d29 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alex=20S=C3=A1nchez?= Date: Tue, 25 Aug 2026 11:05:53 -0600 Subject: [PATCH 1/6] feat[backend](soar): DAG-based flow engine with pluggable executors --- backend/database/migrations.go | 2 + backend/modules.go | 15 +- backend/modules/mcp/tools_soar.go | 65 ++-- backend/modules/soar/connectors/repository.go | 54 +++ backend/modules/soar/domain/execution.go | 112 +++++- backend/modules/soar/domain/flow.go | 192 +++++++++-- backend/modules/soar/dto/execution.go | 7 + backend/modules/soar/dto/rule.go | 76 +++-- backend/modules/soar/executor/executor.go | 40 +++ backend/modules/soar/executor/http.go | 112 ++++++ backend/modules/soar/executor/llm.go | 239 +++++++++++++ backend/modules/soar/executor/notify.go | 58 ++++ backend/modules/soar/executor/selectexec.go | 61 ++++ backend/modules/soar/executor/shell.go | 117 +++++++ backend/modules/soar/module.go | 21 +- .../modules/soar/repository/execution_pg.go | 149 +++++++- .../modules/soar/repository/flow_run_pg.go | 86 +++++ .../soar/usecase/assemble_chain_test.go | 56 --- backend/modules/soar/usecase/dispatch.go | 321 +++++++++++++----- backend/modules/soar/usecase/execution.go | 183 +++++----- backend/modules/soar/usecase/flow_store.go | 8 +- backend/modules/soar/usecase/flow_writer.go | 129 ++++++- .../modules/soar/usecase/flow_writer_test.go | 104 ++++++ backend/modules/soar/usecase/interpolate.go | 110 ++++++ .../modules/soar/usecase/interpolate_test.go | 63 ++++ backend/modules/soar/usecase/rule.go | 87 +++-- 26 files changed, 2070 insertions(+), 397 deletions(-) create mode 100644 backend/modules/soar/executor/executor.go create mode 100644 backend/modules/soar/executor/http.go create mode 100644 backend/modules/soar/executor/llm.go create mode 100644 backend/modules/soar/executor/notify.go create mode 100644 backend/modules/soar/executor/selectexec.go create mode 100644 backend/modules/soar/executor/shell.go create mode 100644 backend/modules/soar/repository/flow_run_pg.go delete mode 100644 backend/modules/soar/usecase/assemble_chain_test.go create mode 100644 backend/modules/soar/usecase/flow_writer_test.go create mode 100644 backend/modules/soar/usecase/interpolate.go create mode 100644 backend/modules/soar/usecase/interpolate_test.go diff --git a/backend/database/migrations.go b/backend/database/migrations.go index 73a5c4421..1d1684f1a 100644 --- a/backend/database/migrations.go +++ b/backend/database/migrations.go @@ -50,6 +50,8 @@ func Models() []any { alerts_domain.AlertTag{}, alerts_domain.AlertTagRule{}, arr_domain.SoarExecution{}, + arr_domain.SoarFlowRun{}, + arr_domain.SoarExecutionEdge{}, arr_domain.SoarVariable{}, compliance_domain.ReportSchedule{}, compliance_domain.Report{}, diff --git a/backend/modules.go b/backend/modules.go index fccf6f59f..1be83bd8f 100644 --- a/backend/modules.go +++ b/backend/modules.go @@ -198,7 +198,16 @@ func initModules(db *gorm.DB, cfg *config) *modules { agentClient = nil } - soarMod := soar.NewModule(db, agentClient, signer, cipher, tenantLister) + // SOC-AI client is built here so the SOAR LLM executors can share it. + // socai.NewModule below reuses the same base URL/key. + socAIClient := socai.NewSocAIClient(cfg.socAIBaseURL, cfg.internalKey) + // Notifications module is built early so its usecase can back the SOAR + // notify executor. Its own dependencies (db + audit logger + leases) are + // already available at this point. + notificationsMod := notifications.NewModule(db, auditMod.Logger(), joblease.New(db), + env.Int("NOTIFICATIONS_READ_RETENTION_DAYS", 30, false), + env.Int("NOTIFICATIONS_RETENTION_DAYS", 365, false)) + soarMod := soar.NewModule(db, agentClient, signer, cipher, socAIClient, notificationsMod.Producer(), tenantLister) eventProcessingMod := eventprocessing.NewModule(db, events, auditMod.Logger(), cfg.playgroundBaseURL, cfg.internalKey) alertsMod.SetCorrelationResolver(eventProcessingMod) @@ -213,10 +222,6 @@ func initModules(db *gorm.DB, cfg *config) *modules { } datasourcesMod := datasources.NewModule(dsUC, dsReconciler, agentClient) - notificationsMod := notifications.NewModule(db, auditMod.Logger(), joblease.New(db), - env.Int("NOTIFICATIONS_READ_RETENTION_DAYS", 30, false), - env.Int("NOTIFICATIONS_RETENTION_DAYS", 365, false)) - iam_handler.AppBaseURL = env.String("APP_BASE_URL", "", false) // Extra purgers: ClickHouse rows and per-tenant filesystem folders. diff --git a/backend/modules/mcp/tools_soar.go b/backend/modules/mcp/tools_soar.go index 5b53859de..398a6baf4 100644 --- a/backend/modules/mcp/tools_soar.go +++ b/backend/modules/mcp/tools_soar.go @@ -21,29 +21,25 @@ func registerSOAR(m *Module) { // ---- soar.rule.* ----------------------------------------------------------- type soarRuleCreateInput struct { - Name string `json:"name"` - Description string `json:"description,omitempty"` - Conditions []dto.FilterVM `json:"conditions"` - Commands []dto.FlowCommandVM `json:"commands"` - Active bool `json:"active"` - AgentPlatform string `json:"agent_platform"` - DefaultAgent string `json:"default_agent,omitempty"` - Shell string `json:"shell,omitempty"` - ExcludedAgents []string `json:"excluded_agents,omitempty"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + Conditions []dto.FilterVM `json:"conditions"` + Roots []string `json:"roots"` + Nodes map[string]dto.FlowNodeVM `json:"nodes"` + MaxDepth int `json:"max_depth,omitempty"` + Active bool `json:"active"` } type soarRuleUpdateInput struct { - RelPath string `json:"rel_path"` - ID *int64 `json:"id,omitempty"` - Name string `json:"name"` - Description string `json:"description,omitempty"` - Conditions []dto.FilterVM `json:"conditions"` - Commands []dto.FlowCommandVM `json:"commands"` - Active bool `json:"active"` - AgentPlatform string `json:"agent_platform"` - DefaultAgent string `json:"default_agent,omitempty"` - Shell string `json:"shell,omitempty"` - ExcludedAgents []string `json:"excluded_agents,omitempty"` + RelPath string `json:"rel_path"` + ID *int64 `json:"id,omitempty"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + Conditions []dto.FilterVM `json:"conditions"` + Roots []string `json:"roots"` + Nodes map[string]dto.FlowNodeVM `json:"nodes"` + MaxDepth int `json:"max_depth,omitempty"` + Active bool `json:"active"` } type soarRuleRelPathInput struct { @@ -56,13 +52,12 @@ type soarRuleSetEnabledInput struct { } type soarRuleListInput struct { - RuleName string `json:"name,omitempty"` - RuleActive *bool `json:"active,omitempty"` - AgentPlatform string `json:"agent_platform,omitempty"` - CreatedBy string `json:"created_by,omitempty"` - SystemOwner *bool `json:"system_owner,omitempty"` - Page int `json:"page,omitempty"` - Size int `json:"size,omitempty"` + RuleName string `json:"name,omitempty"` + RuleActive *bool `json:"active,omitempty"` + CreatedBy string `json:"created_by,omitempty"` + SystemOwner *bool `json:"system_owner,omitempty"` + Page int `json:"page,omitempty"` + Size int `json:"size,omitempty"` } func registerSOARRules(m *Module) { @@ -72,15 +67,14 @@ func registerSOARRules(m *Module) { Name: "soar.rule.create", Title: "Create SOAR rule", }, Gate{Permission: "soar.write"}, func(ctx context.Context, actor *authz.Actor, in soarRuleCreateInput) (any, error) { - if len(in.Conditions) == 0 || len(in.Commands) == 0 { - return nil, fmt.Errorf("conditions and commands are required") + if len(in.Conditions) == 0 || len(in.Roots) == 0 || len(in.Nodes) == 0 { + return nil, fmt.Errorf("conditions, roots, and nodes are required") } active := in.Active return uc.Create(ctx, dto.CreateRuleRequest{ Name: in.Name, Description: in.Description, - Conditions: in.Conditions, Commands: in.Commands, Active: &active, - AgentPlatform: in.AgentPlatform, DefaultAgent: in.DefaultAgent, - Shell: in.Shell, ExcludedAgents: in.ExcludedAgents, + Conditions: in.Conditions, Roots: in.Roots, Nodes: in.Nodes, + MaxDepth: in.MaxDepth, Active: &active, }, actor.Email) }) @@ -91,9 +85,8 @@ func registerSOARRules(m *Module) { active := in.Active return uc.Update(ctx, in.RelPath, dto.UpdateRuleRequest{ ID: in.ID, Name: in.Name, Description: in.Description, - Conditions: in.Conditions, Commands: in.Commands, Active: &active, - AgentPlatform: in.AgentPlatform, DefaultAgent: in.DefaultAgent, - Shell: in.Shell, ExcludedAgents: in.ExcludedAgents, + Conditions: in.Conditions, Roots: in.Roots, Nodes: in.Nodes, + MaxDepth: in.MaxDepth, Active: &active, }, actor.Email) }) @@ -133,7 +126,7 @@ func registerSOARRules(m *Module) { func(ctx context.Context, _ *authz.Actor, in soarRuleListInput) (any, error) { f := dto.RuleFilters{ RuleName: in.RuleName, RuleActive: in.RuleActive, - AgentPlatform: in.AgentPlatform, CreatedBy: in.CreatedBy, SystemOwner: in.SystemOwner, + CreatedBy: in.CreatedBy, SystemOwner: in.SystemOwner, Params: database.Params{Page: in.Page, Size: clampPageSize(in.Size)}, } return uc.List(ctx, f) diff --git a/backend/modules/soar/connectors/repository.go b/backend/modules/soar/connectors/repository.go index c843f8baa..ee6b72f3a 100644 --- a/backend/modules/soar/connectors/repository.go +++ b/backend/modules/soar/connectors/repository.go @@ -35,8 +35,62 @@ type ExecutionStatusUpdate struct { type ExecutionRepository interface { Create(ctx context.Context, e *domain.SoarExecution) (*domain.SoarExecution, error) List(ctx context.Context, f ExecutionFilters) ([]domain.SoarExecution, int64, error) + Get(ctx context.Context, id uuid.UUID) (*domain.SoarExecution, error) UpdateStatus(ctx context.Context, id uuid.UUID, u ExecutionStatusUpdate) error ClaimPending(ctx context.Context, id uuid.UUID, leaseDuration time.Duration) (bool, error) + + // SaveOutput persists an enrichment node's structured output; called after + // a successful Executor.Execute for kind=enrichment. + SaveOutput(ctx context.Context, id uuid.UUID, output []byte) error + + // RecordEdge inserts one parent→child edge and atomically decrements the + // child's pending_parents counter. If the child does not yet exist for + // (flowRunID, childNodeID, childDepth), it is created with the given + // template (status=WAITING, pending_parents=incomingCount, kind/executor + // copied from the flow node). Returns the child's post-update state. + RecordEdge(ctx context.Context, req RecordEdgeRequest) (child *domain.SoarExecution, err error) + + // ListFiredParents returns all fired parents of a child execution, used to + // build the merged context bag when the child transitions to PENDING. + ListFiredParents(ctx context.Context, childID uuid.UUID) ([]domain.SoarExecution, error) + + // TransitionReady moves a child from WAITING to PENDING (or DEAD) once all + // its parents have resolved. context/params/command/shell are the + // interpolated values computed by the caller from ListFiredParents. + TransitionReady(ctx context.Context, id uuid.UUID, ready ReadyUpdate) error +} + +type RecordEdgeRequest struct { + FlowRunID uuid.UUID + TenantID uuid.UUID + RulePath string + AlertID string + Parent domain.SoarExecution + ChildNodeID string + ChildDepth int + ChildKind domain.NodeKind + ChildExecutor string + IncomingCount int + Branch domain.EdgeBranch + Fired bool +} + +type ReadyUpdate struct { + Status domain.ExecutionStatus + Context []byte + Params []byte + Command string + Shell string + Agent string +} + +// FlowRunRepository holds the top-level state of one root-invocation. +type FlowRunRepository interface { + Create(ctx context.Context, r *domain.SoarFlowRun) (*domain.SoarFlowRun, error) + Get(ctx context.Context, id uuid.UUID) (*domain.SoarFlowRun, error) + // MaybeComplete transitions the run to COMPLETED/FAILED when no non-terminal + // executions remain. Returns true when a transition happened. + MaybeComplete(ctx context.Context, id uuid.UUID) (bool, error) } type ResolveFilterRepository interface { diff --git a/backend/modules/soar/domain/execution.go b/backend/modules/soar/domain/execution.go index afabe27d5..bfe47ceb4 100644 --- a/backend/modules/soar/domain/execution.go +++ b/backend/modules/soar/domain/execution.go @@ -1,6 +1,7 @@ package domain import ( + "encoding/json" "time" "github.com/google/uuid" @@ -9,16 +10,30 @@ import ( type ExecutionStatus string const ( - ExecutionStatusExecuted ExecutionStatus = "EXECUTED" - ExecutionStatusPending ExecutionStatus = "PENDING" - ExecutionStatusFailed ExecutionStatus = "FAILED" + ExecutionStatusWaiting ExecutionStatus = "WAITING" + ExecutionStatusPending ExecutionStatus = "PENDING" + ExecutionStatusExecuting ExecutionStatus = "EXECUTING" + ExecutionStatusExecuted ExecutionStatus = "EXECUTED" + ExecutionStatusFailed ExecutionStatus = "FAILED" + ExecutionStatusDead ExecutionStatus = "DEAD" ) +// Terminal returns true if the status represents a finished execution — the +// dispatcher will not touch it again. +func (s ExecutionStatus) Terminal() bool { + switch s { + case ExecutionStatusExecuted, ExecutionStatusFailed, ExecutionStatusDead: + return true + } + return false +} + type NonExecutionCause string const ( NonExecutionCauseAgentOffline NonExecutionCause = "AGENT_OFFLINE" NonExecutionCauseAgentNotFound NonExecutionCause = "AGENT_NOT_FOUND" + NonExecutionCauseMaxDepth NonExecutionCause = "MAX_DEPTH_EXCEEDED" NonExecutionCauseUnknown NonExecutionCause = "UNKNOWN" ) @@ -29,22 +44,85 @@ const ( ExecutionOriginManual ExecutionOrigin = "MANUAL" ) +type EdgeBranch string + +const ( + EdgeBranchSuccess EdgeBranch = "SUCCESS" + EdgeBranchError EdgeBranch = "ERROR" +) + +// SoarExecution is one node instance of a running flow. A single flow run may +// hold many rows; siblings that AND-join land on the same (flow_run_id, node_id, +// depth) tuple and coalesce. type SoarExecution struct { - ID uuid.UUID `gorm:"column:id;type:uuid;primaryKey;default:gen_random_uuid()" json:"id"` - TenantID uuid.UUID `gorm:"column:tenant_id;type:uuid;not null;index:idx_soar_execution_tenant_started,priority:1" json:"-"` - Origin ExecutionOrigin `gorm:"column:origin;size:20;not null" json:"origin"` - RulePath string `gorm:"column:rule_path;size:512;not null" json:"rulePath,omitempty"` // flow only — the YAML file, not an FK - AlertID string `gorm:"column:alert_id;size:150;not null" json:"alertId,omitempty"` // flow only - TriggeredBy string `gorm:"column:triggered_by;size:150;not null" json:"triggeredBy,omitempty"` // manual only - Agent string `gorm:"column:agent;size:150;not null" json:"agent"` - Command string `gorm:"column:command;not null" json:"command"` - Result string `gorm:"column:result" json:"result,omitempty"` - Status ExecutionStatus `gorm:"column:status;size:100;not null" json:"status"` + ID uuid.UUID `gorm:"column:id;type:uuid;primaryKey;default:gen_random_uuid()" json:"id"` + TenantID uuid.UUID `gorm:"column:tenant_id;type:uuid;not null;index:idx_soar_execution_tenant_started,priority:1" json:"-"` + Origin ExecutionOrigin `gorm:"column:origin;size:20;not null" json:"origin"` + + // Manual-execution fields (untouched by the DAG engine). + TriggeredBy string `gorm:"column:triggered_by;size:150;not null" json:"triggeredBy,omitempty"` + + // Flow-run linkage — nullable for manual executions. + FlowRunID *uuid.UUID `gorm:"column:flow_run_id;type:uuid;index" json:"flowRunId,omitempty"` + RulePath string `gorm:"column:rule_path;size:512;not null" json:"rulePath,omitempty"` + AlertID string `gorm:"column:alert_id;size:150;not null" json:"alertId,omitempty"` + + // DAG node identity. + NodeID string `gorm:"column:node_id;size:150;not null;index:idx_soar_execution_run_node,unique,priority:2" json:"nodeId,omitempty"` + Depth int `gorm:"column:depth;not null;default:0;index:idx_soar_execution_run_node,unique,priority:3" json:"depth"` + Kind NodeKind `gorm:"column:kind;size:20;not null" json:"kind"` + + // Executor + interpolated payload. + Executor string `gorm:"column:executor;size:60;not null" json:"executor"` + Params json.RawMessage `gorm:"column:params;type:jsonb" json:"params,omitempty"` + Output json.RawMessage `gorm:"column:output;type:jsonb" json:"output,omitempty"` + Context json.RawMessage `gorm:"column:context;type:jsonb" json:"context,omitempty"` + + // Shell-executor legacy fields (kept for manual execs and shell nodes). + Agent string `gorm:"column:agent;size:150;not null" json:"agent"` + Command string `gorm:"column:command;not null" json:"command"` + Result string `gorm:"column:result" json:"result,omitempty"` + Shell string `gorm:"column:shell;size:20" json:"shell,omitempty"` + + // AND-join accounting. + PendingParents int `gorm:"column:pending_parents;not null;default:0" json:"pendingParents"` + DeadParents int `gorm:"column:dead_parents;not null;default:0" json:"deadParents"` + + Status ExecutionStatus `gorm:"column:status;size:20;not null" json:"status"` StartedAt time.Time `gorm:"column:started_at;not null;index:idx_soar_execution_tenant_started,priority:2,sort:desc" json:"startedAt"` - FinishedAt *time.Time `gorm:"column:finished_at" json:"finishedAt,omitempty"` - ClaimedAt *time.Time `gorm:"column:claimed_at" json:"-"` - Retries int `gorm:"column:retries;not null;default:0" json:"retries"` - NonExecutionCause *NonExecutionCause `gorm:"column:non_execution_cause;size:100" json:"nonExecutionCause,omitempty"` + FinishedAt *time.Time `gorm:"column:finished_at" json:"finishedAt,omitempty"` + ClaimedAt *time.Time `gorm:"column:claimed_at" json:"-"` + Retries int `gorm:"column:retries;not null;default:0" json:"retries"` + NonExecutionCause *NonExecutionCause `gorm:"column:non_execution_cause;size:100" json:"nonExecutionCause,omitempty"` } func (SoarExecution) TableName() string { return "soar_executions" } + +// SoarFlowRun groups every node execution triggered by a single alert match. +type SoarFlowRun struct { + ID uuid.UUID `gorm:"column:id;type:uuid;primaryKey;default:gen_random_uuid()" json:"id"` + TenantID uuid.UUID `gorm:"column:tenant_id;type:uuid;not null;index" json:"-"` + RulePath string `gorm:"column:rule_path;size:512;not null" json:"rulePath"` + AlertID string `gorm:"column:alert_id;size:150;not null" json:"alertId"` + AlertJSON json.RawMessage `gorm:"column:alert_json;type:jsonb;not null" json:"-"` + MaxDepth int `gorm:"column:max_depth;not null;default:50" json:"maxDepth"` + Status ExecutionStatus `gorm:"column:status;size:20;not null" json:"status"` + StartedAt time.Time `gorm:"column:started_at;not null" json:"startedAt"` + FinishedAt *time.Time `gorm:"column:finished_at" json:"finishedAt,omitempty"` +} + +func (SoarFlowRun) TableName() string { return "soar_flow_runs" } + +// SoarExecutionEdge records a resolved incoming edge on a child execution. +// `Fired=false` means the parent branch didn't match — the edge died and the +// child is no longer reachable through this path. +type SoarExecutionEdge struct { + ChildExecID uuid.UUID `gorm:"column:child_exec_id;type:uuid;primaryKey" json:"childExecId"` + ParentExecID uuid.UUID `gorm:"column:parent_exec_id;type:uuid;primaryKey" json:"parentExecId"` + FlowRunID uuid.UUID `gorm:"column:flow_run_id;type:uuid;not null;index" json:"flowRunId"` + Branch EdgeBranch `gorm:"column:branch;size:16;not null" json:"branch"` + Fired bool `gorm:"column:fired;not null" json:"fired"` + CreatedAt time.Time `gorm:"column:created_at;not null" json:"createdAt"` +} + +func (SoarExecutionEdge) TableName() string { return "soar_execution_edges" } diff --git a/backend/modules/soar/domain/flow.go b/backend/modules/soar/domain/flow.go index dbb4cd29c..4c804b3c3 100644 --- a/backend/modules/soar/domain/flow.go +++ b/backend/modules/soar/domain/flow.go @@ -1,60 +1,184 @@ package domain import ( + "encoding/json" "time" "gopkg.in/yaml.v3" ) -type Flow struct { - Name string `yaml:"name" json:"name"` - Description string `yaml:"description,omitempty" json:"description,omitempty"` - Conditions []FilterType `yaml:"conditions" json:"conditions"` - Commands []FlowCommand `yaml:"commands" json:"commands"` - Shell string `yaml:"shell,omitempty" json:"shell,omitempty"` - AgentPlatform string `yaml:"agentPlatform,omitempty" json:"agentPlatform,omitempty"` - DefaultAgent string `yaml:"defaultAgent,omitempty" json:"defaultAgent,omitempty"` - ExcludedAgents []string `yaml:"excludedAgents,omitempty" json:"excludedAgents,omitempty"` -} - -type Condition string +type NodeKind string const ( - ConditionOnSuccess Condition = "OnSuccess" // && - ConditionOnFailure Condition = "OnFailure" // || - ConditionAlways Condition = "Always" // ; + NodeKindExecutor NodeKind = "executor" + NodeKindEnrichment NodeKind = "enrichment" ) -func (c Condition) Operator() string { - switch c { - case ConditionOnSuccess: - return "&&" - case ConditionOnFailure: - return "||" - default: - return ";" - } +const DefaultMaxDepth = 50 + +type Flow struct { + Name string `yaml:"name" json:"name"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` + Conditions []FilterType `yaml:"conditions" json:"conditions"` + MaxDepth int `yaml:"maxDepth,omitempty" json:"maxDepth,omitempty"` + Roots []string `yaml:"roots" json:"roots"` + Nodes map[string]FlowNode `yaml:"nodes" json:"nodes"` } -type FlowCommand struct { - Command string `yaml:"command" json:"command"` - Condition *Condition `yaml:"condition,omitempty" json:"condition,omitempty"` +type FlowNode struct { + Kind NodeKind `yaml:"kind" json:"kind"` + Executor string `yaml:"executor" json:"executor"` + Command string `yaml:"command,omitempty" json:"command,omitempty"` + Shell string `yaml:"shell,omitempty" json:"shell,omitempty"` + Platform string `yaml:"platform,omitempty" json:"platform,omitempty"` + Agent string `yaml:"agent,omitempty" json:"agent,omitempty"` + // ExcludedAgents lists hostnames that must not run this node, even when + // the resolved target (via Agent or the alert source) would otherwise pick + // them. Applied only when Agent is empty (auto-resolve mode). + ExcludedAgents []string `yaml:"excludedAgents,omitempty" json:"excludedAgents,omitempty"` + Params json.RawMessage `yaml:"-" json:"params,omitempty"` + OnSuccess []string `yaml:"onSuccess,omitempty" json:"onSuccess,omitempty"` + OnError []string `yaml:"onError,omitempty" json:"onError,omitempty"` } -func (fc *FlowCommand) UnmarshalYAML(value *yaml.Node) error { - if value.Kind == yaml.ScalarNode { - fc.Command = value.Value +// UnmarshalYAML lets nodes express Params as a native YAML mapping while the +// runtime still holds them as json.RawMessage (executors work on JSON, tests +// diff on JSON, and the DB column is jsonb). The YAML value is decoded into a +// generic any and re-encoded to JSON. +func (n *FlowNode) UnmarshalYAML(value *yaml.Node) error { + type shadow struct { + Kind NodeKind `yaml:"kind"` + Executor string `yaml:"executor"` + Command string `yaml:"command"` + Shell string `yaml:"shell"` + Platform string `yaml:"platform"` + Agent string `yaml:"agent"` + ExcludedAgents []string `yaml:"excludedAgents"` + Params yaml.Node `yaml:"params"` + OnSuccess []string `yaml:"onSuccess"` + OnError []string `yaml:"onError"` + } + var s shadow + if err := value.Decode(&s); err != nil { + return err + } + n.Kind = s.Kind + n.Executor = s.Executor + n.Command = s.Command + n.Shell = s.Shell + n.Platform = s.Platform + n.Agent = s.Agent + n.ExcludedAgents = s.ExcludedAgents + n.OnSuccess = s.OnSuccess + n.OnError = s.OnError + if s.Params.Kind == 0 { return nil } - type raw FlowCommand - var r raw - if err := value.Decode(&r); err != nil { + var raw any + if err := s.Params.Decode(&raw); err != nil { return err } - *fc = FlowCommand(r) + if raw == nil { + return nil + } + raw = normalizeYAMLValue(raw) + buf, err := json.Marshal(raw) + if err != nil { + return err + } + n.Params = json.RawMessage(buf) return nil } +// MarshalYAML emits Params as a plain mapping when it holds JSON, so writing a +// flow back to disk produces the same shape a human would type. +func (n FlowNode) MarshalYAML() (any, error) { + out := map[string]any{ + "kind": n.Kind, + "executor": n.Executor, + } + if n.Command != "" { + out["command"] = n.Command + } + if n.Shell != "" { + out["shell"] = n.Shell + } + if n.Platform != "" { + out["platform"] = n.Platform + } + if n.Agent != "" { + out["agent"] = n.Agent + } + if len(n.ExcludedAgents) > 0 { + out["excludedAgents"] = n.ExcludedAgents + } + if len(n.Params) > 0 { + var v any + if err := json.Unmarshal(n.Params, &v); err != nil { + return nil, err + } + out["params"] = v + } + if len(n.OnSuccess) > 0 { + out["onSuccess"] = n.OnSuccess + } + if len(n.OnError) > 0 { + out["onError"] = n.OnError + } + return out, nil +} + +// normalizeYAMLValue rewrites map[any]any (yaml.v3 default for untyped +// mappings) into map[string]any so encoding/json accepts it. +func normalizeYAMLValue(v any) any { + switch t := v.(type) { + case map[any]any: + out := make(map[string]any, len(t)) + for k, vv := range t { + ks, ok := k.(string) + if !ok { + continue + } + out[ks] = normalizeYAMLValue(vv) + } + return out + case map[string]any: + for k, vv := range t { + t[k] = normalizeYAMLValue(vv) + } + return t + case []any: + for i, vv := range t { + t[i] = normalizeYAMLValue(vv) + } + return t + } + return v +} + +// IncomingCounts returns, for every node id in the flow, how many other nodes +// reference it via on_success or on_error. AND-join sizing depends on this. +func (f Flow) IncomingCounts() map[string]int { + counts := make(map[string]int, len(f.Nodes)) + for _, node := range f.Nodes { + for _, id := range node.OnSuccess { + counts[id]++ + } + for _, id := range node.OnError { + counts[id]++ + } + } + return counts +} + +// ResolvedMaxDepth returns the flow's max_depth, falling back to DefaultMaxDepth. +func (f Flow) ResolvedMaxDepth() int { + if f.MaxDepth > 0 { + return f.MaxDepth + } + return DefaultMaxDepth +} + type StoredFlow struct { Flow RelPath string `yaml:"-"` diff --git a/backend/modules/soar/dto/execution.go b/backend/modules/soar/dto/execution.go index 04b0037c0..a6d7bd9be 100644 --- a/backend/modules/soar/dto/execution.go +++ b/backend/modules/soar/dto/execution.go @@ -24,6 +24,13 @@ type ExecutionResponse struct { FinishedAt *time.Time `json:"finishedAt,omitempty"` NonExecutionCause *domain.NonExecutionCause `json:"nonExecutionCause,omitempty"` Retries int `json:"retries"` + + // DAG node tracking — populated for flow executions, empty for manual. + NodeID string `json:"nodeId,omitempty"` + Kind domain.NodeKind `json:"kind,omitempty"` + Executor string `json:"executor,omitempty"` + FlowRunID *uuid.UUID `json:"flowRunId,omitempty"` + Depth int `json:"depth,omitempty"` } type MatchRequest struct { diff --git a/backend/modules/soar/dto/rule.go b/backend/modules/soar/dto/rule.go index 83ec7a5d9..7e11425df 100644 --- a/backend/modules/soar/dto/rule.go +++ b/backend/modules/soar/dto/rule.go @@ -1,6 +1,7 @@ package dto import ( + "encoding/json" "time" "github.com/utmstack/utmstack/backend/modules/soar/domain" @@ -13,35 +14,41 @@ type FilterVM struct { Value any `json:"value"` } -type FlowCommandVM struct { - Command string `json:"command" binding:"required"` - Condition *domain.Condition `json:"condition,omitempty" binding:"omitempty,oneof=OnSuccess OnFailure Always"` +// FlowNodeVM mirrors domain.FlowNode across the API. YAML on-disk uses the +// same field names — only the DAG shape lives here. +type FlowNodeVM struct { + Kind domain.NodeKind `json:"kind" binding:"required,oneof=executor enrichment"` + Executor string `json:"executor" binding:"required,max=60"` + Command string `json:"command,omitempty"` + Shell string `json:"shell,omitempty" binding:"omitempty,max=20"` + Platform string `json:"platform,omitempty" binding:"omitempty,max=60"` + Agent string `json:"agent,omitempty" binding:"omitempty,max=150"` + ExcludedAgents []string `json:"excludedAgents,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + OnSuccess []string `json:"onSuccess,omitempty"` + OnError []string `json:"onError,omitempty"` } type CreateRuleRequest struct { - ID *int64 `json:"id"` - Name string `json:"name" binding:"required,max=150"` - Description string `json:"description" binding:"omitempty,max=512"` - Conditions []FilterVM `json:"conditions" binding:"required,min=1"` - Commands []FlowCommandVM `json:"commands" binding:"required,min=1,dive"` - Active *bool `json:"active" binding:"required"` - AgentPlatform string `json:"agentPlatform" binding:"required"` - DefaultAgent string `json:"defaultAgent" binding:"omitempty,max=500"` - Shell string `json:"shell" binding:"omitempty,max=20"` - ExcludedAgents []string `json:"excludedAgents"` + ID *int64 `json:"id"` + Name string `json:"name" binding:"required,max=150"` + Description string `json:"description" binding:"omitempty,max=512"` + Conditions []FilterVM `json:"conditions" binding:"required,min=1"` + Roots []string `json:"roots" binding:"required,min=1"` + Nodes map[string]FlowNodeVM `json:"nodes" binding:"required,min=1"` + MaxDepth int `json:"maxDepth" binding:"omitempty,min=1,max=1000"` + Active *bool `json:"active" binding:"required"` } type UpdateRuleRequest struct { - ID *int64 `json:"id"` - Name string `json:"name" binding:"required,max=150"` - Description string `json:"description" binding:"omitempty,max=512"` - Conditions []FilterVM `json:"conditions" binding:"required,min=1"` - Commands []FlowCommandVM `json:"commands" binding:"required,min=1,dive"` - Active *bool `json:"active" binding:"required"` - AgentPlatform string `json:"agentPlatform" binding:"required"` - DefaultAgent string `json:"defaultAgent" binding:"omitempty,max=500"` - Shell string `json:"shell" binding:"omitempty,max=20"` - ExcludedAgents []string `json:"excludedAgents"` + ID *int64 `json:"id"` + Name string `json:"name" binding:"required,max=150"` + Description string `json:"description" binding:"omitempty,max=512"` + Conditions []FilterVM `json:"conditions" binding:"required,min=1"` + Roots []string `json:"roots" binding:"required,min=1"` + Nodes map[string]FlowNodeVM `json:"nodes" binding:"required,min=1"` + MaxDepth int `json:"maxDepth" binding:"omitempty,min=1,max=1000"` + Active *bool `json:"active" binding:"required"` } type ToggleRuleRequest struct { @@ -49,25 +56,22 @@ type ToggleRuleRequest struct { } type RuleResponse struct { - RelPath string `json:"relPath"` - Name string `json:"name"` - Description string `json:"description,omitempty"` - Conditions []FilterVM `json:"conditions"` - Commands []FlowCommandVM `json:"commands"` - Active bool `json:"active"` - AgentPlatform string `json:"agentPlatform,omitempty"` - DefaultAgent string `json:"defaultAgent,omitempty"` - Shell string `json:"shell,omitempty"` - ExcludedAgents []string `json:"excludedAgents,omitempty"` - SystemOwner bool `json:"systemOwner"` - LastModifiedDate *time.Time `json:"lastModifiedDate,omitempty"` + RelPath string `json:"relPath"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + Conditions []FilterVM `json:"conditions"` + Roots []string `json:"roots"` + Nodes map[string]FlowNodeVM `json:"nodes"` + MaxDepth int `json:"maxDepth,omitempty"` + Active bool `json:"active"` + SystemOwner bool `json:"systemOwner"` + LastModifiedDate *time.Time `json:"lastModifiedDate,omitempty"` } type RuleFilters struct { ID int64 `form:"id.equals"` RuleName string `form:"name.contains"` RuleActive *bool `form:"active.equals"` - AgentPlatform string `form:"agentPlatform.equals"` CreatedBy string `form:"createdBy.equals"` LastModifiedBy string `form:"lastModifiedBy.equals"` CreatedDateGTE string `form:"createdDate.greaterThanOrEqual"` diff --git a/backend/modules/soar/executor/executor.go b/backend/modules/soar/executor/executor.go new file mode 100644 index 000000000..b72530950 --- /dev/null +++ b/backend/modules/soar/executor/executor.go @@ -0,0 +1,40 @@ +package executor + +import ( + "context" + "encoding/json" + "errors" + + "github.com/utmstack/utmstack/backend/modules/soar/domain" +) + +// Executor turns a single node instance into a side effect (kind=executor) or a +// data lookup (kind=enrichment). It returns raw JSON `output` on success — the +// dispatcher stores it only when the node kind is enrichment; for executor +// kind it is discarded. +type Executor interface { + Type() string + Execute(ctx context.Context, exec *domain.SoarExecution) (output json.RawMessage, err error) +} + +// Registry is a plain map — executors are wired at module start and never +// mutated afterwards, so no locking is needed. +// ponytail: map, not sync.Map — reads only after Init. +type Registry map[string]Executor + +// ErrExecutorNotFound is returned by the dispatcher when a node references a +// type that was not registered. FlowStore should catch this at parse time, so +// hitting it at dispatch means a mis-wired module. +var ErrExecutorNotFound = errors.New("soar: executor not registered") + +// Lookup fetches an executor by type or returns ErrExecutorNotFound. +func (r Registry) Lookup(t string) (Executor, error) { + if e, ok := r[t]; ok { + return e, nil + } + return nil, ErrExecutorNotFound +} + +// Types returns the sorted set of registered executor names — used by the +// FlowStore validator to reject flows referencing unknown types. +func (r Registry) Has(t string) bool { _, ok := r[t]; return ok } diff --git a/backend/modules/soar/executor/http.go b/backend/modules/soar/executor/http.go new file mode 100644 index 000000000..a7830eccc --- /dev/null +++ b/backend/modules/soar/executor/http.go @@ -0,0 +1,112 @@ +package executor + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/utmstack/utmstack/backend/modules/soar/domain" +) + +// HTTP performs an outgoing HTTP call. As an executor kind, a non-2xx status +// is an error and the response body is ignored (logged into Result). As an +// enrichment kind, the response body must parse as JSON and is returned as the +// node's output. +type HTTP struct { + client *http.Client +} + +func NewHTTP() *HTTP { + return &HTTP{client: &http.Client{Timeout: 30 * time.Second}} +} + +func (HTTP) Type() string { return "http" } + +type httpParams struct { + Method string `json:"method,omitempty"` + URL string `json:"url"` + Headers map[string]string `json:"headers,omitempty"` + Body json.RawMessage `json:"body,omitempty"` + Timeout int `json:"timeoutSec,omitempty"` +} + +func (h *HTTP) Execute(ctx context.Context, exec *domain.SoarExecution) (json.RawMessage, error) { + var p httpParams + if len(exec.Params) > 0 { + if err := json.Unmarshal(exec.Params, &p); err != nil { + return nil, fmt.Errorf("soar http: params: %w", err) + } + } + if strings.TrimSpace(p.URL) == "" { + return nil, errors.New("soar http: url is required") + } + if p.Method == "" { + if len(p.Body) > 0 { + p.Method = http.MethodPost + } else { + p.Method = http.MethodGet + } + } + + client := h.client + if p.Timeout > 0 { + client = &http.Client{Timeout: time.Duration(p.Timeout) * time.Second} + } + + var body io.Reader + if len(p.Body) > 0 { + body = bytes.NewReader(p.Body) + } + req, err := http.NewRequestWithContext(ctx, strings.ToUpper(p.Method), p.URL, body) + if err != nil { + return nil, fmt.Errorf("soar http: build request: %w", err) + } + if len(p.Body) > 0 { + req.Header.Set("Content-Type", "application/json") + } + for k, v := range p.Headers { + req.Header.Set(k, v) + } + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("soar http: %s %s: %w", p.Method, p.URL, err) + } + defer resp.Body.Close() + + raw, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("soar http: read body: %w", err) + } + exec.Result = fmt.Sprintf("%d %s\n%s", resp.StatusCode, resp.Status, truncate(string(raw), 4096)) + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("soar http: %s %s returned %d", p.Method, p.URL, resp.StatusCode) + } + + if exec.Kind != domain.NodeKindEnrichment { + return nil, nil + } + trimmed := bytes.TrimSpace(raw) + if len(trimmed) == 0 { + return json.RawMessage(`{}`), nil + } + var probe any + if err := json.Unmarshal(trimmed, &probe); err != nil { + return nil, fmt.Errorf("soar http enrichment: body is not JSON: %w", err) + } + return json.RawMessage(trimmed), nil +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/backend/modules/soar/executor/llm.go b/backend/modules/soar/executor/llm.go new file mode 100644 index 000000000..751fe7e49 --- /dev/null +++ b/backend/modules/soar/executor/llm.go @@ -0,0 +1,239 @@ +package executor + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + + "github.com/utmstack/utmstack/backend/modules/soar/domain" +) + +// LLMStreamer is the narrow slice of the SOC-AI client that the LLM executors +// need. Defining it here keeps the executor package free of a socai import and +// lets tests inject a fake without spinning up an HTTP server. +type LLMStreamer interface { + StreamAgentTask(ctx context.Context, body []byte) (*http.Response, error) +} + +// LLM is one implementation backing two node types: +// - llm_enrich (kind=enrichment): drives the SOC-AI agent with a prompt and +// returns the final message parsed as JSON — becomes ancestor context for +// downstream nodes. +// - llm_action (kind=executor): drives the SOC-AI agent with a prompt so it +// can use its own tools (list hosts, run commands, page oncall, etc.) and +// succeeds when the stream ends on a `final` event. +type LLM struct { + client LLMStreamer + typ string +} + +// NewLLMEnrich registers a node type that expects a JSON `final` payload. +func NewLLMEnrich(c LLMStreamer) *LLM { return &LLM{client: c, typ: "llm_enrich"} } + +// NewLLMAction registers a node type that treats the `final` payload as free +// text and only cares whether the stream ended cleanly. +func NewLLMAction(c LLMStreamer) *LLM { return &LLM{client: c, typ: "llm_action"} } + +func (l *LLM) Type() string { return l.typ } + +type llmParams struct { + Prompt string `json:"prompt"` + Page string `json:"page,omitempty"` + Lang string `json:"lang,omitempty"` + History []llmChatTurn `json:"history,omitempty"` +} + +type llmChatTurn struct { + Role string `json:"role"` + Content string `json:"content"` +} + +func (l *LLM) Execute(ctx context.Context, exec *domain.SoarExecution) (json.RawMessage, error) { + if l.client == nil { + return nil, errors.New("soar llm: SOC-AI client not configured") + } + var p llmParams + if len(exec.Params) > 0 { + if err := json.Unmarshal(exec.Params, &p); err != nil { + return nil, fmt.Errorf("soar llm: params: %w", err) + } + } + if strings.TrimSpace(p.Prompt) == "" { + return nil, errors.New("soar llm: prompt is required") + } + + body, err := json.Marshal(map[string]any{ + "task": p.Prompt, + "page": defaultString(p.Page, "soar"), + "lang": defaultString(p.Lang, "en"), + "history": p.History, + }) + if err != nil { + return nil, fmt.Errorf("soar llm: build request: %w", err) + } + + resp, err := l.client.StreamAgentTask(ctx, body) + if err != nil { + return nil, fmt.Errorf("soar llm: stream: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("soar llm: upstream %d: %s", resp.StatusCode, string(raw)) + } + + events, finalRaw, errMsg, drainErr := drainSSE(resp.Body) + exec.Result = truncate(events, 8192) + if drainErr != nil { + return nil, fmt.Errorf("soar llm: read stream: %w", drainErr) + } + if errMsg != "" { + return nil, fmt.Errorf("soar llm: agent error: %s", errMsg) + } + if finalRaw == "" { + return nil, errors.New("soar llm: stream ended without a final event") + } + + if exec.Kind != domain.NodeKindEnrichment { + // llm_action: success means the agent reported completion. Nothing + // structured to hand downstream. + return nil, nil + } + output, err := extractJSONOutput(finalRaw) + if err != nil { + return nil, fmt.Errorf("soar llm enrichment: final is not JSON: %w", err) + } + return output, nil +} + +// drainSSE walks a text/event-stream body and returns the concatenated event +// log, the `data:` payload of the last `event: final`, and any `event: error` +// message. It stops on EOF or the first read error, mirroring the chat handler +// proxy — no reconnect logic. +func drainSSE(r io.Reader) (log string, finalData string, errMsg string, err error) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + scanner.Split(splitSSEFrames) + + var logBuf bytes.Buffer + for scanner.Scan() { + frame := scanner.Bytes() + if logBuf.Len() > 0 { + logBuf.WriteByte('\n') + } + logBuf.Write(frame) + + event, data := parseSSEFrame(frame) + switch event { + case "final": + finalData = data + case "error": + errMsg = data + } + } + if serr := scanner.Err(); serr != nil { + return logBuf.String(), finalData, errMsg, serr + } + return logBuf.String(), finalData, errMsg, nil +} + +// splitSSEFrames returns one SSE frame per Scanner call. Frames are separated +// by a blank line (`\n\n` or `\r\n\r\n`). +func splitSSEFrames(data []byte, atEOF bool) (advance int, token []byte, err error) { + if atEOF && len(data) == 0 { + return 0, nil, nil + } + if i := bytes.Index(data, []byte("\n\n")); i >= 0 { + return i + 2, dropCR(data[:i]), nil + } + if i := bytes.Index(data, []byte("\r\n\r\n")); i >= 0 { + return i + 4, dropCR(data[:i]), nil + } + if atEOF { + return len(data), dropCR(data), nil + } + return 0, nil, nil +} + +func dropCR(b []byte) []byte { return bytes.TrimRight(b, "\r") } + +func parseSSEFrame(frame []byte) (event string, data string) { + scanner := bufio.NewScanner(bytes.NewReader(frame)) + var dataBuf bytes.Buffer + for scanner.Scan() { + line := scanner.Text() + switch { + case strings.HasPrefix(line, "event:"): + event = strings.TrimSpace(strings.TrimPrefix(line, "event:")) + case strings.HasPrefix(line, "data:"): + if dataBuf.Len() > 0 { + dataBuf.WriteByte('\n') + } + dataBuf.WriteString(strings.TrimSpace(strings.TrimPrefix(line, "data:"))) + } + } + return event, dataBuf.String() +} + +// extractJSONOutput accepts a few final-message shapes the SOC-AI agent tends +// to produce: bare JSON, a `content` field inside a JSON envelope, or a JSON +// blob wrapped in a ```json fence. +func extractJSONOutput(finalData string) (json.RawMessage, error) { + trimmed := strings.TrimSpace(finalData) + if trimmed == "" { + return nil, errors.New("empty final message") + } + if raw, ok := tryJSON(trimmed); ok { + // Envelope { "content": "..." } — unwrap and retry. + var env struct { + Content string `json:"content"` + } + if err := json.Unmarshal(raw, &env); err == nil && strings.TrimSpace(env.Content) != "" { + if inner, ok := tryJSON(strings.TrimSpace(env.Content)); ok { + return inner, nil + } + if fenced, ok := stripJSONFence(env.Content); ok { + return fenced, nil + } + return nil, fmt.Errorf("content is not JSON: %s", truncate(env.Content, 200)) + } + return raw, nil + } + if fenced, ok := stripJSONFence(trimmed); ok { + return fenced, nil + } + return nil, fmt.Errorf("not JSON: %s", truncate(trimmed, 200)) +} + +func tryJSON(s string) (json.RawMessage, bool) { + var probe any + if err := json.Unmarshal([]byte(s), &probe); err != nil { + return nil, false + } + return json.RawMessage(s), true +} + +func stripJSONFence(s string) (json.RawMessage, bool) { + trimmed := strings.TrimSpace(s) + if !strings.HasPrefix(trimmed, "```") { + return nil, false + } + trimmed = strings.TrimPrefix(trimmed, "```json") + trimmed = strings.TrimPrefix(trimmed, "```") + trimmed = strings.TrimSuffix(trimmed, "```") + return tryJSON(strings.TrimSpace(trimmed)) +} + +func defaultString(s, fallback string) string { + if s == "" { + return fallback + } + return s +} diff --git a/backend/modules/soar/executor/notify.go b/backend/modules/soar/executor/notify.go new file mode 100644 index 000000000..c4dc94798 --- /dev/null +++ b/backend/modules/soar/executor/notify.go @@ -0,0 +1,58 @@ +package executor + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/utmstack/utmstack/backend/modules/notifications/domain" + soardomain "github.com/utmstack/utmstack/backend/modules/soar/domain" +) + +// Notifier is the narrow slice of the notifications usecase that the SOAR +// notify executor consumes. Keeps the soar/executor package free of a broader +// notifications import and lets tests swap in a fake. +type Notifier interface { + Notify(ctx context.Context, source domain.NotificationSource, ntype domain.NotificationType, message string) error +} + +// Notify posts an entry into the in-app notification stream — visible in the +// UI's bell menu. Meant for kind=executor: the branch outcome tracks whether +// the notification was accepted. +type Notify struct { + client Notifier +} + +func NewNotify(c Notifier) *Notify { return &Notify{client: c} } + +func (Notify) Type() string { return "notify" } + +type notifyParams struct { + Message string `json:"message"` + Type string `json:"type,omitempty"` // INFO (default) | WARNING +} + +func (n *Notify) Execute(ctx context.Context, exec *soardomain.SoarExecution) (json.RawMessage, error) { + if n.client == nil { + return nil, errors.New("soar notify: client not configured") + } + var p notifyParams + if len(exec.Params) > 0 { + if err := json.Unmarshal(exec.Params, &p); err != nil { + return nil, fmt.Errorf("soar notify: params: %w", err) + } + } + if p.Message == "" { + return nil, errors.New("soar notify: message is required") + } + ntype := domain.TypeInfo + if p.Type == string(domain.TypeWarning) { + ntype = domain.TypeWarning + } + if err := n.client.Notify(ctx, domain.SourceSystem, ntype, p.Message); err != nil { + return nil, err + } + exec.Result = fmt.Sprintf("notified (%s): %s", ntype, p.Message) + return nil, nil +} diff --git a/backend/modules/soar/executor/selectexec.go b/backend/modules/soar/executor/selectexec.go new file mode 100644 index 000000000..57e54accd --- /dev/null +++ b/backend/modules/soar/executor/selectexec.go @@ -0,0 +1,61 @@ +package executor + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/tidwall/gjson" + + "github.com/utmstack/utmstack/backend/modules/soar/domain" +) + +// Select is a lightweight enrichment: it composes an output object by pulling +// gjson paths out of the current context bag. Handy when downstream nodes want +// a subset (or a renamed subset) of ancestor data without dragging in a jq +// dependency. Meant for kind=enrichment. +// ponytail: gjson is already vendored (used in variable + execution +// interpolation); output built via encoding/json — no new dep. +type Select struct{} + +func NewSelect() *Select { return &Select{} } + +func (Select) Type() string { return "select" } + +type selectParams struct { + Fields map[string]string `json:"fields"` +} + +func (s *Select) Execute(_ context.Context, exec *domain.SoarExecution) (json.RawMessage, error) { + if exec.Kind != domain.NodeKindEnrichment { + return nil, errors.New("soar select: kind must be enrichment") + } + var p selectParams + if len(exec.Params) > 0 { + if err := json.Unmarshal(exec.Params, &p); err != nil { + return nil, fmt.Errorf("soar select: params: %w", err) + } + } + if len(p.Fields) == 0 { + return json.RawMessage(`{}`), nil + } + src := string(exec.Context) + if src == "" { + src = "{}" + } + out := make(map[string]json.RawMessage, len(p.Fields)) + for name, path := range p.Fields { + val := gjson.Get(src, path) + if !val.Exists() { + continue + } + out[name] = json.RawMessage(val.Raw) + } + raw, err := json.Marshal(out) + if err != nil { + return nil, fmt.Errorf("soar select: marshal: %w", err) + } + exec.Result = string(raw) + return raw, nil +} diff --git a/backend/modules/soar/executor/shell.go b/backend/modules/soar/executor/shell.go new file mode 100644 index 000000000..ee40b9bc1 --- /dev/null +++ b/backend/modules/soar/executor/shell.go @@ -0,0 +1,117 @@ +package executor + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/utmstack/utmstack/backend/modules/soar/domain" + "github.com/utmstack/utmstack/backend/pkg/agentmanager" + "github.com/utmstack/utmstack/backend/pkg/agentmanager/agent" +) + +const ( + shellOriginType = "INCIDENT_RESPONSE_AUTOMATION" + shellExecutedBy = "SYSTEM" +) + +// ErrAgentOffline signals that the target agent could not receive the command; +// the dispatcher decides whether to retry or fail based on this sentinel. +var ErrAgentOffline = errors.New("soar shell: agent offline") + +// ErrAgentNotFound signals that no agent matched the requested hostname. +var ErrAgentNotFound = errors.New("soar shell: agent not found") + +// Shell runs the node's command on a UTMStack endpoint agent via the +// agent-manager gRPC bidi stream. Output is the raw stdout — enrichment nodes +// get it as JSON when it parses. +type Shell struct { + agent *agentmanager.AgentManagerClient +} + +func NewShell(client *agentmanager.AgentManagerClient) *Shell { return &Shell{agent: client} } + +func (Shell) Type() string { return "shell" } + +func (s *Shell) Execute(ctx context.Context, exec *domain.SoarExecution) (json.RawMessage, error) { + if s.agent == nil { + return nil, errors.New("soar shell: agent-manager client not configured") + } + if exec.Command == "" { + return nil, errors.New("soar shell: empty command") + } + if exec.Agent == "" { + return nil, ErrAgentNotFound + } + + agentID, found, err := s.resolveAgent(ctx, exec.Agent) + if err != nil { + return nil, err + } + if !found { + return nil, ErrAgentNotFound + } + + cmd := &agent.UtmCommand{ + AgentId: agentID, + Command: exec.Command, + ExecutedBy: shellExecutedBy, + OriginType: shellOriginType, + OriginId: exec.RulePath, + Reason: fmt.Sprintf("Incident response automation: rule %s node %s alert %s", exec.RulePath, exec.NodeID, exec.AlertID), + Shell: exec.Shell, + } + + res, err := s.agent.ProcessCommand(ctx, cmd) + if err != nil { + if isOfflineError(err) { + return nil, ErrAgentOffline + } + return nil, err + } + + result := res.GetResult() + exec.Result = result + + // Enrichment shells get their output surfaced as JSON when the stdout + // parses. Non-JSON output is fine — dispatcher won't record it. + if exec.Kind == domain.NodeKindEnrichment { + trimmed := strings.TrimSpace(result) + if trimmed == "" { + return json.RawMessage(`{}`), nil + } + var probe any + if err := json.Unmarshal([]byte(trimmed), &probe); err != nil { + return nil, fmt.Errorf("soar shell enrichment: stdout is not JSON: %w", err) + } + return json.RawMessage(trimmed), nil + } + return nil, nil +} + +func (s *Shell) resolveAgent(ctx context.Context, hostname string) (string, bool, error) { + rows, _, err := s.agent.ListAgents(ctx, "hostname.Is="+hostname) + if err != nil { + return "", false, err + } + if len(rows) == 0 { + return "", false, nil + } + return strconv.FormatUint(uint64(rows[0].GetId()), 10), true, nil +} + +func isOfflineError(err error) bool { + if err == nil { + return false + } + if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound { + return true + } + return strings.Contains(err.Error(), "not found or is disconnected") +} diff --git a/backend/modules/soar/module.go b/backend/modules/soar/module.go index 75461d3fd..ea5f62140 100644 --- a/backend/modules/soar/module.go +++ b/backend/modules/soar/module.go @@ -8,6 +8,7 @@ import ( "gorm.io/gorm" "github.com/utmstack/utmstack/backend/modules/soar/connectors" + "github.com/utmstack/utmstack/backend/modules/soar/executor" "github.com/utmstack/utmstack/backend/modules/soar/handler" "github.com/utmstack/utmstack/backend/modules/soar/repository" "github.com/utmstack/utmstack/backend/modules/soar/usecase" @@ -39,6 +40,8 @@ func NewModule( agentClient *agentmanager.AgentManagerClient, signer *jwtpkg.Signer, cipher *secret.Cipher, + llm executor.LLMStreamer, + notifier executor.Notifier, tenantLister func(context.Context) ([]string, error), ) *Module { flowsSrc := env.String("SOAR_FLOWS_SRC_DIR", "/utmstack/soar", false) @@ -50,17 +53,31 @@ func NewModule( resolveRepo := repository.NewResolveFilterRepository(db) executionRepo := repository.NewExecutionRepository(db) + flowRunRepo := repository.NewFlowRunRepository(db) variableRepo := repository.NewVariableRepository(db) variableUC := usecase.NewVariableUsecase(variableRepo, cipher) - dispatcher := usecase.NewDispatcher(executionRepo, flowStore, agentClient, variableUC) + registry := executor.Registry{ + "shell": executor.NewShell(agentClient), + "http": executor.NewHTTP(), + "select": executor.NewSelect(), + } + if llm != nil { + registry["llm_enrich"] = executor.NewLLMEnrich(llm) + registry["llm_action"] = executor.NewLLMAction(llm) + } + if notifier != nil { + registry["notify"] = executor.NewNotify(notifier) + } + + dispatcher := usecase.NewDispatcher(executionRepo, flowRunRepo, flowStore, variableUC, registry) agentRepo := repository.NewAgentRepository(db) agentUC := usecase.NewAgentUsecase(agentRepo) ruleUC := usecase.NewRuleUsecase(flowStore, resolveRepo) - executionUC := usecase.NewExecutionUsecase(executionRepo, flowStore, agentUC, dispatcher.Kick) + executionUC := usecase.NewExecutionUsecase(executionRepo, flowRunRepo, flowStore, agentUC, variableUC, dispatcher.Kick) return &Module{ ruleHandler: handler.NewRuleHandler(ruleUC), diff --git a/backend/modules/soar/repository/execution_pg.go b/backend/modules/soar/repository/execution_pg.go index 53e26b4db..c24ef0957 100644 --- a/backend/modules/soar/repository/execution_pg.go +++ b/backend/modules/soar/repository/execution_pg.go @@ -2,10 +2,12 @@ package repository import ( "context" + "errors" "time" "github.com/google/uuid" "gorm.io/gorm" + "gorm.io/gorm/clause" "github.com/utmstack/utmstack/backend/modules/soar/connectors" "github.com/utmstack/utmstack/backend/modules/soar/domain" @@ -29,6 +31,17 @@ func (r *pgExecutionRepository) Create(ctx context.Context, e *domain.SoarExecut return e, nil } +func (r *pgExecutionRepository) Get(ctx context.Context, id uuid.UUID) (*domain.SoarExecution, error) { + var e domain.SoarExecution + if err := r.db.WithContext(ctx).Where("id = ?", id).First(&e).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, domain.ErrIncidentRecordNotFound + } + return nil, err + } + return &e, nil +} + func (r *pgExecutionRepository) List(ctx context.Context, f connectors.ExecutionFilters) ([]domain.SoarExecution, int64, error) { q := r.db.WithContext(ctx).Model(&domain.SoarExecution{}) @@ -113,11 +126,143 @@ func (r *pgExecutionRepository) ClaimPending(ctx context.Context, id uuid.UUID, staleBefore := time.Now().UTC().Add(-leaseDuration) res := r.db.WithContext(ctx). Model(&domain.SoarExecution{}). - Where("id = ? AND origin = ? AND status = ? AND (claimed_at IS NULL OR claimed_at < ?)", - id, domain.ExecutionOriginFlow, domain.ExecutionStatusPending, staleBefore). + Where("id = ? AND status = ? AND (claimed_at IS NULL OR claimed_at < ?)", + id, domain.ExecutionStatusPending, staleBefore). Update("claimed_at", time.Now().UTC()) if res.Error != nil { return false, res.Error } return res.RowsAffected > 0, nil } + +func (r *pgExecutionRepository) SaveOutput(ctx context.Context, id uuid.UUID, output []byte) error { + return r.db.WithContext(ctx). + Model(&domain.SoarExecution{}). + Where("id = ?", id). + Update("output", output).Error +} + +// RecordEdge is the atomic hinge of the DAG engine. Everything happens inside +// one transaction so concurrent parents can't race the pending_parents counter. +func (r *pgExecutionRepository) RecordEdge(ctx context.Context, req connectors.RecordEdgeRequest) (*domain.SoarExecution, error) { + var child *domain.SoarExecution + err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + // 1) find-or-create child, keyed by (flow_run_id, node_id, depth). + var existing domain.SoarExecution + findErr := tx.Where("flow_run_id = ? AND node_id = ? AND depth = ?", + req.FlowRunID, req.ChildNodeID, req.ChildDepth). + First(&existing).Error + + var created domain.SoarExecution + switch { + case errors.Is(findErr, gorm.ErrRecordNotFound): + created = domain.SoarExecution{ + TenantID: req.TenantID, + Origin: domain.ExecutionOriginFlow, + FlowRunID: &req.FlowRunID, + RulePath: req.RulePath, + AlertID: req.AlertID, + NodeID: req.ChildNodeID, + Depth: req.ChildDepth, + Kind: req.ChildKind, + Executor: req.ChildExecutor, + PendingParents: req.IncomingCount, + Status: domain.ExecutionStatusWaiting, + StartedAt: time.Now().UTC(), + } + // ON CONFLICT protects against two parents racing the initial insert. + if err := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "flow_run_id"}, {Name: "node_id"}, {Name: "depth"}}, + DoNothing: true, + }).Create(&created).Error; err != nil { + return err + } + // Re-read to pick up the winning row after ON CONFLICT. + if err := tx.Where("flow_run_id = ? AND node_id = ? AND depth = ?", + req.FlowRunID, req.ChildNodeID, req.ChildDepth). + First(&existing).Error; err != nil { + return err + } + case findErr != nil: + return findErr + } + + // 2) record the edge (may already exist if the parent retries; ignore + // duplicates by primary key). + edge := domain.SoarExecutionEdge{ + ChildExecID: existing.ID, + ParentExecID: req.Parent.ID, + FlowRunID: req.FlowRunID, + Branch: req.Branch, + Fired: req.Fired, + CreatedAt: time.Now().UTC(), + } + if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&edge).Error; err != nil { + return err + } + + // 3) lock the child row and decrement counters. Only decrement when the + // edge was fresh (RowsAffected on the edge insert), otherwise this call + // is a retry. + var locked domain.SoarExecution + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("id = ?", existing.ID).First(&locked).Error; err != nil { + return err + } + if !locked.Status.Terminal() && locked.Status == domain.ExecutionStatusWaiting { + updates := map[string]any{ + "pending_parents": gorm.Expr("GREATEST(pending_parents - 1, 0)"), + } + if !req.Fired { + updates["dead_parents"] = gorm.Expr("dead_parents + 1") + } + if err := tx.Model(&domain.SoarExecution{}). + Where("id = ?", locked.ID).Updates(updates).Error; err != nil { + return err + } + } + if err := tx.Where("id = ?", existing.ID).First(&locked).Error; err != nil { + return err + } + child = &locked + return nil + }) + return child, err +} + +func (r *pgExecutionRepository) ListFiredParents(ctx context.Context, childID uuid.UUID) ([]domain.SoarExecution, error) { + var parents []domain.SoarExecution + err := r.db.WithContext(ctx). + Raw(`SELECT p.* FROM soar_executions p + JOIN soar_execution_edges e ON e.parent_exec_id = p.id + WHERE e.child_exec_id = ? AND e.fired = TRUE`, childID). + Scan(&parents).Error + return parents, err +} + +func (r *pgExecutionRepository) TransitionReady(ctx context.Context, id uuid.UUID, ready connectors.ReadyUpdate) error { + updates := map[string]any{"status": ready.Status} + if len(ready.Context) > 0 { + updates["context"] = ready.Context + } + if len(ready.Params) > 0 { + updates["params"] = ready.Params + } + if ready.Command != "" { + updates["command"] = ready.Command + } + if ready.Shell != "" { + updates["shell"] = ready.Shell + } + if ready.Agent != "" { + updates["agent"] = ready.Agent + } + res := r.db.WithContext(ctx). + Model(&domain.SoarExecution{}). + Where("id = ? AND status = ?", id, domain.ExecutionStatusWaiting). + Updates(updates) + if res.Error != nil { + return res.Error + } + return nil +} diff --git a/backend/modules/soar/repository/flow_run_pg.go b/backend/modules/soar/repository/flow_run_pg.go new file mode 100644 index 000000000..37f57cd54 --- /dev/null +++ b/backend/modules/soar/repository/flow_run_pg.go @@ -0,0 +1,86 @@ +package repository + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" + "gorm.io/gorm" + + "github.com/utmstack/utmstack/backend/modules/soar/connectors" + "github.com/utmstack/utmstack/backend/modules/soar/domain" +) + +type pgFlowRunRepository struct { + db *gorm.DB +} + +func NewFlowRunRepository(db *gorm.DB) connectors.FlowRunRepository { + return &pgFlowRunRepository{db: db} +} + +func (r *pgFlowRunRepository) Create(ctx context.Context, run *domain.SoarFlowRun) (*domain.SoarFlowRun, error) { + if run.StartedAt.IsZero() { + run.StartedAt = time.Now().UTC() + } + if run.Status == "" { + run.Status = domain.ExecutionStatusPending + } + if err := r.db.WithContext(ctx).Create(run).Error; err != nil { + return nil, err + } + return run, nil +} + +func (r *pgFlowRunRepository) Get(ctx context.Context, id uuid.UUID) (*domain.SoarFlowRun, error) { + var run domain.SoarFlowRun + if err := r.db.WithContext(ctx).Where("id = ?", id).First(&run).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, domain.ErrIncidentRecordNotFound + } + return nil, err + } + return &run, nil +} + +// MaybeComplete inspects every execution in the run and transitions the run to +// EXECUTED (all children in a terminal-good state) or FAILED (any node ended +// in FAILED/DEAD with no compensating success). A run with any non-terminal +// execution is left alone. +func (r *pgFlowRunRepository) MaybeComplete(ctx context.Context, id uuid.UUID) (bool, error) { + var counts struct { + NonTerminal int64 + Failed int64 + Executed int64 + } + if err := r.db.WithContext(ctx). + Raw(`SELECT + SUM(CASE WHEN status IN (?, ?, ?) THEN 1 ELSE 0 END) AS non_terminal, + SUM(CASE WHEN status IN (?, ?) THEN 1 ELSE 0 END) AS failed, + SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS executed + FROM soar_executions WHERE flow_run_id = ?`, + domain.ExecutionStatusWaiting, domain.ExecutionStatusPending, domain.ExecutionStatusExecuting, + domain.ExecutionStatusFailed, domain.ExecutionStatusDead, + domain.ExecutionStatusExecuted, + id). + Scan(&counts).Error; err != nil { + return false, err + } + if counts.NonTerminal > 0 { + return false, nil + } + next := domain.ExecutionStatusExecuted + if counts.Executed == 0 && counts.Failed > 0 { + next = domain.ExecutionStatusFailed + } + now := time.Now().UTC() + res := r.db.WithContext(ctx). + Model(&domain.SoarFlowRun{}). + Where("id = ? AND finished_at IS NULL", id). + Updates(map[string]any{"status": next, "finished_at": now}) + if res.Error != nil { + return false, res.Error + } + return res.RowsAffected > 0, nil +} diff --git a/backend/modules/soar/usecase/assemble_chain_test.go b/backend/modules/soar/usecase/assemble_chain_test.go deleted file mode 100644 index ee51f2b20..000000000 --- a/backend/modules/soar/usecase/assemble_chain_test.go +++ /dev/null @@ -1,56 +0,0 @@ -package usecase - -import ( - "testing" - - "gopkg.in/yaml.v3" - - "github.com/utmstack/utmstack/backend/modules/soar/domain" -) - -func TestFlowCommandUnmarshalYAML_LegacyString(t *testing.T) { - src := []byte("commands:\n - net user \"$(x)\" /active:no\n - {command: \"echo b\", condition: OnSuccess}\n") - var wrap struct { - Commands []domain.FlowCommand `yaml:"commands"` - } - if err := yaml.Unmarshal(src, &wrap); err != nil { - t.Fatalf("unmarshal: %v", err) - } - if len(wrap.Commands) != 2 || wrap.Commands[0].Command != `net user "$(x)" /active:no` || wrap.Commands[0].Condition != nil { - t.Fatalf("bare-string entry not decoded: %+v", wrap.Commands) - } - if wrap.Commands[1].Command != "echo b" || wrap.Commands[1].Condition == nil || *wrap.Commands[1].Condition != domain.ConditionOnSuccess { - t.Fatalf("mapping entry not decoded: %+v", wrap.Commands[1]) - } -} - -func TestAssembleChain(t *testing.T) { - ok := domain.ConditionOnSuccess - fail := domain.ConditionOnFailure - always := domain.ConditionAlways - - cases := []struct { - name string - in []domain.FlowCommand - want string - }{ - {"empty", nil, ""}, - {"single command drops leading condition", - []domain.FlowCommand{{Command: "a", Condition: &ok}}, - "a"}, - {"chain uses each entry's condition as the joiner from the previous", - []domain.FlowCommand{{Command: "a"}, {Command: "b", Condition: &ok}, {Command: "c", Condition: &fail}, {Command: "d", Condition: &always}}, - "a && b || c ; d"}, - {"nil condition on non-first defaults to ;", - []domain.FlowCommand{{Command: "a"}, {Command: "b"}}, - "a ; b"}, - {"empty commands are skipped without leaving stray operators", - []domain.FlowCommand{{Command: "a"}, {Command: "", Condition: &ok}, {Command: "c", Condition: &fail}}, - "a || c"}, - } - for _, tc := range cases { - if got := assembleChain(tc.in); got != tc.want { - t.Errorf("%s: got %q want %q", tc.name, got, tc.want) - } - } -} diff --git a/backend/modules/soar/usecase/dispatch.go b/backend/modules/soar/usecase/dispatch.go index b81f015db..104760a57 100644 --- a/backend/modules/soar/usecase/dispatch.go +++ b/backend/modules/soar/usecase/dispatch.go @@ -2,9 +2,8 @@ package usecase import ( "context" - "fmt" - "strconv" - "strings" + "encoding/json" + "errors" "sync" "time" @@ -13,26 +12,34 @@ import ( "github.com/google/uuid" "github.com/threatwinds/go-sdk/catcher" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" "github.com/utmstack/utmstack/backend/modules/soar/connectors" "github.com/utmstack/utmstack/backend/modules/soar/domain" - "github.com/utmstack/utmstack/backend/pkg/agentmanager" - "github.com/utmstack/utmstack/backend/pkg/agentmanager/agent" + "github.com/utmstack/utmstack/backend/modules/soar/executor" "github.com/utmstack/utmstack/backend/pkg/database" ) +// Dispatcher walks the DAG: it pulls PENDING executions, hands each one to its +// registered Executor, then spawns downstream children (on_success or on_error +// edges) using the AND-join semantics described in +// /home/nadie/.claude/plans/more-a-dag-than-sunny-umbrella.md. type Dispatcher struct { - repo connectors.ExecutionRepository + exec connectors.ExecutionRepository + runs connectors.FlowRunRepository flows *FlowStore - agent *agentmanager.AgentManagerClient vars connectors.VariableUsecase + reg executor.Registry kick chan struct{} } -func NewDispatcher(repo connectors.ExecutionRepository, flows *FlowStore, agent *agentmanager.AgentManagerClient, vars connectors.VariableUsecase) *Dispatcher { - return &Dispatcher{repo: repo, flows: flows, agent: agent, vars: vars, kick: make(chan struct{}, 1)} +func NewDispatcher( + exec connectors.ExecutionRepository, + runs connectors.FlowRunRepository, + flows *FlowStore, + vars connectors.VariableUsecase, + reg executor.Registry, +) *Dispatcher { + return &Dispatcher{exec: exec, runs: runs, flows: flows, vars: vars, reg: reg, kick: make(chan struct{}, 1)} } const ( @@ -41,8 +48,6 @@ const ( dispatchConcurrency = 5 dispatchTimeout = 60 * time.Second dispatchMaxRetries = 3 - dispatchOriginType = "INCIDENT_RESPONSE_AUTOMATION" - dispatchExecutedBy = "SYSTEM" ) func (d *Dispatcher) Kick() { @@ -53,8 +58,8 @@ func (d *Dispatcher) Kick() { } func (d *Dispatcher) Start(ctx context.Context) { - if d.agent == nil { - _ = catcher.Error("soar dispatcher disabled: no agent-manager client", nil, nil) + if len(d.reg) == 0 { + _ = catcher.Error("soar dispatcher disabled: no executors registered", nil, nil) return } t := time.NewTicker(dispatchTick) @@ -80,7 +85,7 @@ func (d *Dispatcher) drain(ctx context.Context) { }() listCtx, cancel := context.WithTimeout(tenancy.WithAllTenants(ctx), 20*time.Second) - pending, _, err := d.repo.List(listCtx, connectors.ExecutionFilters{ + pending, _, err := d.exec.List(listCtx, connectors.ExecutionFilters{ Status: domain.ExecutionStatusPending, Params: database.Params{Size: dispatchBatch}, }) @@ -113,6 +118,9 @@ func (d *Dispatcher) drain(ctx context.Context) { wg.Wait() } +// process is the per-execution pipeline: claim → lookup executor → run → +// persist result/output → spawn children (both fired and dead branches) → +// maybe complete the flow run. func (d *Dispatcher) process(parent context.Context, exec domain.SoarExecution) { defer func() { if r := recover(); r != nil { @@ -123,7 +131,7 @@ func (d *Dispatcher) process(parent context.Context, exec domain.SoarExecution) ctx, cancel := context.WithTimeout(authz.WithTenantID(parent, exec.TenantID.String()), dispatchTimeout) defer cancel() - claimed, err := d.repo.ClaimPending(ctx, exec.ID, dispatchTimeout) + claimed, err := d.exec.ClaimPending(ctx, exec.ID, dispatchTimeout) if err != nil { _ = catcher.Error("soar dispatch: claim failed", err, map[string]any{"execution": exec.ID}) return @@ -132,82 +140,249 @@ func (d *Dispatcher) process(parent context.Context, exec domain.SoarExecution) return } - flow := d.flows.Get(exec.TenantID.String(), exec.RulePath) + // Manual executions have no flow-run; they still route through the shell + // executor but skip DAG spawning. + if exec.Origin == domain.ExecutionOriginManual { + d.runOnce(ctx, &exec) + return + } + + flow := d.loadFlow(exec) if flow == nil { d.fail(ctx, exec.ID, domain.NonExecutionCauseUnknown) return } - agentID, found, err := d.resolveAgent(ctx, exec.Agent) - if err != nil { - // Transient lookup failure — leave PENDING for the next tick. - _ = catcher.Error("soar dispatch: agent lookup failed", err, map[string]any{"execution": exec.ID, "agent": exec.Agent}) + // Transition to EXECUTING for visibility while the executor runs. + executing := domain.ExecutionStatusExecuting + _ = d.exec.UpdateStatus(ctx, exec.ID, connectors.ExecutionStatusUpdate{Status: &executing}) + + branch, output := d.invoke(ctx, &exec) + d.settle(ctx, &exec, branch, output) + + node, ok := flow.Nodes[exec.NodeID] + if !ok { + _ = catcher.Error("soar dispatch: node vanished mid-run", nil, map[string]any{"execution": exec.ID, "node": exec.NodeID}) return } - if !found { - d.fail(ctx, exec.ID, domain.NonExecutionCauseAgentNotFound) - return + fired, dead := d.edgesForBranch(node, branch) + d.spawnChildren(ctx, flow, exec, fired, branch, true) + d.spawnChildren(ctx, flow, exec, dead, oppositeBranch(branch), false) + + if exec.FlowRunID != nil { + if _, err := d.runs.MaybeComplete(ctx, *exec.FlowRunID); err != nil { + _ = catcher.Error("soar dispatch: maybeComplete failed", err, map[string]any{"flowRun": *exec.FlowRunID}) + } } +} - command := exec.Command - if d.vars != nil { - interpolated, ierr := d.vars.InterpolateCommand(ctx, exec.Command) - if ierr != nil { - _ = catcher.Error("soar dispatch: variable interpolation failed", ierr, map[string]any{"execution": exec.ID}) +// runOnce handles the legacy manual path: run the shell executor once, no DAG. +func (d *Dispatcher) runOnce(ctx context.Context, exec *domain.SoarExecution) { + branch, output := d.invoke(ctx, exec) + d.settle(ctx, exec, branch, output) +} + +func (d *Dispatcher) invoke(ctx context.Context, exec *domain.SoarExecution) (domain.EdgeBranch, json.RawMessage) { + e, err := d.reg.Lookup(exec.Executor) + if err != nil { + exec.Result = err.Error() + return domain.EdgeBranchError, nil + } + output, err := e.Execute(ctx, exec) + if err != nil { + if errors.Is(err, executor.ErrAgentOffline) { + d.handleOffline(ctx, *exec) + return domain.EdgeBranchError, nil } - command = interpolated + if exec.Result == "" { + exec.Result = err.Error() + } + return domain.EdgeBranchError, nil } + return domain.EdgeBranchSuccess, output +} - cmd := &agent.UtmCommand{ - AgentId: agentID, - Command: command, - ExecutedBy: dispatchExecutedBy, - OriginType: dispatchOriginType, - OriginId: exec.RulePath, - Reason: fmt.Sprintf("Incident response automation: rule %s matched alert %s", exec.RulePath, exec.AlertID), - Shell: flow.Shell, +func (d *Dispatcher) settle(ctx context.Context, exec *domain.SoarExecution, branch domain.EdgeBranch, output json.RawMessage) { + result := exec.Result + if d.vars != nil && result != "" { + masked, merr := d.vars.MaskSecrets(ctx, result) + if merr == nil { + result = masked + } } + now := time.Now().UTC() + status := domain.ExecutionStatusExecuted + if branch == domain.EdgeBranchError { + status = domain.ExecutionStatusFailed + } + _ = d.exec.UpdateStatus(ctx, exec.ID, connectors.ExecutionStatusUpdate{ + Status: &status, + Result: &result, + FinishedAt: &now, + }) + exec.Status = status + if branch == domain.EdgeBranchSuccess && exec.Kind == domain.NodeKindEnrichment && len(output) > 0 { + if err := d.exec.SaveOutput(ctx, exec.ID, output); err != nil { + _ = catcher.Error("soar dispatch: failed to save enrichment output", err, map[string]any{"execution": exec.ID}) + } + exec.Output = output + } +} - res, err := d.agent.ProcessCommand(ctx, cmd) - if err != nil { - if isOfflineError(err) { - d.handleOffline(ctx, exec) - return +func (d *Dispatcher) edgesForBranch(node domain.FlowNode, branch domain.EdgeBranch) (fired []string, dead []string) { + if branch == domain.EdgeBranchSuccess { + return node.OnSuccess, node.OnError + } + return node.OnError, node.OnSuccess +} + +func oppositeBranch(b domain.EdgeBranch) domain.EdgeBranch { + if b == domain.EdgeBranchSuccess { + return domain.EdgeBranchError + } + return domain.EdgeBranchSuccess +} + +// spawnChildren records edges and, when a child's incoming edges are fully +// resolved, transitions it to PENDING (or DEAD when any parent came in on the +// wrong branch). +func (d *Dispatcher) spawnChildren(ctx context.Context, flow *domain.Flow, parent domain.SoarExecution, childIDs []string, branch domain.EdgeBranch, fired bool) { + if len(childIDs) == 0 || parent.FlowRunID == nil { + return + } + incoming := flow.IncomingCounts() + maxDepth := flow.ResolvedMaxDepth() + for _, id := range childIDs { + node, ok := flow.Nodes[id] + if !ok { + _ = catcher.Error("soar dispatch: edge points to missing node", nil, map[string]any{"parent": parent.ID, "child": id}) + continue } - _ = catcher.Error("soar dispatch: command failed", err, map[string]any{"execution": exec.ID}) - d.fail(ctx, exec.ID, domain.NonExecutionCauseUnknown) + childDepth := parent.Depth + 1 + if childDepth > maxDepth { + _ = catcher.Error("soar dispatch: max_depth exceeded", nil, map[string]any{"parent": parent.ID, "child": id, "depth": childDepth}) + continue + } + child, err := d.exec.RecordEdge(ctx, connectors.RecordEdgeRequest{ + FlowRunID: *parent.FlowRunID, + TenantID: parent.TenantID, + RulePath: parent.RulePath, + AlertID: parent.AlertID, + Parent: parent, + ChildNodeID: id, + ChildDepth: childDepth, + ChildKind: node.Kind, + ChildExecutor: node.Executor, + IncomingCount: incoming[id], + Branch: branch, + Fired: fired, + }) + if err != nil { + _ = catcher.Error("soar dispatch: record edge failed", err, map[string]any{"parent": parent.ID, "child": id}) + continue + } + if child.PendingParents > 0 { + continue + } + if child.Status.Terminal() { + continue // already settled by another parent racing us + } + d.transitionChild(ctx, flow, node, child) + } +} + +// transitionChild fires when a child's last parent resolves. Dead-if-any-dead +// wins: an AND-join that had any incoming edge on the wrong branch cannot +// execute and its subtree is marked dead. +func (d *Dispatcher) transitionChild(ctx context.Context, flow *domain.Flow, node domain.FlowNode, child *domain.SoarExecution) { + if child.DeadParents > 0 { + dead := domain.ExecutionStatusDead + now := time.Now().UTC() + _ = d.exec.UpdateStatus(ctx, child.ID, connectors.ExecutionStatusUpdate{ + Status: &dead, + FinishedAt: &now, + }) + child.Status = dead + // Propagate death down both branches — every downstream node is dead. + d.spawnChildren(ctx, flow, *child, node.OnSuccess, domain.EdgeBranchSuccess, false) + d.spawnChildren(ctx, flow, *child, node.OnError, domain.EdgeBranchError, false) return } - executed := domain.ExecutionStatusExecuted - result := res.GetResult() + parents, err := d.exec.ListFiredParents(ctx, child.ID) + if err != nil { + _ = catcher.Error("soar dispatch: list parents failed", err, map[string]any{"execution": child.ID}) + return + } + contribs := make([]ParentContribution, 0, len(parents)) + for _, p := range parents { + c := ParentContribution{Context: p.Context} + if p.Kind == domain.NodeKindEnrichment && len(p.Output) > 0 { + c.EnrichmentNodeID = p.NodeID + c.Output = p.Output + } + contribs = append(contribs, c) + } + bag := MergeContexts(contribs) - if d.vars != nil { - masked, merr := d.vars.MaskSecrets(ctx, result) - if merr != nil { - _ = catcher.Error("soar dispatch: mask secrets failed", merr, map[string]any{"execution": exec.ID}) + command, err := Interpolate(ctx, d.vars, bag, node.Command) + if err != nil { + _ = catcher.Error("soar dispatch: command interpolation failed", err, map[string]any{"execution": child.ID}) + return + } + params, err := InterpolateJSON(ctx, d.vars, bag, node.Params) + if err != nil { + _ = catcher.Error("soar dispatch: params interpolation failed", err, map[string]any{"execution": child.ID}) + return + } + shell, err := Interpolate(ctx, d.vars, bag, node.Shell) + if err != nil { + _ = catcher.Error("soar dispatch: shell interpolation failed", err, map[string]any{"execution": child.ID}) + return + } + // Children of the same flow-run inherit the parent's agent unless the node + // overrides it. Root-level agent resolution has already run in HandleMatch. + agent := node.Agent + if agent == "" && len(parents) > 0 { + agent = parents[0].Agent + } + if agent != "" { + if resolved, err := Interpolate(ctx, d.vars, bag, agent); err == nil { + agent = resolved } - result = masked } - if err := d.repo.UpdateStatus(ctx, exec.ID, connectors.ExecutionStatusUpdate{ - Status: &executed, - Result: &result, - }); err != nil { - _ = catcher.Error("soar dispatch: failed to persist result", err, map[string]any{"execution": exec.ID}) + + _ = d.exec.TransitionReady(ctx, child.ID, connectors.ReadyUpdate{ + Status: domain.ExecutionStatusPending, + Context: bag, + Params: params, + Command: command, + Shell: shell, + Agent: agent, + }) + d.Kick() +} + +func (d *Dispatcher) loadFlow(exec domain.SoarExecution) *domain.Flow { + sf := d.flows.Get(exec.TenantID.String(), exec.RulePath) + if sf == nil { + return nil } + f := sf.Flow + return &f } func (d *Dispatcher) handleOffline(ctx context.Context, exec domain.SoarExecution) { cause := domain.NonExecutionCauseAgentOffline if exec.Retries+1 >= dispatchMaxRetries { failed := domain.ExecutionStatusFailed - _ = d.repo.UpdateStatus(ctx, exec.ID, connectors.ExecutionStatusUpdate{ + _ = d.exec.UpdateStatus(ctx, exec.ID, connectors.ExecutionStatusUpdate{ Status: &failed, NonExecutionCause: &cause, }) return } - _ = d.repo.UpdateStatus(ctx, exec.ID, connectors.ExecutionStatusUpdate{ + _ = d.exec.UpdateStatus(ctx, exec.ID, connectors.ExecutionStatusUpdate{ NonExecutionCause: &cause, IncrementRetries: true, }) @@ -215,32 +390,8 @@ func (d *Dispatcher) handleOffline(ctx context.Context, exec domain.SoarExecutio func (d *Dispatcher) fail(ctx context.Context, id uuid.UUID, cause domain.NonExecutionCause) { failed := domain.ExecutionStatusFailed - _ = d.repo.UpdateStatus(ctx, id, connectors.ExecutionStatusUpdate{ + _ = d.exec.UpdateStatus(ctx, id, connectors.ExecutionStatusUpdate{ Status: &failed, NonExecutionCause: &cause, }) } - -func (d *Dispatcher) resolveAgent(ctx context.Context, hostname string) (string, bool, error) { - if hostname == "" { - return "", false, nil - } - rows, _, err := d.agent.ListAgents(ctx, "hostname.Is="+hostname) - if err != nil { - return "", false, err - } - if len(rows) == 0 { - return "", false, nil - } - return strconv.FormatUint(uint64(rows[0].GetId()), 10), true, nil -} - -func isOfflineError(err error) bool { - if err == nil { - return false - } - if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound { - return true - } - return strings.Contains(err.Error(), "not found or is disconnected") -} diff --git a/backend/modules/soar/usecase/execution.go b/backend/modules/soar/usecase/execution.go index ff05e1d13..ff81bdc00 100644 --- a/backend/modules/soar/usecase/execution.go +++ b/backend/modules/soar/usecase/execution.go @@ -2,13 +2,11 @@ package usecase import ( "context" - "regexp" - "strings" + "encoding/json" + "time" "github.com/utmstack/utmstack/backend/pkg/authz" - "time" - "github.com/google/uuid" "github.com/threatwinds/go-sdk/catcher" "github.com/tidwall/gjson" @@ -21,114 +19,129 @@ import ( type executionUsecase struct { repo connectors.ExecutionRepository + runs connectors.FlowRunRepository flows *FlowStore agents connectors.AgentUsecase + vars connectors.VariableUsecase notify func() // signals the dispatcher to drain immediately after enqueue (may be nil) } -func NewExecutionUsecase(repo connectors.ExecutionRepository, flows *FlowStore, agents connectors.AgentUsecase, notify func()) connectors.ExecutionUsecase { - return &executionUsecase{repo: repo, flows: flows, agents: agents, notify: notify} +func NewExecutionUsecase( + repo connectors.ExecutionRepository, + runs connectors.FlowRunRepository, + flows *FlowStore, + agents connectors.AgentUsecase, + vars connectors.VariableUsecase, + notify func(), +) connectors.ExecutionUsecase { + return &executionUsecase{repo: repo, runs: runs, flows: flows, agents: agents, vars: vars, notify: notify} } -var commandPlaceholderRE = regexp.MustCompile(`\$\(([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z0-9_]+)*)\)`) - +// HandleMatch starts a new flow run — creates the SoarFlowRun row plus one +// PENDING SoarExecution per declared root. Non-root nodes are spawned lazily +// by the dispatcher as their parents complete. func (u *executionUsecase) HandleMatch(ctx context.Context, req dto.MatchRequest) error { - sf := u.flows.Get(authz.TenantIDFromContext(ctx), req.RulePath) + tenant := authz.TenantIDFromContext(ctx) + sf := u.flows.Get(tenant, req.RulePath) if sf == nil || !sf.Active() { - return nil // flow disabled/deleted since the plugin last loaded it — ignore + return nil } flow := sf.Flow - alertJSON := string(req.Alert) - - target, err := u.resolveAgent(ctx, flow, alertJSON) - if err != nil { - return err + if len(flow.Roots) == 0 || len(flow.Nodes) == 0 { + return nil } - if target == "" { - return nil // nothing to run on (source unmanaged + no default, or excluded) + alertJSON := req.Alert + alertID := gjson.GetBytes(alertJSON, "id").String() + tenantUUID, tenantErr := uuid.Parse(tenant) + if tenantErr != nil { + return catcher.Error("soar: invalid tenant id in context", tenantErr, map[string]any{"tenant": tenant}) } - alertID := gjson.Get(alertJSON, "id").String() - command := buildCommand(assembleChain(flow.Commands), alertJSON) - if command == "" { - return nil - } - if _, err := u.repo.Create(ctx, &domain.SoarExecution{ - Origin: domain.ExecutionOriginFlow, + run, err := u.runs.Create(ctx, &domain.SoarFlowRun{ + TenantID: tenantUUID, RulePath: req.RulePath, AlertID: alertID, - Command: command, - Agent: target, + AlertJSON: alertJSON, + MaxDepth: flow.ResolvedMaxDepth(), Status: domain.ExecutionStatusPending, StartedAt: time.Now().UTC(), - }); err != nil { - return catcher.Error("soar: failed to enqueue execution", err, map[string]any{"rule": req.RulePath, "alert": alertID}) - } - if u.notify != nil { - u.notify() + }) + if err != nil { + return catcher.Error("soar: failed to create flow run", err, map[string]any{"rule": req.RulePath}) } - return nil -} -func assembleChain(cmds []domain.FlowCommand) string { - var b strings.Builder - for i, c := range cmds { - if c.Command == "" { + bag := NewRootContext(alertJSON) + + for _, rootID := range flow.Roots { + node, ok := flow.Nodes[rootID] + if !ok { + _ = catcher.Error("soar: root references missing node", nil, map[string]any{"rule": req.RulePath, "root": rootID}) + continue + } + agent, resErr := u.resolveAgentForNode(ctx, flow, node, bag, alertJSON) + if resErr != nil { + _ = catcher.Error("soar: agent resolution failed", resErr, map[string]any{"rule": req.RulePath, "root": rootID}) + continue + } + params, ierr := InterpolateJSON(ctx, u.vars, bag, node.Params) + if ierr != nil { + _ = catcher.Error("soar: params interpolation failed", ierr, map[string]any{"rule": req.RulePath, "root": rootID}) + continue + } + command, ierr := Interpolate(ctx, u.vars, bag, node.Command) + if ierr != nil { + _ = catcher.Error("soar: command interpolation failed", ierr, map[string]any{"rule": req.RulePath, "root": rootID}) continue } - if b.Len() > 0 { - op := domain.ConditionAlways.Operator() - if i > 0 && c.Condition != nil { - op = c.Condition.Operator() - } - b.WriteByte(' ') - b.WriteString(op) - b.WriteByte(' ') + exec := &domain.SoarExecution{ + TenantID: tenantUUID, + Origin: domain.ExecutionOriginFlow, + FlowRunID: &run.ID, + RulePath: req.RulePath, + AlertID: alertID, + NodeID: rootID, + Depth: 0, + Kind: node.Kind, + Executor: node.Executor, + Params: params, + Context: json.RawMessage(bag), + Command: command, + Shell: node.Shell, + Agent: agent, + Status: domain.ExecutionStatusPending, + StartedAt: time.Now().UTC(), + } + if _, err := u.repo.Create(ctx, exec); err != nil { + return catcher.Error("soar: failed to enqueue root execution", err, + map[string]any{"rule": req.RulePath, "root": rootID}) } - b.WriteString(c.Command) } - return b.String() + if u.notify != nil { + u.notify() + } + return nil } -func (u *executionUsecase) resolveAgent(ctx context.Context, flow domain.Flow, alertJSON string) (string, error) { - src := gjson.Get(alertJSON, "dataSource").String() - if agentInList(flow.ExcludedAgents, src) { +// resolveAgentForNode picks the target agent for a shell node. Non-shell +// executors return "" (no agent needed). Node.Agent wins when set; otherwise +// we default to the box that raised the alert (alert.dataSource) — the common +// "restart the service on the affected host" pattern. ExcludedAgents only +// applies in the auto-resolve path: an explicit Agent is treated as an +// operator override that bypasses the deny list. +func (u *executionUsecase) resolveAgentForNode(ctx context.Context, _ domain.Flow, node domain.FlowNode, bag ContextBag, alertJSON []byte) (string, error) { + if node.Executor != "shell" { return "", nil } - platformAgents, err := u.agents.ListByPlatform(ctx, flow.AgentPlatform) - if err != nil { - return "", err - } - if src != "" && agentInList(platformAgents, src) { - return src, nil - } - if flow.DefaultAgent != "" { - return flow.DefaultAgent, nil - } - return "", nil -} - -func buildCommand(template, alertJSON string) string { - return commandPlaceholderRE.ReplaceAllStringFunc(template, func(match string) string { - field := strings.TrimSuffix(strings.TrimPrefix(match, "$("), ")") - val := gjson.Get(alertJSON, field) - if !val.Exists() { - return match - } - return val.String() - }) -} - -func agentInList(list []string, v string) bool { - if v == "" { - return false + if node.Agent != "" { + return Interpolate(ctx, u.vars, bag, node.Agent) } - for _, x := range list { - if x == v { - return true + src := gjson.GetBytes(alertJSON, "dataSource").String() + for _, x := range node.ExcludedAgents { + if x == src { + return "", nil // returns empty → shell executor fails with ErrAgentNotFound → on_error fires } } - return false + return src, nil } func (u *executionUsecase) List(ctx context.Context, f dto.ExecutionFilters) (*database.List[dto.ExecutionResponse], error) { @@ -164,6 +177,11 @@ func (u *executionUsecase) List(ctx context.Context, f dto.ExecutionFilters) (*d FinishedAt: e.FinishedAt, NonExecutionCause: e.NonExecutionCause, Retries: e.Retries, + NodeID: e.NodeID, + Kind: e.Kind, + Executor: e.Executor, + FlowRunID: e.FlowRunID, + Depth: e.Depth, } } @@ -176,6 +194,9 @@ func (u *executionUsecase) StartManual(ctx context.Context, agent, command, trig TriggeredBy: triggeredBy, Agent: agent, Command: command, + Executor: "shell", + Kind: domain.NodeKindExecutor, + NodeID: "manual", Status: domain.ExecutionStatusPending, StartedAt: time.Now().UTC(), }) diff --git a/backend/modules/soar/usecase/flow_store.go b/backend/modules/soar/usecase/flow_store.go index 8af5ab3e1..778eba72d 100644 --- a/backend/modules/soar/usecase/flow_store.go +++ b/backend/modules/soar/usecase/flow_store.go @@ -24,9 +24,8 @@ type FlowListFilter struct { Name string // case-insensitive partial on flow name Search string // case-insensitive partial on flow name - Active *bool // enabled state - SystemOwner *bool // shipped with the product rather than written here - AgentPlatform string // exact match + Active *bool // enabled state + SystemOwner *bool // shipped with the product rather than written here } type flowKey struct { @@ -263,9 +262,6 @@ func flowMatches(sf *domain.StoredFlow, f FlowListFilter) bool { if f.SystemOwner != nil && sf.System != *f.SystemOwner { return false } - if f.AgentPlatform != "" && sf.AgentPlatform != f.AgentPlatform { - return false - } return true } diff --git a/backend/modules/soar/usecase/flow_writer.go b/backend/modules/soar/usecase/flow_writer.go index 490c2b993..7373ee988 100644 --- a/backend/modules/soar/usecase/flow_writer.go +++ b/backend/modules/soar/usecase/flow_writer.go @@ -1,6 +1,7 @@ package usecase import ( + "encoding/json" "fmt" "os" "path/filepath" @@ -41,20 +42,144 @@ func writeFlowFile(path string, flow domain.Flow) error { return nil } +// readFlowFile reads one Flow from disk. It accepts the current DAG shape +// (`roots:` + `nodes:`) and transparently upgrades the legacy chain shape +// (`commands:` + `shell:`) so existing on-disk playbooks keep parsing. func readFlowFile(path string) (domain.Flow, error) { var flow domain.Flow data, err := os.ReadFile(path) if err != nil { return flow, err } - var list []domain.Flow + var list []legacyOrDAGFlow if err := yaml.Unmarshal(data, &list); err != nil { return flow, err } if len(list) == 0 { return flow, fmt.Errorf("flow file %s contains no flows", path) } - return list[0], nil + return list[0].asFlow() +} + +// legacyOrDAGFlow captures both YAML shapes so we can pick which one this file +// used and normalize to the DAG shape. It carries every field either shape +// recognises; the mapper checks whether nodes/roots or commands is populated +// and produces a canonical domain.Flow. +type legacyOrDAGFlow struct { + Name string `yaml:"name"` + Description string `yaml:"description,omitempty"` + Conditions []domain.FilterType `yaml:"conditions"` + MaxDepth int `yaml:"maxDepth,omitempty"` + Roots []string `yaml:"roots,omitempty"` + Nodes map[string]domain.FlowNode `yaml:"nodes,omitempty"` + + // Legacy chain shape. Platform + default agent are copied into every + // legacy shell node so the upgrade preserves behavior. + Commands []legacyCommand `yaml:"commands,omitempty"` + Shell string `yaml:"shell,omitempty"` + AgentPlatform string `yaml:"agentPlatform,omitempty"` + DefaultAgent string `yaml:"defaultAgent,omitempty"` + ExcludedAgents []string `yaml:"excludedAgents,omitempty"` +} + +type legacyCommand struct { + Command string `yaml:"command"` + Condition *string `yaml:"condition,omitempty"` +} + +// UnmarshalYAML lets old files write a bare string for a command, matching the +// previous behaviour where `- systemctl restart wazuh` was valid. +func (lc *legacyCommand) UnmarshalYAML(value *yaml.Node) error { + if value.Kind == yaml.ScalarNode { + lc.Command = value.Value + return nil + } + type raw legacyCommand + var r raw + if err := value.Decode(&r); err != nil { + return err + } + *lc = legacyCommand(r) + return nil +} + +func (l legacyOrDAGFlow) asFlow() (domain.Flow, error) { + base := domain.Flow{ + Name: l.Name, + Description: l.Description, + Conditions: l.Conditions, + MaxDepth: l.MaxDepth, + } + if len(l.Nodes) > 0 { + base.Roots = l.Roots + base.Nodes = l.Nodes + return base, nil + } + // ponytail: legacy chain → linear DAG. Every command becomes one shell + // node; Condition (OnSuccess/OnFailure/Always) decides which of the + // previous node's edges points here. Legacy flow-level platform/agent + // are stamped onto every shell node so behavior survives the upgrade. + base.Nodes = make(map[string]domain.FlowNode, len(l.Commands)) + if len(l.Commands) == 0 { + return base, nil + } + base.Roots = []string{"step_0"} + for i, c := range l.Commands { + id := fmt.Sprintf("step_%d", i) + node := domain.FlowNode{ + Kind: domain.NodeKindExecutor, + Executor: "shell", + Command: c.Command, + Shell: l.Shell, + Platform: l.AgentPlatform, + Agent: l.DefaultAgent, + ExcludedAgents: l.ExcludedAgents, + } + if len(c.Command) > 0 && looksLikeJSONParams(c.Command) { + // Not expected in existing bundled flows, but keeps the door open. + node.Params = json.RawMessage(c.Command) + node.Command = "" + } + base.Nodes[id] = node + if i == 0 { + continue + } + prevID := fmt.Sprintf("step_%d", i-1) + prev := base.Nodes[prevID] + branch := legacyBranchFromCondition(c.Condition) + switch branch { + case domain.EdgeBranchSuccess: + prev.OnSuccess = append(prev.OnSuccess, id) + case domain.EdgeBranchError: + prev.OnError = append(prev.OnError, id) + default: + prev.OnSuccess = append(prev.OnSuccess, id) + prev.OnError = append(prev.OnError, id) + } + base.Nodes[prevID] = prev + } + return base, nil +} + +func legacyBranchFromCondition(cond *string) domain.EdgeBranch { + if cond == nil { + return "always" + } + switch *cond { + case "OnSuccess": + return domain.EdgeBranchSuccess + case "OnFailure": + return domain.EdgeBranchError + default: + return "always" + } +} + +func looksLikeJSONParams(s string) bool { + if len(s) < 2 { + return false + } + return s[0] == '{' && s[len(s)-1] == '}' } func renameFlowFile(src, dst string) error { diff --git a/backend/modules/soar/usecase/flow_writer_test.go b/backend/modules/soar/usecase/flow_writer_test.go new file mode 100644 index 000000000..6709835e6 --- /dev/null +++ b/backend/modules/soar/usecase/flow_writer_test.go @@ -0,0 +1,104 @@ +package usecase + +import ( + "os" + "path/filepath" + "testing" + + "github.com/utmstack/utmstack/backend/modules/soar/domain" +) + +// Existing on-disk playbooks use the old `commands:` chain shape. The reader +// has to keep parsing them, upgrading to a linear DAG at load time — otherwise +// every deployment loses its shipped flows the day this ships. +func TestReadFlowFile_LegacyChainUpgrades(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "legacy.yaml") + body := []byte(` +- name: Restart wazuh + agentPlatform: linux + commands: + - systemctl restart wazuh + - echo done + shell: bash +`) + if err := os.WriteFile(path, body, 0o644); err != nil { + t.Fatal(err) + } + flow, err := readFlowFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if len(flow.Roots) != 1 || flow.Roots[0] != "step_0" { + t.Fatalf("roots = %v, want [step_0]", flow.Roots) + } + if flow.Nodes["step_0"].Kind != domain.NodeKindExecutor { + t.Errorf("step_0 kind = %q", flow.Nodes["step_0"].Kind) + } + if got := flow.Nodes["step_0"].Command; got != "systemctl restart wazuh" { + t.Errorf("step_0 command = %q", got) + } + if got := flow.Nodes["step_0"].Shell; got != "bash" { + t.Errorf("step_0 shell = %q, want bash", got) + } + // A legacy step with no `condition:` links via both edges — the old + // `;` operator ran the next command regardless of outcome. + step0 := flow.Nodes["step_0"] + if len(step0.OnSuccess) != 1 || step0.OnSuccess[0] != "step_1" { + t.Errorf("step_0.OnSuccess = %v", step0.OnSuccess) + } + if len(step0.OnError) != 1 || step0.OnError[0] != "step_1" { + t.Errorf("step_0.OnError = %v", step0.OnError) + } +} + +func TestReadFlowFile_DAGShapePreserved(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "dag.yaml") + body := []byte(` +- name: Fan-in + agentPlatform: linux + roots: [geoip, extract] + nodes: + geoip: + kind: enrichment + executor: http + params: {"url":"https://geo"} + onSuccess: [notify] + extract: + kind: enrichment + executor: select + params: {"fields":{"user":"alert.user.name"}} + onSuccess: [notify] + notify: + kind: executor + executor: http + params: {"url":"https://slack"} +`) + if err := os.WriteFile(path, body, 0o644); err != nil { + t.Fatal(err) + } + flow, err := readFlowFile(path) + if err != nil { + t.Fatal(err) + } + if len(flow.Roots) != 2 { + t.Errorf("roots = %v", flow.Roots) + } + if flow.Nodes["geoip"].Kind != domain.NodeKindEnrichment { + t.Errorf("geoip kind = %q", flow.Nodes["geoip"].Kind) + } + // notify must be the AND-join sink — both roots reference it. + if got := flow.IncomingCounts()["notify"]; got != 2 { + t.Errorf("incoming[notify] = %d, want 2", got) + } +} + +func TestFlow_ResolvedMaxDepthDefaults(t *testing.T) { + if got := (domain.Flow{}).ResolvedMaxDepth(); got != domain.DefaultMaxDepth { + t.Errorf("default = %d, want %d", got, domain.DefaultMaxDepth) + } + if got := (domain.Flow{MaxDepth: 7}).ResolvedMaxDepth(); got != 7 { + t.Errorf("explicit = %d, want 7", got) + } +} diff --git a/backend/modules/soar/usecase/interpolate.go b/backend/modules/soar/usecase/interpolate.go new file mode 100644 index 000000000..d0aa9c909 --- /dev/null +++ b/backend/modules/soar/usecase/interpolate.go @@ -0,0 +1,110 @@ +package usecase + +import ( + "context" + "encoding/json" + "regexp" + "strings" + + "github.com/tidwall/gjson" + + "github.com/utmstack/utmstack/backend/modules/soar/connectors" +) + +// placeholderRE matches `$(a.b.c)` — a dotted path whose first segment is +// either "alert", a variable-context key, or an enrichment node id. +var placeholderRE = regexp.MustCompile(`\$\(([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z0-9_-]+)*)\)`) + +// ContextBag is the merged JSON context an execution sees. Top-level keys are +// "alert" plus one entry per ancestor enrichment node. +type ContextBag json.RawMessage + +// NewRootContext builds the context every root execution starts with: just the +// frozen alert payload keyed under "alert". Every downstream node inherits and +// augments this bag. +func NewRootContext(alertJSON json.RawMessage) ContextBag { + if len(alertJSON) == 0 { + return ContextBag(`{"alert":{}}`) + } + // Wrap the alert JSON under an "alert" key. Since alertJSON may be raw and + // large, we build the wrapper by string concat rather than round-tripping + // through map[string]any. + buf := make([]byte, 0, len(alertJSON)+16) + buf = append(buf, `{"alert":`...) + buf = append(buf, alertJSON...) + buf = append(buf, '}') + return ContextBag(buf) +} + +// MergeContexts produces a child's context bag from its resolved parents. +// Each parent contributes its own context, plus (for enrichment parents only) +// its output stored under the parent's node id. Later parents overwrite earlier +// ones on key collision — cycles land the deepest instance last, so retries +// win, matching the "most recent value" intuition. +func MergeContexts(parents []ParentContribution) ContextBag { + merged := map[string]json.RawMessage{} + for _, p := range parents { + if len(p.Context) > 0 { + var kv map[string]json.RawMessage + if err := json.Unmarshal(p.Context, &kv); err == nil { + for k, v := range kv { + merged[k] = v + } + } + } + if p.EnrichmentNodeID != "" && len(p.Output) > 0 { + merged[p.EnrichmentNodeID] = p.Output + } + } + if len(merged) == 0 { + return ContextBag(`{}`) + } + raw, err := json.Marshal(merged) + if err != nil { + return ContextBag(`{}`) + } + return ContextBag(raw) +} + +// ParentContribution is one parent's slice of state that flows into the child's +// context. EnrichmentNodeID is empty when the parent is an executor node. +type ParentContribution struct { + EnrichmentNodeID string + Context json.RawMessage + Output json.RawMessage +} + +// Interpolate substitutes `$(...)` placeholders in `input` using the merged +// context and then applies `$[variables.NAME]` secrets via the variable +// usecase. The two syntaxes are independent — a template may use either or +// both. Returns the input verbatim when nothing matches. +func Interpolate(ctx context.Context, vars connectors.VariableUsecase, bag ContextBag, input string) (string, error) { + if input == "" { + return "", nil + } + out := placeholderRE.ReplaceAllStringFunc(input, func(match string) string { + path := strings.TrimSuffix(strings.TrimPrefix(match, "$("), ")") + val := gjson.GetBytes(bag, path) + if !val.Exists() { + return match + } + return val.String() + }) + if vars == nil { + return out, nil + } + return vars.InterpolateCommand(ctx, out) +} + +// InterpolateJSON is a shortcut for interpolating a JSON blob and returning it +// as json.RawMessage. When input is empty, returns nil (nothing to write). +func InterpolateJSON(ctx context.Context, vars connectors.VariableUsecase, bag ContextBag, input json.RawMessage) (json.RawMessage, error) { + if len(input) == 0 { + return nil, nil + } + s, err := Interpolate(ctx, vars, bag, string(input)) + if err != nil { + return nil, err + } + return json.RawMessage(s), nil +} diff --git a/backend/modules/soar/usecase/interpolate_test.go b/backend/modules/soar/usecase/interpolate_test.go new file mode 100644 index 000000000..6dfe6ef88 --- /dev/null +++ b/backend/modules/soar/usecase/interpolate_test.go @@ -0,0 +1,63 @@ +package usecase + +import ( + "context" + "encoding/json" + "testing" +) + +func TestInterpolate_AlertLookup(t *testing.T) { + bag := NewRootContext(json.RawMessage(`{"src_ip":"10.0.0.5","user":{"name":"alice"}}`)) + got, err := Interpolate(context.Background(), nil, bag, "block $(alert.src_ip) for $(alert.user.name)") + if err != nil { + t.Fatal(err) + } + if want := "block 10.0.0.5 for alice"; got != want { + t.Errorf("got %q, want %q", got, want) + } +} + +func TestInterpolate_UnknownPathStaysLiteral(t *testing.T) { + bag := NewRootContext(json.RawMessage(`{"src_ip":"10.0.0.5"}`)) + got, err := Interpolate(context.Background(), nil, bag, "$(alert.missing) $(alert.src_ip)") + if err != nil { + t.Fatal(err) + } + // Missing paths stay verbatim so an operator can spot the typo instead of + // running a command with a silently blank field. + if want := "$(alert.missing) 10.0.0.5"; got != want { + t.Errorf("got %q, want %q", got, want) + } +} + +func TestMergeContexts_EnrichmentOutputPropagates(t *testing.T) { + alertBag := NewRootContext(json.RawMessage(`{"id":"a-1"}`)) + geoOutput := json.RawMessage(`{"country":"US","asn":15169}`) + merged := MergeContexts([]ParentContribution{ + {Context: json.RawMessage(alertBag)}, + {EnrichmentNodeID: "geoip", Output: geoOutput, Context: json.RawMessage(alertBag)}, + }) + got, err := Interpolate(context.Background(), nil, merged, "hit from $(geoip.country) alert=$(alert.id)") + if err != nil { + t.Fatal(err) + } + if want := "hit from US alert=a-1"; got != want { + t.Errorf("got %q, want %q", got, want) + } +} + +func TestMergeContexts_ExecutorParentContributesNothing(t *testing.T) { + base := NewRootContext(json.RawMessage(`{"id":"a-1"}`)) + // Parent is an executor (no EnrichmentNodeID) — its Output must not surface + // in the merged bag even if supplied by mistake. + merged := MergeContexts([]ParentContribution{ + {Context: json.RawMessage(base), Output: json.RawMessage(`{"leaked":true}`)}, + }) + got, err := Interpolate(context.Background(), nil, merged, "$(leaked.something)") + if err != nil { + t.Fatal(err) + } + if got != "$(leaked.something)" { + t.Errorf("executor output leaked into context: %q", got) + } +} diff --git a/backend/modules/soar/usecase/rule.go b/backend/modules/soar/usecase/rule.go index 4a54bf5f3..cc7471c9a 100644 --- a/backend/modules/soar/usecase/rule.go +++ b/backend/modules/soar/usecase/rule.go @@ -23,7 +23,7 @@ func NewRuleUsecase(store *FlowStore, resolve connectors.ResolveFilterRepository } func (u *ruleUsecase) Create(ctx context.Context, req dto.CreateRuleRequest, createdBy string) (*dto.RuleResponse, error) { - sf, err := u.store.Create(tenantOf(ctx), requestToFlow(req.Name, req.Description, req.Conditions, req.Commands, req.Shell, req.AgentPlatform, req.DefaultAgent, req.ExcludedAgents)) + sf, err := u.store.Create(tenantOf(ctx), requestToFlow(req.Name, req.Description, req.Conditions, req.Roots, req.Nodes, req.MaxDepth)) if err != nil { return nil, mapStoreErr(err) } @@ -35,7 +35,7 @@ func (u *ruleUsecase) Create(ctx context.Context, req dto.CreateRuleRequest, cre } func (u *ruleUsecase) Update(ctx context.Context, relPath string, req dto.UpdateRuleRequest, modifiedBy string) (*dto.RuleResponse, error) { - sf, err := u.store.Update(tenantOf(ctx), relPath, requestToFlow(req.Name, req.Description, req.Conditions, req.Commands, req.Shell, req.AgentPlatform, req.DefaultAgent, req.ExcludedAgents)) + sf, err := u.store.Update(tenantOf(ctx), relPath, requestToFlow(req.Name, req.Description, req.Conditions, req.Roots, req.Nodes, req.MaxDepth)) if err != nil { return nil, mapStoreErr(err) } @@ -64,12 +64,11 @@ func (u *ruleUsecase) SetEnabled(ctx context.Context, relPath string, enabled bo func (u *ruleUsecase) List(ctx context.Context, f dto.RuleFilters) (*database.List[dto.RuleResponse], error) { flows, total := u.store.List(tenantOf(ctx), FlowListFilter{ - Page: f.Page, - Size: f.Size, - Name: f.RuleName, - Active: f.RuleActive, - SystemOwner: f.SystemOwner, - AgentPlatform: f.AgentPlatform, + Page: f.Page, + Size: f.Size, + Name: f.RuleName, + Active: f.RuleActive, + SystemOwner: f.SystemOwner, }) items := make([]dto.RuleResponse, 0, len(flows)) for _, sf := range flows { @@ -105,16 +104,14 @@ func mapStoreErr(err error) error { } } -func requestToFlow(name, description string, conds []dto.FilterVM, commands []dto.FlowCommandVM, shell, agentPlatform, defaultAgent string, excludedAgents []string) domain.Flow { +func requestToFlow(name, description string, conds []dto.FilterVM, roots []string, nodes map[string]dto.FlowNodeVM, maxDepth int) domain.Flow { return domain.Flow{ - Name: name, - Description: description, - Conditions: toFlowConditions(conds), - Commands: toFlowCommands(commands), - Shell: shell, - AgentPlatform: agentPlatform, - DefaultAgent: defaultAgent, - ExcludedAgents: excludedAgents, + Name: name, + Description: description, + Conditions: toFlowConditions(conds), + Roots: roots, + Nodes: toFlowNodes(nodes), + MaxDepth: maxDepth, } } @@ -126,18 +123,40 @@ func toFlowConditions(vms []dto.FilterVM) []domain.FilterType { return out } -func toFlowCommands(vms []dto.FlowCommandVM) []domain.FlowCommand { - out := make([]domain.FlowCommand, 0, len(vms)) - for _, v := range vms { - out = append(out, domain.FlowCommand{Command: v.Command, Condition: v.Condition}) +func toFlowNodes(vms map[string]dto.FlowNodeVM) map[string]domain.FlowNode { + out := make(map[string]domain.FlowNode, len(vms)) + for id, v := range vms { + out[id] = domain.FlowNode{ + Kind: v.Kind, + Executor: v.Executor, + Command: v.Command, + Shell: v.Shell, + Platform: v.Platform, + Agent: v.Agent, + ExcludedAgents: v.ExcludedAgents, + Params: v.Params, + OnSuccess: v.OnSuccess, + OnError: v.OnError, + } } return out } -func flowCommandsToVMs(cmds []domain.FlowCommand) []dto.FlowCommandVM { - out := make([]dto.FlowCommandVM, 0, len(cmds)) - for _, c := range cmds { - out = append(out, dto.FlowCommandVM{Command: c.Command, Condition: c.Condition}) +func flowNodesToVMs(nodes map[string]domain.FlowNode) map[string]dto.FlowNodeVM { + out := make(map[string]dto.FlowNodeVM, len(nodes)) + for id, n := range nodes { + out[id] = dto.FlowNodeVM{ + Kind: n.Kind, + Executor: n.Executor, + Command: n.Command, + Shell: n.Shell, + Platform: n.Platform, + Agent: n.Agent, + ExcludedAgents: n.ExcludedAgents, + Params: n.Params, + OnSuccess: n.OnSuccess, + OnError: n.OnError, + } } return out } @@ -147,16 +166,14 @@ func storedFlowToResponse(sf *domain.StoredFlow) *dto.RuleResponse { return nil } resp := &dto.RuleResponse{ - RelPath: sf.RelPath, - Name: sf.Name, - Description: sf.Description, - Commands: flowCommandsToVMs(sf.Commands), - Active: sf.Active(), - AgentPlatform: sf.AgentPlatform, - DefaultAgent: sf.DefaultAgent, - Shell: sf.Shell, - SystemOwner: sf.SystemOwned(), - ExcludedAgents: sf.ExcludedAgents, + RelPath: sf.RelPath, + Name: sf.Name, + Description: sf.Description, + Roots: sf.Roots, + Nodes: flowNodesToVMs(sf.Nodes), + MaxDepth: sf.MaxDepth, + Active: sf.Active(), + SystemOwner: sf.SystemOwned(), } for _, c := range sf.Conditions { resp.Conditions = append(resp.Conditions, dto.FilterVM{Operator: domain.OperatorType(c.Operator), Field: c.Field, Value: c.Value}) From 60da43e0526a95fe100a3d2698f40c33a24f1030 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alex=20S=C3=A1nchez?= Date: Tue, 25 Aug 2026 11:08:26 -0600 Subject: [PATCH 2/6] feat[frontend](soar): DAG canvas editor for flows --- frontend/package.json | 1 + .../features/soar/components/AgentPicker.tsx | 270 +++++++ .../soar/components/ExecutionsView.tsx | 5 +- .../features/soar/components/FlowCanvas.tsx | 372 +++++++++ .../features/soar/components/FlowEditor.tsx | 742 ++++-------------- .../soar/components/InsertFieldMenu.tsx | 136 ++++ .../soar/components/NodeInspector.tsx | 295 +++++++ .../features/soar/components/NodePalette.tsx | 65 ++ .../soar/components/nodes/DAGNode.tsx | 81 ++ .../soar/components/nodes/TriggerNode.tsx | 28 + frontend/src/features/soar/lib/ancestors.ts | 63 ++ .../features/soar/lib/command-templates.ts | 41 +- frontend/src/features/soar/lib/flow-yaml.ts | 170 ++-- .../src/features/soar/pages/FlowsPage.tsx | 29 +- .../soar/services/soar-flows.service.ts | 1 - .../src/features/soar/types/soar.types.ts | 88 ++- 16 files changed, 1683 insertions(+), 704 deletions(-) create mode 100644 frontend/src/features/soar/components/AgentPicker.tsx create mode 100644 frontend/src/features/soar/components/FlowCanvas.tsx create mode 100644 frontend/src/features/soar/components/InsertFieldMenu.tsx create mode 100644 frontend/src/features/soar/components/NodeInspector.tsx create mode 100644 frontend/src/features/soar/components/NodePalette.tsx create mode 100644 frontend/src/features/soar/components/nodes/DAGNode.tsx create mode 100644 frontend/src/features/soar/components/nodes/TriggerNode.tsx create mode 100644 frontend/src/features/soar/lib/ancestors.ts diff --git a/frontend/package.json b/frontend/package.json index 69b1397e4..71a1ca9e3 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -19,6 +19,7 @@ "@radix-ui/react-slot": "^1.1.1", "@radix-ui/react-tooltip": "^1.1.6", "@tanstack/react-query": "^5.101.0", + "@xyflow/react": "^12.11.3", "axios": "^1.7.7", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", diff --git a/frontend/src/features/soar/components/AgentPicker.tsx b/frontend/src/features/soar/components/AgentPicker.tsx new file mode 100644 index 000000000..1f3f6886d --- /dev/null +++ b/frontend/src/features/soar/components/AgentPicker.tsx @@ -0,0 +1,270 @@ +import { useEffect, useMemo, useState } from 'react' +import { Check, ChevronDown, Plus, X } from 'lucide-react' +import { cn } from '@/shared/lib/utils' +import { Input } from '@/shared/components/ui/input' +import { datasourcesHttpService } from '@/features/datasources/services/datasources-http.service' +import { COMMON_PLATFORMS, defaultShellForPlatform, shellsForPlatform } from '../lib/alert-fields' + +const SELECT = 'h-8 rounded-md border border-input bg-background px-2 text-xs focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring' + +interface AgentOption { + name: string + platform: string +} + +interface Props { + platform?: string + agent?: string + excludedAgents?: string[] + shell?: string + readOnly?: boolean + onChange: (patch: { platform?: string; agent?: string; excludedAgents?: string[]; shell?: string }) => void +} + +type Scope = 'auto' | 'specific' | 'except' + +function scopeFrom(agent?: string, excluded?: string[]): Scope { + if (agent) return 'specific' + if (excluded && excluded.length) return 'except' + return 'auto' +} + +/** Per-node platform + agent + shell picker for `shell` nodes. Three modes: + * + * - auto: run on `$(alert.dataSource)` (the box that raised the alert) + * - specific: pick one hostname (or a template) + * - except: run auto, but skip if the alert source is on the exclusion list + * + * Platform is a UI-only filter (limits the agent list); the runtime uses + * `agent` + `excludedAgents`. */ +export function AgentPicker({ platform, agent, excludedAgents, shell, readOnly, onChange }: Props) { + const [agents, setAgents] = useState([]) + const [scope, setScope] = useState(() => scopeFrom(agent, excludedAgents)) + + // Keep the tab in sync when the caller changes agent/excluded externally. + useEffect(() => { + setScope(scopeFrom(agent, excludedAgents)) + }, [agent, excludedAgents]) + + useEffect(() => { + datasourcesHttpService + .list({ page: 1, size: 1000, kind: 'agent' }) + .then((r) => + setAgents( + (r.items ?? []).map((d) => ({ + name: d.name, + platform: typeof d.metadata?.osPlatform === 'string' ? d.metadata.osPlatform : '', + })), + ), + ) + .catch(() => setAgents([])) + }, []) + + const platforms = useMemo(() => { + const seen = new Set() + const out: string[] = [] + for (const p of [...COMMON_PLATFORMS, ...agents.map((a) => a.platform), platform ?? '']) { + const v = p.trim() + if (v && !seen.has(v.toLowerCase())) { + seen.add(v.toLowerCase()) + out.push(v) + } + } + return out + }, [agents, platform]) + + const platformAgents = useMemo(() => { + const p = (platform ?? '').trim().toLowerCase() + if (!p) return agents + const m = agents.filter((a) => a.platform && (a.platform.toLowerCase().includes(p) || p.includes(a.platform.toLowerCase()))) + return m.length ? m : agents + }, [agents, platform]) + + const shells = shellsForPlatform(platform ?? '') + + const onPlatformChange = (v: string) => { + const nextShell = shells.includes(shell ?? '') ? shell : defaultShellForPlatform(v) + onChange({ platform: v || undefined, shell: nextShell || undefined }) + } + + const switchScope = (next: Scope) => { + if (next === scope) return + setScope(next) + if (next === 'auto') onChange({ agent: undefined, excludedAgents: undefined }) + else if (next === 'specific') onChange({ agent: undefined, excludedAgents: undefined }) + else onChange({ agent: undefined, excludedAgents: excludedAgents ?? [] }) + } + + const options = platformAgents.map((a) => a.name) + + return ( +
+
+ + + + + + +
+ + +
+ {(['auto', 'specific', 'except'] as const).map((s) => ( + + ))} +
+ + + {scope === 'auto' && ( +

+ Runs on $(alert.dataSource) — the host that + raised the matched alert. +

+ )} + + {scope === 'specific' && ( +
+ +

+ Or type a template — $(alert.field) resolves per + execution. +

+ onChange({ agent: e.target.value || undefined, excludedAgents: undefined })} + placeholder="$(alert.dataSource)" + className="h-8 font-mono text-xs" + /> +
+ )} + + {scope === 'except' && ( + onChange({ agent: undefined, excludedAgents: v.length ? v : undefined })} + /> + )} +
+ ) +} + +function Field({ label, children }: { label: string; children: React.ReactNode }) { + return ( +
+ + {children} +
+ ) +} + +/** Searchable multi-select of agent hostnames. Used for the exclusion list on + * the shell node. Kept local so it doesn't accidentally bleed into other + * parts of the UI. */ +function AgentMultiSelect({ + options, + values, + readOnly, + onChange, +}: { + options: string[] + values: string[] + readOnly?: boolean + onChange: (v: string[]) => void +}) { + const [open, setOpen] = useState(false) + const [q, setQ] = useState('') + + const toggle = (name: string) => onChange(values.includes(name) ? values.filter((x) => x !== name) : [...values, name]) + const filtered = options.filter((o) => (q ? o.toLowerCase().includes(q.toLowerCase()) : true)) + + return ( +
+
+ {values.map((v) => ( + + {v} + {!readOnly && ( + + )} + + ))} + {!readOnly && ( + + )} +
+ {open && !readOnly && ( +
+
+ setQ(e.target.value)} + autoFocus + placeholder="search agents…" + className="h-7 w-full rounded-md border border-input bg-background px-2 text-xs focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring" + /> +
+
+ {filtered.length === 0 &&
no agents
} + {filtered.map((o) => { + const on = values.includes(o) + return ( + + ) + })} +
+
+ )} +
+ ) +} diff --git a/frontend/src/features/soar/components/ExecutionsView.tsx b/frontend/src/features/soar/components/ExecutionsView.tsx index 3dd642fd1..0d7c07dd8 100644 --- a/frontend/src/features/soar/components/ExecutionsView.tsx +++ b/frontend/src/features/soar/components/ExecutionsView.tsx @@ -11,14 +11,17 @@ import { datasourcesHttpService } from '@/features/datasources/services/datasour import { soarExecutionsService } from '../services/soar-executions.service' import type { Execution, ExecutionOrigin, ExecutionStatus, ExecutionListQuery } from '../types/soar.types' -const STATUSES: (ExecutionStatus | 'all')[] = ['all', 'EXECUTED', 'PENDING', 'FAILED'] +const STATUSES: (ExecutionStatus | 'all')[] = ['all', 'EXECUTED', 'PENDING', 'WAITING', 'EXECUTING', 'FAILED', 'DEAD'] const ORIGINS: (ExecutionOrigin | 'all')[] = ['all', 'FLOW', 'MANUAL'] const COLS = '90px minmax(160px,1.2fr) minmax(180px,1.6fr) 120px 150px 60px' const STATUS_META: Record = { EXECUTED: { icon: CheckCircle2, cls: 'text-emerald-500' }, PENDING: { icon: Clock, cls: 'text-amber-500' }, + WAITING: { icon: Clock, cls: 'text-muted-foreground' }, + EXECUTING: { icon: Loader2, cls: 'text-sky-500 animate-spin' }, FAILED: { icon: XCircle, cls: 'text-red-500' }, + DEAD: { icon: AlertTriangle, cls: 'text-muted-foreground' }, } export function ExecutionsView() { diff --git a/frontend/src/features/soar/components/FlowCanvas.tsx b/frontend/src/features/soar/components/FlowCanvas.tsx new file mode 100644 index 000000000..87bfd6ab8 --- /dev/null +++ b/frontend/src/features/soar/components/FlowCanvas.tsx @@ -0,0 +1,372 @@ +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import { + Background, + Controls, + MiniMap, + ReactFlow, + ReactFlowProvider, + addEdge, + useEdgesState, + useNodesState, + useReactFlow, + type Connection, + type Edge, + type EdgeChange, + type Node, + type NodeChange, + type NodeTypes, +} from '@xyflow/react' +import '@xyflow/react/dist/style.css' +import { ChevronLeft, ChevronRight, PanelLeft, PanelRight } from 'lucide-react' +import { cn } from '@/shared/lib/utils' +import type { FlowNode, NodeKind } from '../types/soar.types' +import { NodePalette } from './NodePalette' +import { NodeInspector } from './NodeInspector' +import { DAGNode } from './nodes/DAGNode' +import { TriggerNode } from './nodes/TriggerNode' + +const NODE_TYPES: NodeTypes = { dag: DAGNode as unknown as NodeTypes[string], trigger: TriggerNode as unknown as NodeTypes[string] } +const TRIGGER_ID = '__trigger__' + +interface Props { + roots: string[] + nodes: Record + readOnly?: boolean + onChange: (patch: { roots: string[]; nodes: Record }) => void +} + +/** Node-red style DAG editor for a SOAR flow. Nodes come from the flow's + * `nodes` map; edges are derived from each node's `onSuccess`/`onError`. A + * virtual trigger node hosts the roots list — dragging from it adds a root. */ +export function FlowCanvas(props: Props) { + return ( + + + + ) +} + +function FlowCanvasInner({ roots, nodes, readOnly, onChange }: Props) { + const wrapperRef = useRef(null) + const { screenToFlowPosition } = useReactFlow() + + // Layout positions are stashed by node id so they survive re-derivation + // whenever the flow model updates. Fresh nodes get a top-down default. + const layoutRef = useRef>({}) + const [selectedId, setSelectedId] = useState(null) + const [paletteOpen, setPaletteOpen] = useState(true) + const [inspectorOpen, setInspectorOpen] = useState(true) + + const { rfNodes, rfEdges } = useMemo(() => { + const posFor = (id: string, fallback: { x: number; y: number }) => + layoutRef.current[id] ?? (layoutRef.current[id] = fallback) + + const rfN: Node[] = [ + { + id: TRIGGER_ID, + type: 'trigger', + position: posFor(TRIGGER_ID, { x: 120, y: 0 }), + data: {}, + draggable: !readOnly, + deletable: false, + }, + ] + let i = 0 + for (const [id, n] of Object.entries(nodes)) { + rfN.push({ + id, + type: 'dag', + position: posFor(id, { x: (i % 3) * 260, y: 180 + Math.floor(i / 3) * 180 }), + data: { nodeId: id, ...n } as unknown as Record, + selected: id === selectedId, + draggable: !readOnly, + }) + i++ + } + + const rfE: Edge[] = [] + roots.forEach((r) => + rfE.push({ + id: `${TRIGGER_ID}->${r}`, + source: TRIGGER_ID, + sourceHandle: 'trigger', + target: r, + animated: true, + style: { stroke: '#f59e0b', strokeWidth: 2 }, + }), + ) + for (const [id, n] of Object.entries(nodes)) { + for (const child of n.onSuccess ?? []) { + rfE.push({ + id: `${id}-s->${child}`, + source: id, + sourceHandle: 'success', + target: child, + style: { stroke: '#10b981', strokeWidth: 2 }, + }) + } + for (const child of n.onError ?? []) { + rfE.push({ + id: `${id}-e->${child}`, + source: id, + sourceHandle: 'error', + target: child, + style: { stroke: '#ef4444', strokeWidth: 2, strokeDasharray: '5 3' }, + }) + } + } + return { rfNodes: rfN, rfEdges: rfE } + }, [roots, nodes, readOnly, selectedId]) + + const [flowNodes, setFlowNodes, onNodesChange] = useNodesState(rfNodes) + const [flowEdges, setFlowEdges, onEdgesChange] = useEdgesState(rfEdges) + + useEffect(() => { + setFlowNodes(rfNodes) + setFlowEdges(rfEdges) + }, [rfNodes, rfEdges, setFlowNodes, setFlowEdges]) + + const handleNodesChange = useCallback( + (changes: NodeChange[]) => { + for (const c of changes) { + if (c.type === 'position' && c.position) layoutRef.current[c.id] = c.position + if (c.type === 'select') { + if (c.selected && c.id !== TRIGGER_ID) setSelectedId(c.id) + else if (!c.selected && selectedId === c.id) setSelectedId(null) + } + } + onNodesChange(changes) + }, + [onNodesChange, selectedId], + ) + + const handleEdgesChange = useCallback( + (changes: EdgeChange[]) => { + let modelDirty = false + const nextRoots = [...roots] + const nextNodes: Record = Object.fromEntries( + Object.entries(nodes).map(([id, n]) => [id, { ...n, onSuccess: [...(n.onSuccess ?? [])], onError: [...(n.onError ?? [])] }]), + ) + for (const c of changes) { + if (c.type !== 'remove') continue + const edge = flowEdges.find((e) => e.id === c.id) + if (!edge) continue + if (edge.source === TRIGGER_ID) { + const i = nextRoots.indexOf(edge.target) + if (i >= 0) { + nextRoots.splice(i, 1) + modelDirty = true + } + continue + } + const src = nextNodes[edge.source] + if (!src) continue + if (edge.sourceHandle === 'success') { + src.onSuccess = src.onSuccess?.filter((t) => t !== edge.target) + modelDirty = true + } else if (edge.sourceHandle === 'error') { + src.onError = src.onError?.filter((t) => t !== edge.target) + modelDirty = true + } + } + if (modelDirty) onChange({ roots: nextRoots, nodes: nextNodes }) + onEdgesChange(changes) + }, + [flowEdges, roots, nodes, onChange, onEdgesChange], + ) + + const onConnect = useCallback( + (conn: Connection) => { + if (!conn.source || !conn.target) return + if (conn.source === conn.target) return + const nextRoots = [...roots] + const nextNodes: Record = Object.fromEntries( + Object.entries(nodes).map(([id, n]) => [id, { ...n, onSuccess: [...(n.onSuccess ?? [])], onError: [...(n.onError ?? [])] }]), + ) + if (conn.source === TRIGGER_ID) { + if (!nextRoots.includes(conn.target)) nextRoots.push(conn.target) + } else { + const src = nextNodes[conn.source] + if (!src) return + const list = conn.sourceHandle === 'error' ? (src.onError ??= []) : (src.onSuccess ??= []) + if (!list.includes(conn.target)) list.push(conn.target) + } + onChange({ roots: nextRoots, nodes: nextNodes }) + setFlowEdges((es) => addEdge(conn, es)) + }, + [roots, nodes, onChange, setFlowEdges], + ) + + const onDragOver = useCallback((event: React.DragEvent) => { + event.preventDefault() + event.dataTransfer.dropEffect = 'move' + }, []) + + const onDrop = useCallback( + (event: React.DragEvent) => { + event.preventDefault() + if (readOnly) return + const raw = event.dataTransfer.getData('application/soar-node') + if (!raw) return + let payload: { executor: string; kind: NodeKind; paramsPlaceholder?: unknown } + try { + payload = JSON.parse(raw) + } catch { + return + } + const position = screenToFlowPosition({ x: event.clientX, y: event.clientY }) + const id = uniqueId(payload.executor, nodes) + const newNode: FlowNode = { + kind: payload.kind, + executor: payload.executor, + params: payload.paramsPlaceholder, + } + layoutRef.current[id] = position + onChange({ roots, nodes: { ...nodes, [id]: newNode } }) + setSelectedId(id) + }, + [readOnly, screenToFlowPosition, roots, nodes, onChange], + ) + + const selected = selectedId ? nodes[selectedId] : null + + const renameNode = (nextId: string) => { + if (!selectedId || nextId === selectedId || nodes[nextId]) return + const nextNodes: Record = {} + for (const [id, n] of Object.entries(nodes)) { + const copy: FlowNode = { + ...n, + onSuccess: n.onSuccess?.map((t) => (t === selectedId ? nextId : t)), + onError: n.onError?.map((t) => (t === selectedId ? nextId : t)), + } + nextNodes[id === selectedId ? nextId : id] = copy + } + const nextRoots = roots.map((r) => (r === selectedId ? nextId : r)) + layoutRef.current[nextId] = layoutRef.current[selectedId] + delete layoutRef.current[selectedId] + onChange({ roots: nextRoots, nodes: nextNodes }) + setSelectedId(nextId) + } + + const patchNode = (patch: Partial) => { + if (!selectedId || !nodes[selectedId]) return + onChange({ roots, nodes: { ...nodes, [selectedId]: { ...nodes[selectedId], ...patch } } }) + } + + const deleteNode = () => { + if (!selectedId) return + const nextNodes: Record = {} + for (const [id, n] of Object.entries(nodes)) { + if (id === selectedId) continue + nextNodes[id] = { + ...n, + onSuccess: n.onSuccess?.filter((t) => t !== selectedId), + onError: n.onError?.filter((t) => t !== selectedId), + } + } + const nextRoots = roots.filter((r) => r !== selectedId) + delete layoutRef.current[selectedId] + onChange({ roots: nextRoots, nodes: nextNodes }) + setSelectedId(null) + } + + return ( +
+ {paletteOpen ? ( +
+ + +
+ ) : ( + setPaletteOpen(true)} /> + )} + +
+ setSelectedId(null)} + nodesDraggable={!readOnly} + nodesConnectable={!readOnly} + edgesFocusable={!readOnly} + fitView + fitViewOptions={{ padding: 0.2 }} + proOptions={{ hideAttribution: true }} + > + + + + +
+ + {selected && selectedId ? ( + inspectorOpen ? ( +
+ + +
+ ) : ( + setInspectorOpen(true)} /> + ) + ) : null} +
+ ) +} + +function CollapsedRail({ side, label, onClick }: { side: 'left' | 'right'; label: string; onClick: () => void }) { + const Icon = side === 'left' ? PanelLeft : PanelRight + return ( + + ) +} + +function uniqueId(executor: string, existing: Record): string { + let i = 1 + let id = executor + while (existing[id]) { + id = `${executor}_${i++}` + } + return id +} diff --git a/frontend/src/features/soar/components/FlowEditor.tsx b/frontend/src/features/soar/components/FlowEditor.tsx index 83ebfe271..096e24e74 100644 --- a/frontend/src/features/soar/components/FlowEditor.tsx +++ b/frontend/src/features/soar/components/FlowEditor.tsx @@ -1,29 +1,27 @@ -import { useEffect, useMemo, useRef, useState } from 'react' +import { useState } from 'react' import { useTranslation } from 'react-i18next' -import { ArrowDown, Check, ChevronDown, Code2, LayoutList, Loader2, Lock, Play, Plus, Server, Trash2, X, Zap } from 'lucide-react' +import { Code2, LayoutList, Loader2, Lock, Play, Plus, Trash2, X, Zap } from 'lucide-react' import { toast } from 'sonner' import { cn } from '@/shared/lib/utils' import { Button } from '@/shared/components/ui/button' import { Input } from '@/shared/components/ui/input' import { YamlCodeEditor } from '@/shared/components/YamlCodeEditor' import { PlatformBroadcastButton, broadcast, BULK_PATHS } from '@/features/platform-broadcast' -import { datasourcesHttpService } from '@/features/datasources/services/datasources-http.service' -import { VariablesManager } from '@/features/datasources/components/VariablesManager' import { soarFlowsService, SoarHttpError } from '../services/soar-flows.service' import { flowToForm, formToInput, flowFormToYaml, yamlToFlowForm, type FlowFormState } from '../lib/flow-yaml' -import { ALERT_FIELDS, COMMON_PLATFORMS, defaultShellForPlatform, shellsForPlatform } from '../lib/alert-fields' -import { COMMAND_TEMPLATES, shellKindFor } from '../lib/command-templates' -import { SOAR_MULTI_VALUE_OPERATORS, SOAR_NO_VALUE_OPERATORS, SOAR_OPERATORS, type Flow, type FlowCommand, type FlowCondition, type SoarCondition, type SoarOperator } from '../types/soar.types' +import { ALERT_FIELDS } from '../lib/alert-fields' +import { + SOAR_MULTI_VALUE_OPERATORS, + SOAR_NO_VALUE_OPERATORS, + SOAR_OPERATORS, + type Flow, + type FlowCondition, + type SoarOperator, +} from '../types/soar.types' +import { FlowCanvas } from './FlowCanvas' const SELECT = 'h-8 rounded-md border border-input bg-background px-2 text-xs focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring' -interface AgentOption { - name: string - platform: string - /** Refuses commands, so a flow that targets it would never do anything. */ - noRemoteControl: boolean -} - export function FlowEditor({ flow, creating, @@ -43,55 +41,8 @@ export function FlowEditor({ const [busy, setBusy] = useState(false) const [confirmDelete, setConfirmDelete] = useState(false) - const [agents, setAgents] = useState([]) - const scrollRef = useRef(null) - const set = (k: K, v: FlowFormState[K]) => setForm((f) => ({ ...f, [k]: v })) - // Load agent datasources to populate the platform + agent selects. - useEffect(() => { - datasourcesHttpService - .list({ page: 1, size: 1000, kind: 'agent' }) - .then((r) => - setAgents( - (r.items ?? []).map((d) => ({ - name: d.name, - platform: typeof d.metadata?.osPlatform === 'string' ? d.metadata.osPlatform : '', - noRemoteControl: d.metadata?.noRemoteControl === 'true', - })), - ), - ) - .catch(() => setAgents([])) - }, []) - - // Platform options: common defaults + platforms seen on agents + current value. - const platforms = useMemo(() => { - const seen = new Set() - const out: string[] = [] - for (const p of [...COMMON_PLATFORMS, ...agents.map((a) => a.platform), form.agentPlatform]) { - const v = p.trim() - if (v && !seen.has(v.toLowerCase())) { - seen.add(v.toLowerCase()) - out.push(v) - } - } - return out - }, [agents, form.agentPlatform]) - - // Agents matching the selected platform (fall back to all when none match). - const platformAgents = useMemo(() => { - const p = form.agentPlatform.trim().toLowerCase() - if (!p) return agents - const m = agents.filter((a) => a.platform && (a.platform.toLowerCase().includes(p) || p.includes(a.platform.toLowerCase()))) - return m.length ? m : agents - }, [agents, form.agentPlatform]) - - const onPlatformChange = (v: string) => - setForm((f) => { - const shells = shellsForPlatform(v) - return { ...f, agentPlatform: v, shell: shells.includes(f.shell) ? f.shell : defaultShellForPlatform(v) } - }) - const toCode = () => { setYaml(flowFormToYaml(form)) setMode('code') @@ -102,7 +53,7 @@ export function FlowEditor({ toast.error(t('soar.editor.yamlError', { error: r.error })) return } - setForm({ ...r.form, active: form.active }) // active isn't in YAML + setForm({ ...r.form, active: form.active }) setMode('visual') } @@ -127,12 +78,8 @@ export function FlowEditor({ toast.error(t('soar.editor.conditionsRequired')) return } - if (input.commands.length === 0) { - toast.error(t('soar.editor.commandsRequired')) - return - } - if (!input.agentPlatform) { - toast.error(t('soar.editor.platformRequired')) + if (input.roots.length === 0 || Object.keys(input.nodes).length === 0) { + toast.error(t('soar.editor.nodesRequired', 'Add at least one node and connect it to the trigger.')) return } setBusy(true) @@ -192,49 +139,50 @@ export function FlowEditor({ return (
-
-
-

- {creating ? t('soar.editor.createTitle') : flow?.name} - {readOnly && ( - - {t('soar.system')} - - )} -

- {!creating &&

{flow?.relPath}

} -
-
-
- - -
- +
-
+ +
+ - {mode === 'code' ? ( -
- -
- ) : ( -
- {/* Flow identity */} -
+ {mode === 'code' ? ( +
+ +
+ ) : ( +
+ {/* Identity + flow-level knobs */} +
+
-