diff --git a/api/v1alpha1/taskspawner_types.go b/api/v1alpha1/taskspawner_types.go index 626fa1d17..c5d068f1e 100644 --- a/api/v1alpha1/taskspawner_types.go +++ b/api/v1alpha1/taskspawner_types.go @@ -916,6 +916,56 @@ type TaskTemplate struct { UpstreamRepo string `json:"upstreamRepo,omitempty"` } +// TerminalTaskPhase is a TaskPhase restricted to terminal values. +// +kubebuilder:validation:Enum=Succeeded;Failed +type TerminalTaskPhase string + +// NotificationHook defines an outbound notification destination triggered +// when a spawned Task reaches a terminal phase. +type NotificationHook struct { + // Name identifies this hook for logging and status reporting. + // +kubebuilder:validation:Required + // +kubebuilder:validation:MinLength=1 + Name string `json:"name"` + + // Phases specifies which terminal phases trigger this hook. + // Defaults to both Succeeded and Failed. + // +optional + Phases []TerminalTaskPhase `json:"phases,omitempty"` + + // Webhook sends an HTTP POST with task details to the given URL. + // +kubebuilder:validation:Required + Webhook WebhookNotification `json:"webhook"` +} + +// WebhookNotification configures an HTTP webhook notification. +// Note: The URL is stored in a Task annotation for dispatch. Do not embed +// secrets in the URL path; use SecretRef for authentication instead. +type WebhookNotification struct { + // URL is the webhook endpoint. Must use HTTPS. This value is stored + // in the spawned Task's annotations and is visible to anyone with + // read access to the Task resource. Do not use URLs with embedded + // tokens; use secretRef for authentication. + // +kubebuilder:validation:Required + // +kubebuilder:validation:Pattern="^https://[^@/]+(/.*)?$" + URL string `json:"url"` + + // SecretRef optionally references a Secret whose keys are sent as HTTP + // headers. For example, a Secret with keys "Authorization" and "X-Api-Key" + // will set both as request headers with their respective values. + // +optional + SecretRef *SecretReference `json:"secretRef,omitempty"` +} + +// OnCompletion configures outbound notifications when spawned Tasks +// reach terminal phases. +type OnCompletion struct { + // Hooks is a list of notification destinations. + // +kubebuilder:validation:MinItems=1 + // +kubebuilder:validation:MaxItems=8 + Hooks []NotificationHook `json:"hooks"` +} + // TaskSpawnerSpec defines the desired state of TaskSpawner. // +kubebuilder:validation:XValidation:rule="!(has(self.when.githubIssues) || has(self.when.githubPullRequests) || has(self.when.githubWebhook) || has(self.when.linearWebhook)) || has(self.taskTemplate.workspaceRef)",message="taskTemplate.workspaceRef is required when using githubIssues, githubPullRequests, githubWebhook, or linearWebhook source" type TaskSpawnerSpec struct { @@ -956,6 +1006,11 @@ type TaskSpawnerSpec struct { // +optional // +kubebuilder:validation:Minimum=0 MaxTotalTasks *int32 `json:"maxTotalTasks,omitempty"` + + // OnCompletion configures outbound notifications when spawned Tasks + // reach terminal phases (Succeeded or Failed). + // +optional + OnCompletion *OnCompletion `json:"onCompletion,omitempty"` } // TaskSpawnerStatus defines the observed state of TaskSpawner. diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index b6e0ed2db..6d700b1bc 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -728,6 +728,49 @@ func (in *MCPServerSpec) DeepCopy() *MCPServerSpec { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *NotificationHook) DeepCopyInto(out *NotificationHook) { + *out = *in + if in.Phases != nil { + in, out := &in.Phases, &out.Phases + *out = make([]TerminalTaskPhase, len(*in)) + copy(*out, *in) + } + in.Webhook.DeepCopyInto(&out.Webhook) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new NotificationHook. +func (in *NotificationHook) DeepCopy() *NotificationHook { + if in == nil { + return nil + } + out := new(NotificationHook) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OnCompletion) DeepCopyInto(out *OnCompletion) { + *out = *in + if in.Hooks != nil { + in, out := &in.Hooks, &out.Hooks + *out = make([]NotificationHook, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OnCompletion. +func (in *OnCompletion) DeepCopy() *OnCompletion { + if in == nil { + return nil + } + out := new(OnCompletion) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *PluginSpec) DeepCopyInto(out *PluginSpec) { *out = *in @@ -1120,6 +1163,11 @@ func (in *TaskSpawnerSpec) DeepCopyInto(out *TaskSpawnerSpec) { *out = new(int32) **out = **in } + if in.OnCompletion != nil { + in, out := &in.OnCompletion, &out.OnCompletion + *out = new(OnCompletion) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TaskSpawnerSpec. @@ -1326,6 +1374,26 @@ func (in *TaskTemplateMetadata) DeepCopy() *TaskTemplateMetadata { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *WebhookNotification) DeepCopyInto(out *WebhookNotification) { + *out = *in + if in.SecretRef != nil { + in, out := &in.SecretRef, &out.SecretRef + *out = new(SecretReference) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WebhookNotification. +func (in *WebhookNotification) DeepCopy() *WebhookNotification { + if in == nil { + return nil + } + out := new(WebhookNotification) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *When) DeepCopyInto(out *When) { *out = *in diff --git a/cmd/kelos-spawner/main.go b/cmd/kelos-spawner/main.go index 7efb31290..372292a7e 100644 --- a/cmd/kelos-spawner/main.go +++ b/cmd/kelos-spawner/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "flag" "fmt" "net/http" @@ -415,6 +416,15 @@ func runCycleWithSourceCore(ctx context.Context, cl client.Client, key types.Nam } } + // Stamp onCompletion hooks config into task annotation for the + // reporting loop to dispatch without looking up the TaskSpawner. + if hookJSON := onCompletionAnnotation(&ts); hookJSON != "" { + if task.Annotations == nil { + task.Annotations = make(map[string]string) + } + task.Annotations[reporting.AnnotationOnCompletion] = hookJSON + } + // Propagate upstream repo for fork workflows. Explicit template // value takes precedence; otherwise derive from the source repo // override (githubIssues.repo or githubPullRequests.repo). @@ -819,3 +829,41 @@ func parsePollInterval(s string) time.Duration { } return d } + +// onCompletionAnnotation returns the serialized onCompletion hooks config +// to stamp onto spawned Tasks, or empty string when not configured. +func onCompletionAnnotation(ts *kelosv1alpha1.TaskSpawner) string { + if ts.Spec.OnCompletion == nil || len(ts.Spec.OnCompletion.Hooks) == 0 { + return "" + } + data, err := json.Marshal(ts.Spec.OnCompletion.Hooks) + if err != nil { + ctrl.Log.WithName("spawner").Error(err, "Serializing onCompletion hooks") + return "" + } + return string(data) +} + +// runWebhookReportingCycle lists all Tasks owned by the given TaskSpawner and +// dispatches pending webhook notifications for those in terminal phases. +func runWebhookReportingCycle(ctx context.Context, cl client.Client, key types.NamespacedName, httpClient *http.Client) error { + var taskList kelosv1alpha1.TaskList + if err := cl.List(ctx, &taskList, client.InNamespace(key.Namespace), client.MatchingLabels{"kelos.dev/taskspawner": key.Name}); err != nil { + return fmt.Errorf("listing tasks for webhook reporting: %w", err) + } + + reporter := &reporting.WebhookReporter{ + Client: cl, + HTTPClient: httpClient, + SecretReader: &reporting.KubeSecretReader{ + Client: cl, + }, + } + + for i := range taskList.Items { + if err := reporter.ReportWebhooks(ctx, &taskList.Items[i]); err != nil { + ctrl.Log.WithName("spawner").Error(err, "Webhook reporting", "task", taskList.Items[i].Name) + } + } + return nil +} diff --git a/cmd/kelos-spawner/reconciler.go b/cmd/kelos-spawner/reconciler.go index 81b457f36..2393fed5d 100644 --- a/cmd/kelos-spawner/reconciler.go +++ b/cmd/kelos-spawner/reconciler.go @@ -69,6 +69,12 @@ func (r *spawnerReconciler) SetupWithManager(mgr ctrl.Manager) error { } func runOnce(ctx context.Context, cl client.Client, key types.NamespacedName, cfg spawnerRuntimeConfig) (time.Duration, error) { + // Run webhook reporting first — it operates on already-terminal tasks + // and must not be blocked by source-polling or GitHub reporting failures. + if err := runWebhookReportingCycle(ctx, cl, key, cfg.HTTPClient); err != nil { + ctrl.Log.WithName("spawner").Error(err, "Webhook reporting cycle failed") + } + if err := runCycleWithProxy(ctx, cl, key, cfg.GitHubOwner, cfg.GitHubRepo, cfg.GHProxyURL, cfg.GitHubAPIBaseURL, cfg.TokenResolver, cfg.JiraBaseURL, cfg.JiraProject, cfg.JiraJQL, cfg.HTTPClient); err != nil { return 0, err } diff --git a/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml b/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml index dc8c41e41..8f076c4b8 100644 --- a/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml +++ b/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml @@ -82,6 +82,72 @@ spec: format: int32 minimum: 0 type: integer + onCompletion: + description: |- + OnCompletion configures outbound notifications when spawned Tasks + reach terminal phases (Succeeded or Failed). + properties: + hooks: + description: Hooks is a list of notification destinations. + items: + description: |- + NotificationHook defines an outbound notification destination triggered + when a spawned Task reaches a terminal phase. + properties: + name: + description: Name identifies this hook for logging and status + reporting. + minLength: 1 + type: string + phases: + description: |- + Phases specifies which terminal phases trigger this hook. + Defaults to both Succeeded and Failed. + items: + description: TerminalTaskPhase is a TaskPhase restricted + to terminal values. + enum: + - Succeeded + - Failed + type: string + type: array + webhook: + description: Webhook sends an HTTP POST with task details + to the given URL. + properties: + secretRef: + description: |- + SecretRef optionally references a Secret whose keys are sent as HTTP + headers. For example, a Secret with keys "Authorization" and "X-Api-Key" + will set both as request headers with their respective values. + properties: + name: + description: Name is the name of the secret. + type: string + required: + - name + type: object + url: + description: |- + URL is the webhook endpoint. Must use HTTPS. This value is stored + in the spawned Task's annotations and is visible to anyone with + read access to the Task resource. Do not use URLs with embedded + tokens; use secretRef for authentication. + pattern: ^https://[^@/]+(/.*)?$ + type: string + required: + - url + type: object + required: + - name + - webhook + type: object + maxItems: 8 + minItems: 1 + type: array + required: + - hooks + type: object pollInterval: default: 5m description: |- diff --git a/internal/manifests/install-crd.yaml b/internal/manifests/install-crd.yaml index 46af34380..bb4d0e531 100644 --- a/internal/manifests/install-crd.yaml +++ b/internal/manifests/install-crd.yaml @@ -7310,6 +7310,72 @@ spec: format: int32 minimum: 0 type: integer + onCompletion: + description: |- + OnCompletion configures outbound notifications when spawned Tasks + reach terminal phases (Succeeded or Failed). + properties: + hooks: + description: Hooks is a list of notification destinations. + items: + description: |- + NotificationHook defines an outbound notification destination triggered + when a spawned Task reaches a terminal phase. + properties: + name: + description: Name identifies this hook for logging and status + reporting. + minLength: 1 + type: string + phases: + description: |- + Phases specifies which terminal phases trigger this hook. + Defaults to both Succeeded and Failed. + items: + description: TerminalTaskPhase is a TaskPhase restricted + to terminal values. + enum: + - Succeeded + - Failed + type: string + type: array + webhook: + description: Webhook sends an HTTP POST with task details + to the given URL. + properties: + secretRef: + description: |- + SecretRef optionally references a Secret whose keys are sent as HTTP + headers. For example, a Secret with keys "Authorization" and "X-Api-Key" + will set both as request headers with their respective values. + properties: + name: + description: Name is the name of the secret. + type: string + required: + - name + type: object + url: + description: |- + URL is the webhook endpoint. Must use HTTPS. This value is stored + in the spawned Task's annotations and is visible to anyone with + read access to the Task resource. Do not use URLs with embedded + tokens; use secretRef for authentication. + pattern: ^https://[^@/]+(/.*)?$ + type: string + required: + - url + type: object + required: + - name + - webhook + type: object + maxItems: 8 + minItems: 1 + type: array + required: + - hooks + type: object pollInterval: default: 5m description: |- diff --git a/internal/reporting/secret_reader.go b/internal/reporting/secret_reader.go new file mode 100644 index 000000000..613acbba1 --- /dev/null +++ b/internal/reporting/secret_reader.go @@ -0,0 +1,26 @@ +package reporting + +import ( + "context" + "fmt" + + corev1 "k8s.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// KubeSecretReader reads secrets from the Kubernetes API. +type KubeSecretReader struct { + Client client.Client +} + +func (r *KubeSecretReader) ReadHeaders(ctx context.Context, namespace, name string) (map[string]string, error) { + var secret corev1.Secret + if err := r.Client.Get(ctx, client.ObjectKey{Namespace: namespace, Name: name}, &secret); err != nil { + return nil, fmt.Errorf("getting secret %s/%s: %w", namespace, name, err) + } + headers := make(map[string]string, len(secret.Data)) + for k, v := range secret.Data { + headers[k] = string(v) + } + return headers, nil +} diff --git a/internal/reporting/webhook.go b/internal/reporting/webhook.go new file mode 100644 index 000000000..ebaeb7dc2 --- /dev/null +++ b/internal/reporting/webhook.go @@ -0,0 +1,342 @@ +package reporting + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/url" + "time" + + "k8s.io/client-go/util/retry" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + kelosv1alpha1 "github.com/kelos-dev/kelos/api/v1alpha1" +) + +const ( + // AnnotationOnCompletion stores the serialized onCompletion hooks config + // so the reporting loop can dispatch without looking up the TaskSpawner. + AnnotationOnCompletion = "kelos.dev/on-completion" + + // AnnotationWebhookReportPhase records the last Task phase that was + // reported via webhook hooks, preventing duplicate deliveries. + AnnotationWebhookReportPhase = "kelos.dev/webhook-report-phase" +) + +// WebhookPayload is the JSON body sent to onCompletion webhook endpoints. +type WebhookPayload struct { + Task string `json:"task"` + Namespace string `json:"namespace"` + Spawner string `json:"spawner,omitempty"` + Phase string `json:"phase"` + Message string `json:"message,omitempty"` + AgentType string `json:"agentType,omitempty"` + Model string `json:"model,omitempty"` + StartTime *time.Time `json:"startTime,omitempty"` + CompletionTime *time.Time `json:"completionTime,omitempty"` + Outputs []string `json:"outputs,omitempty"` + Results map[string]string `json:"results,omitempty"` +} + +// WebhookReporter dispatches onCompletion webhook notifications for Tasks. +type WebhookReporter struct { + Client client.Client + HTTPClient *http.Client + // SecretReader resolves secret values. When nil, secretRef is ignored. + SecretReader SecretReader + // skipURLValidation disables SSRF checks (for testing only). + skipURLValidation bool + // safeClient is lazily initialized by httpClient(). + safeClient *http.Client +} + +// SecretReader reads headers from a named Secret in a namespace. +type SecretReader interface { + // ReadHeaders returns all key-value pairs from the named Secret. + // Each key is used as an HTTP header name and the value as its value. + ReadHeaders(ctx context.Context, namespace, name string) (map[string]string, error) +} + +// ReportWebhooks checks whether the task has onCompletion hooks configured +// and dispatches webhooks for terminal phases that haven't been reported yet. +func (wr *WebhookReporter) ReportWebhooks(ctx context.Context, task *kelosv1alpha1.Task) error { + log := ctrl.Log.WithName("webhook-reporter") + + annotations := task.Annotations + if annotations == nil { + return nil + } + + hooksJSON := annotations[AnnotationOnCompletion] + if hooksJSON == "" { + return nil + } + + // Only fire for terminal phases. + if task.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && task.Status.Phase != kelosv1alpha1.TaskPhaseFailed { + return nil + } + + // Skip if already reported this phase. + if annotations[AnnotationWebhookReportPhase] == string(task.Status.Phase) { + return nil + } + + var hooks []kelosv1alpha1.NotificationHook + if err := json.Unmarshal([]byte(hooksJSON), &hooks); err != nil { + return fmt.Errorf("parsing on-completion hooks annotation: %w", err) + } + + payload := buildWebhookPayload(task) + + var lastErr error + dispatched := 0 + for _, hook := range hooks { + if !phaseMatches(hook.Phases, task.Status.Phase) { + continue + } + + dispatched++ + if err := wr.sendWebhook(ctx, task.Namespace, hook, payload); err != nil { + log.Error(err, "Sending webhook", "task", task.Name, "hook", hook.Name) + lastErr = err + continue + } + log.Info("Sent webhook notification", "task", task.Name, "hook", hook.Name, "phase", task.Status.Phase) + } + + // Persist the annotation to prevent re-evaluation on future cycles. + // This is done even on partial failure to avoid duplicate deliveries + // to hooks that already succeeded. + if err := wr.persistWebhookReportPhase(ctx, task, string(task.Status.Phase)); err != nil { + return err + } + + return lastErr +} + +func (wr *WebhookReporter) sendWebhook(ctx context.Context, namespace string, hook kelosv1alpha1.NotificationHook, payload WebhookPayload) error { + if !wr.skipURLValidation { + if err := validateWebhookURL(hook.Webhook.URL); err != nil { + return err + } + } + + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshaling webhook payload: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, hook.Webhook.URL, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("creating webhook request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + if hook.Webhook.SecretRef != nil && wr.SecretReader != nil { + headers, err := wr.SecretReader.ReadHeaders(ctx, namespace, hook.Webhook.SecretRef.Name) + if err != nil { + return fmt.Errorf("reading webhook secret %q: %w", hook.Webhook.SecretRef.Name, err) + } + for k, v := range headers { + req.Header.Set(k, v) + } + } + + httpClient := wr.httpClient() + + resp, err := httpClient.Do(req) + if err != nil { + return fmt.Errorf("sending webhook to hook %q: %w", hook.Name, err) + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + + if resp.StatusCode >= 300 { + return fmt.Errorf("webhook %q returned status %d", hook.Name, resp.StatusCode) + } + + return nil +} + +// httpClient returns an HTTP client with SSRF-safe transport and redirect +// policy. The result is cached after first construction. +func (wr *WebhookReporter) httpClient() *http.Client { + if wr.skipURLValidation && wr.HTTPClient != nil { + return wr.HTTPClient + } + if wr.safeClient != nil { + return wr.safeClient + } + if wr.HTTPClient == nil { + wr.safeClient = defaultWebhookHTTPClient + } else { + clone := *wr.HTTPClient + clone.CheckRedirect = ssrfCheckRedirect + clone.Transport = ssrfSafeTransport(clone.Transport) + wr.safeClient = &clone + } + return wr.safeClient +} + +var defaultWebhookHTTPClient = &http.Client{ + Timeout: 10 * time.Second, + Transport: ssrfSafeTransport(nil), + CheckRedirect: ssrfCheckRedirect, +} + +func ssrfSafeTransport(base http.RoundTripper) http.RoundTripper { + if base == nil { + base = http.DefaultTransport + } + t, ok := base.(*http.Transport) + if !ok { + ctrl.Log.WithName("webhook-reporter").Info("Custom transport is not *http.Transport, using default transport for SSRF protection") + t = http.DefaultTransport.(*http.Transport) + } + clone := t.Clone() + clone.DialContext = ssrfDialContext + return clone +} + +func ssrfDialContext(ctx context.Context, network, addr string) (net.Conn, error) { + host, port, err := net.SplitHostPort(addr) + if err != nil { + return nil, err + } + ips, err := net.DefaultResolver.LookupIPAddr(ctx, host) + if err != nil { + return nil, err + } + if len(ips) == 0 { + return nil, fmt.Errorf("no addresses found for host %q", host) + } + for _, ip := range ips { + if isPrivateIP(ip.IP) { + return nil, fmt.Errorf("webhook URL must not target private/internal addresses") + } + } + var dialer net.Dialer + return dialer.DialContext(ctx, network, net.JoinHostPort(ips[0].IP.String(), port)) +} + +func ssrfCheckRedirect(req *http.Request, via []*http.Request) error { + if err := validateWebhookURL(req.URL.String()); err != nil { + return err + } + if len(via) >= 10 { + return fmt.Errorf("too many redirects") + } + return nil +} + +// validateWebhookURL rejects URLs that are not HTTPS or contain embedded +// credentials (userinfo). IP-level SSRF checks are enforced at dial time +// via the transport's DialContext. +func validateWebhookURL(rawURL string) error { + u, err := url.Parse(rawURL) + if err != nil { + return fmt.Errorf("invalid webhook URL: %w", err) + } + if u.Scheme != "https" { + return fmt.Errorf("webhook URL must use HTTPS") + } + if u.User != nil { + return fmt.Errorf("webhook URL must not contain embedded credentials") + } + return nil +} + +var privateRanges = []net.IPNet{ + {IP: net.IPv4(0, 0, 0, 0), Mask: net.CIDRMask(8, 32)}, + {IP: net.IPv4(10, 0, 0, 0), Mask: net.CIDRMask(8, 32)}, + {IP: net.IPv4(127, 0, 0, 0), Mask: net.CIDRMask(8, 32)}, + {IP: net.IPv4(169, 254, 0, 0), Mask: net.CIDRMask(16, 32)}, + {IP: net.IPv4(172, 16, 0, 0), Mask: net.CIDRMask(12, 32)}, + {IP: net.IPv4(192, 168, 0, 0), Mask: net.CIDRMask(16, 32)}, + {IP: net.ParseIP("::1"), Mask: net.CIDRMask(128, 128)}, + {IP: net.ParseIP("fe80::"), Mask: net.CIDRMask(10, 128)}, + {IP: net.ParseIP("fc00::"), Mask: net.CIDRMask(7, 128)}, +} + +func isPrivateIP(ip net.IP) bool { + for _, r := range privateRanges { + if r.Contains(ip) { + return true + } + } + return false +} + +func (wr *WebhookReporter) persistWebhookReportPhase(ctx context.Context, task *kelosv1alpha1.Task, phase string) error { + return persistAnnotationRetry(ctx, wr.Client, task, map[string]string{ + AnnotationWebhookReportPhase: phase, + }) +} + +func buildWebhookPayload(task *kelosv1alpha1.Task) WebhookPayload { + p := WebhookPayload{ + Task: task.Name, + Namespace: task.Namespace, + Spawner: task.Labels["kelos.dev/taskspawner"], + Phase: string(task.Status.Phase), + Message: task.Status.Message, + AgentType: task.Spec.Type, + Model: task.Spec.Model, + Outputs: task.Status.Outputs, + Results: task.Status.Results, + } + if task.Status.StartTime != nil { + t := task.Status.StartTime.Time + p.StartTime = &t + } + if task.Status.CompletionTime != nil { + t := task.Status.CompletionTime.Time + p.CompletionTime = &t + } + return p +} + +func phaseMatches(configured []kelosv1alpha1.TerminalTaskPhase, actual kelosv1alpha1.TaskPhase) bool { + if len(configured) == 0 { + return true + } + for _, p := range configured { + if kelosv1alpha1.TaskPhase(p) == actual { + return true + } + } + return false +} + +// persistAnnotationRetry updates annotations on a Task using a merge patch +// with retry on conflict, avoiding full-object writes that could clobber +// concurrent changes from other controllers. +func persistAnnotationRetry(ctx context.Context, cl client.Client, task *kelosv1alpha1.Task, annotations map[string]string) error { + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + var current kelosv1alpha1.Task + if err := cl.Get(ctx, client.ObjectKeyFromObject(task), ¤t); err != nil { + return err + } + base := current.DeepCopy() + if current.Annotations == nil { + current.Annotations = make(map[string]string) + } + for k, v := range annotations { + current.Annotations[k] = v + } + if err := cl.Patch(ctx, ¤t, client.MergeFrom(base)); err != nil { + return err + } + task.Annotations = current.Annotations + return nil + }); err != nil { + return fmt.Errorf("persisting webhook annotations on task %s: %w", task.Name, err) + } + return nil +} diff --git a/internal/reporting/webhook_test.go b/internal/reporting/webhook_test.go new file mode 100644 index 000000000..5278ddbca --- /dev/null +++ b/internal/reporting/webhook_test.go @@ -0,0 +1,502 @@ +package reporting + +import ( + "context" + "encoding/json" + "io" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + kelosv1alpha1 "github.com/kelos-dev/kelos/api/v1alpha1" +) + +type fakeSecretReader struct { + headers map[string]map[string]string // namespace/name -> headers +} + +func (f *fakeSecretReader) ReadHeaders(_ context.Context, namespace, name string) (map[string]string, error) { + return f.headers[namespace+"/"+name], nil +} + +func newFakeClient(objs ...client.Object) client.Client { + scheme := runtime.NewScheme() + utilruntime.Must(kelosv1alpha1.AddToScheme(scheme)) + return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).WithStatusSubresource(&kelosv1alpha1.Task{}).Build() +} + +func TestWebhookReporter_ReportWebhooks(t *testing.T) { + tests := []struct { + name string + task *kelosv1alpha1.Task + serverStatus int + wantRequests int + wantPayload *WebhookPayload + wantAuthHeader string + wantHeaders map[string]string + wantAnnotation string + wantErr bool + }{ + { + name: "sends webhook on task succeeded", + task: &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-task-1", + Namespace: "default", + Labels: map[string]string{"kelos.dev/taskspawner": "my-spawner"}, + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"slack-alert","webhook":{"url":"PLACEHOLDER"}}]`, + }, + }, + Spec: kelosv1alpha1.TaskSpec{ + Type: "claude-code", + Model: "claude-sonnet-4-20250514", + }, + Status: kelosv1alpha1.TaskStatus{ + Phase: kelosv1alpha1.TaskPhaseSucceeded, + Message: "Task completed successfully", + Outputs: []string{"https://github.com/org/repo/pull/1"}, + Results: map[string]string{"cost-usd": "0.42"}, + }, + }, + serverStatus: 200, + wantRequests: 1, + wantPayload: &WebhookPayload{ + Task: "test-task-1", + Namespace: "default", + Spawner: "my-spawner", + Phase: "Succeeded", + Message: "Task completed successfully", + AgentType: "claude-code", + Model: "claude-sonnet-4-20250514", + Outputs: []string{"https://github.com/org/repo/pull/1"}, + Results: map[string]string{"cost-usd": "0.42"}, + }, + wantAnnotation: "Succeeded", + }, + { + name: "sends webhook on task failed", + task: &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-task-2", + Namespace: "default", + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"alert","webhook":{"url":"PLACEHOLDER"}}]`, + }, + }, + Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, + Status: kelosv1alpha1.TaskStatus{ + Phase: kelosv1alpha1.TaskPhaseFailed, + Message: "OOM killed", + }, + }, + serverStatus: 200, + wantRequests: 1, + wantAnnotation: "Failed", + }, + { + name: "skips non-terminal phase", + task: &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-task-3", + Namespace: "default", + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"alert","webhook":{"url":"PLACEHOLDER"}}]`, + }, + }, + Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, + Status: kelosv1alpha1.TaskStatus{Phase: kelosv1alpha1.TaskPhaseRunning}, + }, + serverStatus: 200, + wantRequests: 0, + }, + { + name: "skips already reported phase", + task: &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-task-4", + Namespace: "default", + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"alert","webhook":{"url":"PLACEHOLDER"}}]`, + AnnotationWebhookReportPhase: "Succeeded", + }, + }, + Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, + Status: kelosv1alpha1.TaskStatus{Phase: kelosv1alpha1.TaskPhaseSucceeded}, + }, + serverStatus: 200, + wantRequests: 0, + }, + { + name: "filters by phase - only Failed configured but task Succeeded", + task: &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-task-5", + Namespace: "default", + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"alert","phases":["Failed"],"webhook":{"url":"PLACEHOLDER"}}]`, + }, + }, + Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, + Status: kelosv1alpha1.TaskStatus{Phase: kelosv1alpha1.TaskPhaseSucceeded}, + }, + serverStatus: 200, + wantRequests: 0, + wantAnnotation: "Succeeded", + }, + { + name: "persists annotation on delivery failure to prevent duplicates", + task: &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-task-fail-delivery", + Namespace: "default", + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"alert","webhook":{"url":"PLACEHOLDER"}}]`, + }, + }, + Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, + Status: kelosv1alpha1.TaskStatus{Phase: kelosv1alpha1.TaskPhaseSucceeded}, + }, + serverStatus: 500, + wantRequests: 1, + wantErr: true, + wantAnnotation: "Succeeded", + }, + { + name: "includes auth header from secret", + task: &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-task-6", + Namespace: "default", + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"alert","webhook":{"url":"PLACEHOLDER","secretRef":{"name":"webhook-secret"}}}]`, + }, + }, + Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, + Status: kelosv1alpha1.TaskStatus{Phase: kelosv1alpha1.TaskPhaseSucceeded}, + }, + serverStatus: 200, + wantRequests: 1, + wantAuthHeader: "Bearer my-token", + wantAnnotation: "Succeeded", + }, + { + name: "sets multiple headers from secret", + task: &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-task-7", + Namespace: "default", + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"alert","webhook":{"url":"PLACEHOLDER","secretRef":{"name":"multi-header-secret"}}}]`, + }, + }, + Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, + Status: kelosv1alpha1.TaskStatus{Phase: kelosv1alpha1.TaskPhaseSucceeded}, + }, + serverStatus: 200, + wantRequests: 1, + wantHeaders: map[string]string{ + "Authorization": "Bearer multi-token", + "X-Api-Key": "key-123", + }, + wantAnnotation: "Succeeded", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + requestCount := 0 + var lastBody []byte + var lastAuthHeader string + var lastHeaders http.Header + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestCount++ + lastAuthHeader = r.Header.Get("Authorization") + lastHeaders = r.Header.Clone() + lastBody, _ = io.ReadAll(r.Body) + w.WriteHeader(tt.serverStatus) + })) + defer server.Close() + + // Replace PLACEHOLDER URL in annotations with actual test server URL. + if ann := tt.task.Annotations[AnnotationOnCompletion]; ann != "" { + var hooks []json.RawMessage + json.Unmarshal([]byte(ann), &hooks) + for i := range hooks { + var h map[string]interface{} + json.Unmarshal(hooks[i], &h) + if wh, ok := h["webhook"].(map[string]interface{}); ok { + wh["url"] = server.URL + } + hooks[i], _ = json.Marshal(h) + } + updated, _ := json.Marshal(hooks) + tt.task.Annotations[AnnotationOnCompletion] = string(updated) + } + + cl := newFakeClient(tt.task) + + secretReader := &fakeSecretReader{ + headers: map[string]map[string]string{ + "default/webhook-secret": { + "Authorization": "Bearer my-token", + }, + "default/multi-header-secret": { + "Authorization": "Bearer multi-token", + "X-Api-Key": "key-123", + }, + }, + } + + wr := &WebhookReporter{ + Client: cl, + HTTPClient: server.Client(), + SecretReader: secretReader, + skipURLValidation: true, + } + + err := wr.ReportWebhooks(context.Background(), tt.task) + if tt.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + } else if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if requestCount != tt.wantRequests { + t.Errorf("expected %d requests, got %d", tt.wantRequests, requestCount) + } + + if tt.wantPayload != nil && lastBody != nil { + var got WebhookPayload + if err := json.Unmarshal(lastBody, &got); err != nil { + t.Fatalf("unmarshal payload: %v", err) + } + if got.Task != tt.wantPayload.Task { + t.Errorf("payload.Task = %q, want %q", got.Task, tt.wantPayload.Task) + } + if got.Phase != tt.wantPayload.Phase { + t.Errorf("payload.Phase = %q, want %q", got.Phase, tt.wantPayload.Phase) + } + if got.Spawner != tt.wantPayload.Spawner { + t.Errorf("payload.Spawner = %q, want %q", got.Spawner, tt.wantPayload.Spawner) + } + if got.AgentType != tt.wantPayload.AgentType { + t.Errorf("payload.AgentType = %q, want %q", got.AgentType, tt.wantPayload.AgentType) + } + } + + if tt.wantAuthHeader != "" && lastAuthHeader != tt.wantAuthHeader { + t.Errorf("Authorization header = %q, want %q", lastAuthHeader, tt.wantAuthHeader) + } + + for k, v := range tt.wantHeaders { + if got := lastHeaders.Get(k); got != v { + t.Errorf("header %s = %q, want %q", k, got, v) + } + } + + // Verify annotation persistence for cases that dispatched webhooks. + if tt.wantAnnotation != "" { + var updated kelosv1alpha1.Task + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(tt.task), &updated); err != nil { + t.Fatalf("fetching task: %v", err) + } + got := updated.Annotations[AnnotationWebhookReportPhase] + if got != tt.wantAnnotation { + t.Errorf("annotation %s = %q, want %q", AnnotationWebhookReportPhase, got, tt.wantAnnotation) + } + } + + }) + } +} + +func TestReportWebhooks_Idempotency(t *testing.T) { + requestCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + requestCount++ + w.WriteHeader(200) + })) + defer server.Close() + + task := &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "idem-task", + Namespace: "default", + Annotations: map[string]string{ + AnnotationOnCompletion: `[{"name":"hook","webhook":{"url":"` + server.URL + `"}}]`, + }, + }, + Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, + Status: kelosv1alpha1.TaskStatus{Phase: kelosv1alpha1.TaskPhaseSucceeded}, + } + + cl := newFakeClient(task) + wr := &WebhookReporter{Client: cl, HTTPClient: server.Client(), skipURLValidation: true} + + // First call should dispatch and persist. + if err := wr.ReportWebhooks(context.Background(), task); err != nil { + t.Fatalf("first call: %v", err) + } + if requestCount != 1 { + t.Fatalf("expected 1 request on first call, got %d", requestCount) + } + + // Re-read task with updated annotations. + var updated kelosv1alpha1.Task + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(task), &updated); err != nil { + t.Fatalf("fetching task: %v", err) + } + + // Second call should be a no-op (annotation already set) — same server, + // so any duplicate delivery would increment requestCount. + if err := wr.ReportWebhooks(context.Background(), &updated); err != nil { + t.Fatalf("second call: %v", err) + } + if requestCount != 1 { + t.Errorf("expected no additional requests on second call, got %d total", requestCount) + } +} + +func TestBuildWebhookPayload(t *testing.T) { + startTime := metav1.NewTime(time.Date(2026, 3, 20, 10, 0, 0, 0, time.UTC)) + completionTime := metav1.NewTime(time.Date(2026, 3, 20, 10, 5, 30, 0, time.UTC)) + + task := &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-task", + Namespace: "prod", + Labels: map[string]string{"kelos.dev/taskspawner": "my-spawner"}, + }, + Spec: kelosv1alpha1.TaskSpec{ + Type: "claude-code", + Model: "claude-sonnet-4-20250514", + }, + Status: kelosv1alpha1.TaskStatus{ + Phase: kelosv1alpha1.TaskPhaseSucceeded, + Message: "Task completed successfully", + StartTime: &startTime, + CompletionTime: &completionTime, + Outputs: []string{"https://github.com/org/repo/pull/123"}, + Results: map[string]string{"cost-usd": "0.42", "input-tokens": "15000"}, + }, + } + + payload := buildWebhookPayload(task) + + if payload.Task != "my-task" { + t.Errorf("Task = %q, want %q", payload.Task, "my-task") + } + if payload.Namespace != "prod" { + t.Errorf("Namespace = %q, want %q", payload.Namespace, "prod") + } + if payload.Spawner != "my-spawner" { + t.Errorf("Spawner = %q, want %q", payload.Spawner, "my-spawner") + } + if payload.Phase != "Succeeded" { + t.Errorf("Phase = %q, want %q", payload.Phase, "Succeeded") + } + if payload.StartTime == nil || !payload.StartTime.Equal(startTime.Time) { + t.Errorf("StartTime = %v, want %v", payload.StartTime, startTime.Time) + } + if payload.CompletionTime == nil || !payload.CompletionTime.Equal(completionTime.Time) { + t.Errorf("CompletionTime = %v, want %v", payload.CompletionTime, completionTime.Time) + } + if len(payload.Outputs) != 1 || payload.Outputs[0] != "https://github.com/org/repo/pull/123" { + t.Errorf("Outputs = %v, want [https://github.com/org/repo/pull/123]", payload.Outputs) + } + if payload.Results["cost-usd"] != "0.42" { + t.Errorf("Results[cost-usd] = %q, want %q", payload.Results["cost-usd"], "0.42") + } +} + +func TestHttpClient_AppliesSSRFTransport(t *testing.T) { + injected := &http.Client{Timeout: 5 * time.Second} + wr := &WebhookReporter{HTTPClient: injected} + cl := wr.httpClient() + if cl.CheckRedirect == nil { + t.Fatal("expected CheckRedirect to be set on cloned client") + } + if cl.Transport == nil { + t.Fatal("expected Transport to be set on cloned client") + } + if cl == injected { + t.Fatal("expected a clone, not the original client") + } +} + +func TestValidateWebhookURL(t *testing.T) { + tests := []struct { + url string + wantErr bool + }{ + {"https://hooks.slack.com/services/T00/B00/xxx", false}, + {"https://example.com/webhook", false}, + {"http://example.com/webhook", true}, + {"https://user:pass@example.com/webhook", true}, + {"https://token@example.com/webhook", true}, + } + for _, tt := range tests { + err := validateWebhookURL(tt.url) + if (err != nil) != tt.wantErr { + t.Errorf("validateWebhookURL(%q) error = %v, wantErr %v", tt.url, err, tt.wantErr) + } + } +} + +func TestIsPrivateIP(t *testing.T) { + tests := []struct { + ip string + want bool + }{ + {"0.0.0.0", true}, + {"10.0.0.1", true}, + {"172.16.0.1", true}, + {"192.168.1.1", true}, + {"169.254.169.254", true}, + {"127.0.0.1", true}, + {"::1", true}, + {"fe80::1", true}, + {"fd00::1", true}, + {"8.8.8.8", false}, + {"2001:4860:4860::8888", false}, + } + for _, tt := range tests { + ip := net.ParseIP(tt.ip) + if got := isPrivateIP(ip); got != tt.want { + t.Errorf("isPrivateIP(%s) = %v, want %v", tt.ip, got, tt.want) + } + } +} + +func TestPhaseMatches(t *testing.T) { + tests := []struct { + configured []kelosv1alpha1.TerminalTaskPhase + actual kelosv1alpha1.TaskPhase + want bool + }{ + {nil, kelosv1alpha1.TaskPhaseSucceeded, true}, + {nil, kelosv1alpha1.TaskPhaseFailed, true}, + {[]kelosv1alpha1.TerminalTaskPhase{"Failed"}, kelosv1alpha1.TaskPhaseFailed, true}, + {[]kelosv1alpha1.TerminalTaskPhase{"Failed"}, kelosv1alpha1.TaskPhaseSucceeded, false}, + {[]kelosv1alpha1.TerminalTaskPhase{"Succeeded", "Failed"}, kelosv1alpha1.TaskPhaseSucceeded, true}, + } + + for _, tt := range tests { + got := phaseMatches(tt.configured, tt.actual) + if got != tt.want { + t.Errorf("phaseMatches(%v, %v) = %v, want %v", tt.configured, tt.actual, got, tt.want) + } + } +}