From e825be5a744baad3f6545252a190da75a2d91d18 Mon Sep 17 00:00:00 2001 From: Irina Truong Date: Fri, 29 May 2026 15:52:50 -0700 Subject: [PATCH 1/3] feat: add onCompletion webhook hooks to TaskSpawner Add a new `spec.onCompletion` field to TaskSpawnerSpec that configures outbound HTTP webhook notifications when spawned Tasks reach terminal phases (Succeeded or Failed). Key design: - Hooks config is serialized into a Task annotation at spawn time so the reporting loop can dispatch without looking up the TaskSpawner - SSRF protection: HTTPS-only, runtime IP blocklist (IPv4/IPv6 private ranges), DNS resolution of hostnames, and redirect guard - Idempotency via annotation with conflict-retrying merge patch - All keys in the referenced Secret are sent as HTTP headers - Phase filtering defaults to both Succeeded and Failed Fixes #749 Co-Authored-By: Claude Opus 4.6 --- api/v1alpha1/taskspawner_types.go | 55 ++ api/v1alpha1/zz_generated.deepcopy.go | 68 +++ cmd/kelos-spawner/main.go | 39 ++ cmd/kelos-spawner/reconciler.go | 7 + .../kelos/templates/crds/taskspawner-crd.yaml | 66 +++ internal/manifests/install-crd.yaml | 66 +++ internal/reporting/secret_reader.go | 26 + internal/reporting/webhook.go | 314 +++++++++++ internal/reporting/webhook_test.go | 502 ++++++++++++++++++ 9 files changed, 1143 insertions(+) create mode 100644 internal/reporting/secret_reader.go create mode 100644 internal/reporting/webhook.go create mode 100644 internal/reporting/webhook_test.go diff --git a/api/v1alpha1/taskspawner_types.go b/api/v1alpha1/taskspawner_types.go index 626fa1d17..50554ef12 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..c0c60e134 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" @@ -819,3 +820,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..0a9145dc7 100644 --- a/cmd/kelos-spawner/reconciler.go +++ b/cmd/kelos-spawner/reconciler.go @@ -116,6 +116,13 @@ func runOnce(ctx context.Context, cl client.Client, key types.NamespacedName, cf } } + // Run onCompletion webhook reporting unconditionally — tasks carry their + // hook config in annotations, so webhooks must fire even if the spawner + // spec was updated after task creation. + if err := runWebhookReportingCycle(ctx, cl, key, cfg.HTTPClient); err != nil { + ctrl.Log.WithName("spawner").Error(err, "Webhook reporting cycle failed") + } + return resolvedPollInterval(&ts), nil } diff --git a/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml b/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml index dc8c41e41..9156d628d 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..459f20bcb 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..7ec770fac --- /dev/null +++ b/internal/reporting/webhook.go @@ -0,0 +1,314 @@ +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 +} + +// 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) + } + + if dispatched == 0 { + // All hooks were filtered out by phase — persist the annotation to + // avoid re-evaluating this task on every reporting cycle. + return wr.persistWebhookReportPhase(ctx, task, string(task.Status.Phase)) + } + + // Only persist the reported phase if all hooks succeeded. + if lastErr != nil { + return lastErr + } + + return wr.persistWebhookReportPhase(ctx, task, string(task.Status.Phase)) +} + +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 redirect policy applied. +// If no client is configured, a shared default is used. If an injected client +// lacks a CheckRedirect, a shallow copy with the policy is returned. +func (wr *WebhookReporter) httpClient() *http.Client { + if wr.HTTPClient == nil { + return defaultWebhookHTTPClient + } + if wr.HTTPClient.CheckRedirect != nil { + return wr.HTTPClient + } + clone := *wr.HTTPClient + clone.CheckRedirect = ssrfCheckRedirect + return &clone +} + +var defaultWebhookHTTPClient = &http.Client{ + Timeout: 10 * time.Second, + CheckRedirect: ssrfCheckRedirect, +} + +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 target private, loopback, or +// link-local addresses to prevent SSRF attacks. Domain names are resolved +// and all resulting IPs are checked. +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") + } + host := u.Hostname() + if ip := net.ParseIP(host); ip != nil { + if isPrivateIP(ip) { + return fmt.Errorf("webhook URL must not target private/internal addresses") + } + return nil + } + ips, err := net.LookupIP(host) + if err != nil { + return fmt.Errorf("resolving webhook host %q: %w", host, err) + } + for _, ip := range ips { + if isPrivateIP(ip) { + return fmt.Errorf("webhook URL must not target private/internal addresses") + } + } + return nil +} + +func isPrivateIP(ip net.IP) bool { + privateRanges := []net.IPNet{ + {IP: net.IPv4(10, 0, 0, 0), Mask: net.CIDRMask(8, 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.IPv4(169, 254, 0, 0), Mask: net.CIDRMask(16, 32)}, + {IP: net.IPv4(127, 0, 0, 0), Mask: net.CIDRMask(8, 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)}, + } + 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..bc84ff745 --- /dev/null +++ b/internal/reporting/webhook_test.go @@ -0,0 +1,502 @@ +package reporting + +import ( + "context" + "encoding/json" + "io" + "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 + wantNoAnnotation 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: "does not persist annotation on delivery failure", + 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, + wantNoAnnotation: true, + }, + { + 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) + } + } + + if tt.wantNoAnnotation { + var updated kelosv1alpha1.Task + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(tt.task), &updated); err != nil { + t.Fatalf("fetching task: %v", err) + } + if got, exists := updated.Annotations[AnnotationWebhookReportPhase]; exists { + t.Errorf("expected no %s annotation, got %q", AnnotationWebhookReportPhase, got) + } + } + }) + } +} + +func TestReportWebhooks_Idempotency(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + 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) + } + + // Second call should be a no-op (annotation already set). + requestCount := 0 + server2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + requestCount++ + w.WriteHeader(200) + })) + defer server2.Close() + + // 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) + } + + wr2 := &WebhookReporter{Client: cl, HTTPClient: server2.Client(), skipURLValidation: true} + if err := wr2.ReportWebhooks(context.Background(), &updated); err != nil { + t.Fatalf("second call: %v", err) + } + if requestCount != 0 { + t.Errorf("expected 0 requests on second call, got %d", 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_AppliesRedirectGuard(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 injected client without one") + } + if cl == injected { + t.Fatal("expected a clone, not the original client") + } + + withRedirect := &http.Client{ + Timeout: 5 * time.Second, + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { return nil }, + } + wr2 := &WebhookReporter{HTTPClient: withRedirect} + cl2 := wr2.httpClient() + if cl2 != withRedirect { + t.Fatal("expected original client to be returned when CheckRedirect is already set") + } +} + +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://169.254.169.254/latest/meta-data/", true}, + {"https://10.0.0.1/internal", true}, + {"https://172.16.0.1/internal", true}, + {"https://192.168.1.1/internal", true}, + {"https://127.0.0.1/local", true}, + {"https://[::1]:8080/exfil", true}, + {"https://[fe80::1]/link-local", true}, + {"https://[fd00::1]/unique-local", true}, + {"https://localhost/ssrf", 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 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) + } + } +} From c26261ca9c7ebf75db55f3dd7c2f3d395b97cf94 Mon Sep 17 00:00:00 2001 From: Irina Truong Date: Tue, 2 Jun 2026 15:56:43 -0700 Subject: [PATCH 2/3] fix: address review feedback on webhook reporter - Fix partial hook failure causing duplicate deliveries: persist the idempotency annotation even on partial failure to prevent re-delivery to hooks that already succeeded - Run webhook reporting before source-polling (not in defer) to avoid operating on a potentially-cancelled context - Fix CRD URL pattern to only block @ in the authority (host) portion, allowing legitimate @ in URL paths (e.g. MS Teams webhooks) - Fix ssrfSafeTransport to cleanly fall back when base is not *http.Transport, with extracted ssrfDialContext function - Cache the SSRF-wrapped HTTP client in WebhookReporter to avoid per-call transport cloning - Add 0.0.0.0/8 to private IP ranges (routes to loopback on Linux) - Guard against empty DNS result panic in ssrfDialContext - Move privateRanges to package-level var to avoid per-call allocation - Add missing onCompletionAnnotation call to stamp hooks onto tasks - Reject URLs with embedded credentials (userinfo) at both CRD and runtime validation layers Co-Authored-By: Claude Opus 4.6 --- api/v1alpha1/taskspawner_types.go | 2 +- cmd/kelos-spawner/main.go | 9 ++ cmd/kelos-spawner/reconciler.go | 13 +- .../kelos/templates/crds/taskspawner-crd.yaml | 2 +- internal/manifests/install-crd.yaml | 2 +- internal/reporting/webhook.go | 123 +++++++++++------- internal/reporting/webhook_test.go | 96 +++++++------- 7 files changed, 141 insertions(+), 106 deletions(-) diff --git a/api/v1alpha1/taskspawner_types.go b/api/v1alpha1/taskspawner_types.go index 50554ef12..c5d068f1e 100644 --- a/api/v1alpha1/taskspawner_types.go +++ b/api/v1alpha1/taskspawner_types.go @@ -947,7 +947,7 @@ type WebhookNotification struct { // read access to the Task resource. Do not use URLs with embedded // tokens; use secretRef for authentication. // +kubebuilder:validation:Required - // +kubebuilder:validation:Pattern="^https://.+" + // +kubebuilder:validation:Pattern="^https://[^@/]+(/.*)?$" URL string `json:"url"` // SecretRef optionally references a Secret whose keys are sent as HTTP diff --git a/cmd/kelos-spawner/main.go b/cmd/kelos-spawner/main.go index c0c60e134..372292a7e 100644 --- a/cmd/kelos-spawner/main.go +++ b/cmd/kelos-spawner/main.go @@ -416,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). diff --git a/cmd/kelos-spawner/reconciler.go b/cmd/kelos-spawner/reconciler.go index 0a9145dc7..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 } @@ -116,13 +122,6 @@ func runOnce(ctx context.Context, cl client.Client, key types.NamespacedName, cf } } - // Run onCompletion webhook reporting unconditionally — tasks carry their - // hook config in annotations, so webhooks must fire even if the spawner - // spec was updated after task creation. - if err := runWebhookReportingCycle(ctx, cl, key, cfg.HTTPClient); err != nil { - ctrl.Log.WithName("spawner").Error(err, "Webhook reporting cycle failed") - } - return resolvedPollInterval(&ts), nil } diff --git a/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml b/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml index 9156d628d..8f076c4b8 100644 --- a/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml +++ b/internal/manifests/charts/kelos/templates/crds/taskspawner-crd.yaml @@ -133,7 +133,7 @@ spec: 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://.+ + pattern: ^https://[^@/]+(/.*)?$ type: string required: - url diff --git a/internal/manifests/install-crd.yaml b/internal/manifests/install-crd.yaml index 459f20bcb..bb4d0e531 100644 --- a/internal/manifests/install-crd.yaml +++ b/internal/manifests/install-crd.yaml @@ -7361,7 +7361,7 @@ spec: 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://.+ + pattern: ^https://[^@/]+(/.*)?$ type: string required: - url diff --git a/internal/reporting/webhook.go b/internal/reporting/webhook.go index 7ec770fac..88d8894e9 100644 --- a/internal/reporting/webhook.go +++ b/internal/reporting/webhook.go @@ -51,6 +51,8 @@ type WebhookReporter struct { 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. @@ -108,18 +110,14 @@ func (wr *WebhookReporter) ReportWebhooks(ctx context.Context, task *kelosv1alph log.Info("Sent webhook notification", "task", task.Name, "hook", hook.Name, "phase", task.Status.Phase) } - if dispatched == 0 { - // All hooks were filtered out by phase — persist the annotation to - // avoid re-evaluating this task on every reporting cycle. - return wr.persistWebhookReportPhase(ctx, task, string(task.Status.Phase)) - } - - // Only persist the reported phase if all hooks succeeded. - if lastErr != nil { - return lastErr + // 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 wr.persistWebhookReportPhase(ctx, task, string(task.Status.Phase)) + return lastErr } func (wr *WebhookReporter) sendWebhook(ctx context.Context, namespace string, hook kelosv1alpha1.NotificationHook, payload WebhookPayload) error { @@ -166,26 +164,66 @@ func (wr *WebhookReporter) sendWebhook(ctx context.Context, namespace string, ho return nil } -// httpClient returns an HTTP client with SSRF-safe redirect policy applied. -// If no client is configured, a shared default is used. If an injected client -// lacks a CheckRedirect, a shallow copy with the policy is returned. +// 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.HTTPClient == nil { - return defaultWebhookHTTPClient - } - if wr.HTTPClient.CheckRedirect != nil { + if wr.skipURLValidation && wr.HTTPClient != nil { return wr.HTTPClient } - clone := *wr.HTTPClient - clone.CheckRedirect = ssrfCheckRedirect - return &clone + 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 { + 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 @@ -196,9 +234,9 @@ func ssrfCheckRedirect(req *http.Request, via []*http.Request) error { return nil } -// validateWebhookURL rejects URLs that target private, loopback, or -// link-local addresses to prevent SSRF attacks. Domain names are resolved -// and all resulting IPs are checked. +// 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 { @@ -207,36 +245,25 @@ func validateWebhookURL(rawURL string) error { if u.Scheme != "https" { return fmt.Errorf("webhook URL must use HTTPS") } - host := u.Hostname() - if ip := net.ParseIP(host); ip != nil { - if isPrivateIP(ip) { - return fmt.Errorf("webhook URL must not target private/internal addresses") - } - return nil - } - ips, err := net.LookupIP(host) - if err != nil { - return fmt.Errorf("resolving webhook host %q: %w", host, err) - } - for _, ip := range ips { - if isPrivateIP(ip) { - return fmt.Errorf("webhook URL must not target private/internal addresses") - } + 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 { - privateRanges := []net.IPNet{ - {IP: net.IPv4(10, 0, 0, 0), Mask: net.CIDRMask(8, 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.IPv4(169, 254, 0, 0), Mask: net.CIDRMask(16, 32)}, - {IP: net.IPv4(127, 0, 0, 0), Mask: net.CIDRMask(8, 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)}, - } for _, r := range privateRanges { if r.Contains(ip) { return true diff --git a/internal/reporting/webhook_test.go b/internal/reporting/webhook_test.go index bc84ff745..5278ddbca 100644 --- a/internal/reporting/webhook_test.go +++ b/internal/reporting/webhook_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "io" + "net" "net/http" "net/http/httptest" "testing" @@ -43,7 +44,6 @@ func TestWebhookReporter_ReportWebhooks(t *testing.T) { wantHeaders map[string]string wantAnnotation string wantErr bool - wantNoAnnotation bool }{ { name: "sends webhook on task succeeded", @@ -153,7 +153,7 @@ func TestWebhookReporter_ReportWebhooks(t *testing.T) { wantAnnotation: "Succeeded", }, { - name: "does not persist annotation on delivery failure", + name: "persists annotation on delivery failure to prevent duplicates", task: &kelosv1alpha1.Task{ ObjectMeta: metav1.ObjectMeta{ Name: "test-task-fail-delivery", @@ -165,10 +165,10 @@ func TestWebhookReporter_ReportWebhooks(t *testing.T) { Spec: kelosv1alpha1.TaskSpec{Type: "claude-code"}, Status: kelosv1alpha1.TaskStatus{Phase: kelosv1alpha1.TaskPhaseSucceeded}, }, - serverStatus: 500, - wantRequests: 1, - wantErr: true, - wantNoAnnotation: true, + serverStatus: 500, + wantRequests: 1, + wantErr: true, + wantAnnotation: "Succeeded", }, { name: "includes auth header from secret", @@ -318,21 +318,14 @@ func TestWebhookReporter_ReportWebhooks(t *testing.T) { } } - if tt.wantNoAnnotation { - var updated kelosv1alpha1.Task - if err := cl.Get(context.Background(), client.ObjectKeyFromObject(tt.task), &updated); err != nil { - t.Fatalf("fetching task: %v", err) - } - if got, exists := updated.Annotations[AnnotationWebhookReportPhase]; exists { - t.Errorf("expected no %s annotation, got %q", AnnotationWebhookReportPhase, got) - } - } }) } } 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() @@ -356,14 +349,9 @@ func TestReportWebhooks_Idempotency(t *testing.T) { if err := wr.ReportWebhooks(context.Background(), task); err != nil { t.Fatalf("first call: %v", err) } - - // Second call should be a no-op (annotation already set). - requestCount := 0 - server2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - requestCount++ - w.WriteHeader(200) - })) - defer server2.Close() + if requestCount != 1 { + t.Fatalf("expected 1 request on first call, got %d", requestCount) + } // Re-read task with updated annotations. var updated kelosv1alpha1.Task @@ -371,12 +359,13 @@ func TestReportWebhooks_Idempotency(t *testing.T) { t.Fatalf("fetching task: %v", err) } - wr2 := &WebhookReporter{Client: cl, HTTPClient: server2.Client(), skipURLValidation: true} - if err := wr2.ReportWebhooks(context.Background(), &updated); err != nil { + // 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 != 0 { - t.Errorf("expected 0 requests on second call, got %d", requestCount) + if requestCount != 1 { + t.Errorf("expected no additional requests on second call, got %d total", requestCount) } } @@ -432,26 +421,19 @@ func TestBuildWebhookPayload(t *testing.T) { } } -func TestHttpClient_AppliesRedirectGuard(t *testing.T) { +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 injected client without one") + 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") } - - withRedirect := &http.Client{ - Timeout: 5 * time.Second, - CheckRedirect: func(_ *http.Request, _ []*http.Request) error { return nil }, - } - wr2 := &WebhookReporter{HTTPClient: withRedirect} - cl2 := wr2.httpClient() - if cl2 != withRedirect { - t.Fatal("expected original client to be returned when CheckRedirect is already set") - } } func TestValidateWebhookURL(t *testing.T) { @@ -462,15 +444,8 @@ func TestValidateWebhookURL(t *testing.T) { {"https://hooks.slack.com/services/T00/B00/xxx", false}, {"https://example.com/webhook", false}, {"http://example.com/webhook", true}, - {"https://169.254.169.254/latest/meta-data/", true}, - {"https://10.0.0.1/internal", true}, - {"https://172.16.0.1/internal", true}, - {"https://192.168.1.1/internal", true}, - {"https://127.0.0.1/local", true}, - {"https://[::1]:8080/exfil", true}, - {"https://[fe80::1]/link-local", true}, - {"https://[fd00::1]/unique-local", true}, - {"https://localhost/ssrf", true}, + {"https://user:pass@example.com/webhook", true}, + {"https://token@example.com/webhook", true}, } for _, tt := range tests { err := validateWebhookURL(tt.url) @@ -480,6 +455,31 @@ func TestValidateWebhookURL(t *testing.T) { } } +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 From c8898c5309e60533bd8437017950421e939444b8 Mon Sep 17 00:00:00 2001 From: Irina Truong Date: Tue, 2 Jun 2026 16:05:16 -0700 Subject: [PATCH 3/3] fix: log warning when custom transport is not *http.Transport When the injected HTTP client uses a non-standard RoundTripper that cannot be type-asserted to *http.Transport, emit a log message so operators know the SSRF transport falls back to DefaultTransport settings. Co-Authored-By: Claude Opus 4.6 --- internal/reporting/webhook.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/reporting/webhook.go b/internal/reporting/webhook.go index 88d8894e9..ebaeb7dc2 100644 --- a/internal/reporting/webhook.go +++ b/internal/reporting/webhook.go @@ -196,6 +196,7 @@ func ssrfSafeTransport(base http.RoundTripper) http.RoundTripper { } 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()