// Package e2e contains read-only delivery verification primitives. package e2e import ( "context" "encoding/json" "errors" "fmt" "io" "net/http" "net/url" "os" "os/exec" "regexp" "strings" "time" ) const maxResponseBytes = 1 << 20 var ( dnsLabel = regexp.MustCompile(`^[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?$`) forgejoPart = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*$`) branchPattern = regexp.MustCompile(`^[A-Za-z0-9._/-]+$`) ) type Options struct { Kubeconfig string Context string FluxKustomizations []string ExternalSecret string PipelineRun string PreviewURL string PreviewSentinel string PromotionPullsURL string PromotionOwner string PromotionHead string PromotionToken string Timeout time.Duration Interval time.Duration } type Check struct { Name string `json:"name"` Status string `json:"status"` Detail string `json:"detail"` } type Result struct { Passed bool `json:"passed"` Checks []Check `json:"checks"` } // Command is deliberately small so command boundaries can be faked in tests. type Command interface { Output(context.Context, string, ...string) ([]byte, error) } type HTTPDoer interface { Do(*http.Request) (*http.Response, error) } type Runner struct { Kubectl Command HTTP HTTPDoer } type execCommand struct{} func (execCommand) Output(ctx context.Context, name string, args ...string) ([]byte, error) { return exec.CommandContext(ctx, name, args...).Output() } // ReadToken accepts only a reference to a token, never a token flag. func ReadToken(environment, path string) (string, error) { if environment != "" && path != "" { return "", errors.New("use only one promotion token reference") } var token string if environment != "" { var present bool token, present = os.LookupEnv(environment) if !present { return "", errors.New("promotion token environment variable is not set") } } else if path != "" { data, err := os.ReadFile(path) if err != nil { return "", errors.New("read promotion token file") } token = string(data) } else { return "", errors.New("a promotion token environment or file reference is required") } token = strings.TrimSpace(token) if token == "" || strings.ContainsAny(token, "\r\n") { return "", errors.New("promotion token reference is empty or invalid") } return token, nil } func (o Options) Validate() error { if o.Kubeconfig == "" { return errors.New("kubeconfig path is required") } if len(o.FluxKustomizations) == 0 { return errors.New("at least one Flux Kustomization is required") } seen := map[string]bool{} for _, resource := range o.FluxKustomizations { if !validNamespacedName(resource) || seen[resource] { return errors.New("Flux Kustomizations must be unique namespace/name identifiers") } seen[resource] = true } for _, resource := range []string{o.ExternalSecret, o.PipelineRun} { if !validNamespacedName(resource) { return errors.New("ExternalSecret and PipelineRun must be namespace/name identifiers") } } if err := validURL(o.PreviewURL); err != nil { return fmt.Errorf("preview URL: %w", err) } if o.PreviewSentinel == "" { return errors.New("preview sentinel is required") } if err := validURL(o.PromotionPullsURL); err != nil { return fmt.Errorf("promotion pulls URL: %w", err) } if !forgejoPart.MatchString(o.PromotionOwner) || strings.Contains(o.PromotionOwner, "..") { return errors.New("promotion owner is invalid") } if !branchPattern.MatchString(o.PromotionHead) || strings.Contains(o.PromotionHead, "..") || strings.HasPrefix(o.PromotionHead, "/") || strings.HasSuffix(o.PromotionHead, "/") || strings.Contains(o.PromotionHead, "//") { return errors.New("promotion head is invalid") } if o.PromotionToken == "" || strings.ContainsAny(o.PromotionToken, "\r\n") { return errors.New("promotion token is required") } if o.Timeout <= 0 || o.Timeout > 10*time.Minute { return errors.New("timeout must be between zero and ten minutes") } if o.Interval <= 0 || o.Interval > o.Timeout { return errors.New("interval must be positive and no longer than timeout") } return nil } func validNamespacedName(value string) bool { parts := strings.Split(value, "/") return len(parts) == 2 && dnsLabel.MatchString(parts[0]) && dnsLabel.MatchString(parts[1]) } func validURL(value string) error { parsed, err := url.Parse(value) if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" { return errors.New("must be a credential-free HTTP(S) URL without query or fragment") } return nil } // Run executes the fixed read-only verification order. It never applies, // reconciles, bootstraps, or writes cluster state. func (r Runner) Run(ctx context.Context, options Options) (Result, error) { if err := options.Validate(); err != nil { return Result{}, err } if r.Kubectl == nil || r.HTTP == nil { return Result{}, errors.New("e2e runner dependencies are required") } result := Result{Passed: true} result.add("flux_ready", r.wait(ctx, options, func(ctx context.Context) state { for _, resource := range options.FluxKustomizations { if !r.readyCondition(ctx, options, "kustomizations.kustomize.toolkit.fluxcd.io", resource) { return pending } } return ready })) result.add("external_secret_ready", r.wait(ctx, options, func(ctx context.Context) state { if r.readyCondition(ctx, options, "externalsecrets.external-secrets.io", options.ExternalSecret) { return ready } return pending })) result.add("pipeline_run_terminal", r.wait(ctx, options, func(ctx context.Context) state { return r.pipelineState(ctx, options) })) result.add("preview_sentinel", r.wait(ctx, options, func(ctx context.Context) state { return r.previewState(ctx, options) })) result.add("promotion_pr_open", r.wait(ctx, options, func(ctx context.Context) state { return r.promotionState(ctx, options) })) return result, nil } func (r *Result) add(name string, status state) { check := Check{Name: name, Status: "pass", Detail: "ready"} if status == failed { check.Status, check.Detail, r.Passed = "fail", "failed", false } if status == timedOut { check.Status, check.Detail, r.Passed = "fail", "timed_out", false } r.Checks = append(r.Checks, check) } type state int const ( pending state = iota ready failed timedOut ) func (r Runner) wait(parent context.Context, options Options, probe func(context.Context) state) state { ctx, cancel := context.WithTimeout(parent, options.Timeout) defer cancel() for { if current := probe(ctx); current != pending { return current } timer := time.NewTimer(options.Interval) select { case <-ctx.Done(): if !timer.Stop() { <-timer.C } return timedOut case <-timer.C: } } } func (r Runner) readyCondition(ctx context.Context, options Options, kind, resource string) bool { output, err := r.kubectl(ctx, options, kind, resource) if err != nil { return false } var value struct { Status struct { Conditions []struct { Type string `json:"type"` Status string `json:"status"` } `json:"conditions"` } `json:"status"` } if json.Unmarshal(output, &value) != nil { return false } for _, condition := range value.Status.Conditions { if condition.Type == "Ready" && condition.Status == "True" { return true } } return false } func (r Runner) pipelineState(ctx context.Context, options Options) state { output, err := r.kubectl(ctx, options, "pipelineruns.tekton.dev", options.PipelineRun) if err != nil { return pending } var value struct { Status struct { Conditions []struct { Type string `json:"type"` Status string `json:"status"` } `json:"conditions"` } `json:"status"` } if json.Unmarshal(output, &value) != nil { return pending } for _, condition := range value.Status.Conditions { if condition.Type != "Succeeded" { continue } switch condition.Status { case "True": return ready case "False": return failed } } return pending } func (r Runner) previewState(ctx context.Context, options Options) state { request, err := http.NewRequestWithContext(ctx, http.MethodGet, options.PreviewURL, nil) if err != nil { return failed } response, err := r.HTTP.Do(request) if err != nil { return pending } defer response.Body.Close() if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { return pending } body, err := io.ReadAll(io.LimitReader(response.Body, maxResponseBytes)) if err != nil { return pending } if strings.Contains(string(body), options.PreviewSentinel) { return ready } return pending } func (r Runner) promotionState(ctx context.Context, options Options) state { endpoint, err := url.Parse(options.PromotionPullsURL) if err != nil { return failed } query := url.Values{"state": {"open"}, "head": {options.PromotionOwner + ":" + options.PromotionHead}} endpoint.RawQuery = query.Encode() request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint.String(), nil) if err != nil { return failed } request.Header.Set("Authorization", "token "+options.PromotionToken) response, err := r.HTTP.Do(request) if err != nil { return pending } defer response.Body.Close() if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { return pending } var pulls []struct { State string `json:"state"` } if err := json.NewDecoder(io.LimitReader(response.Body, maxResponseBytes)).Decode(&pulls); err != nil { return pending } if len(pulls) == 0 { return pending } if len(pulls) != 1 || pulls[0].State != "open" { return failed } return ready } func (r Runner) kubectl(ctx context.Context, options Options, kind, resource string) ([]byte, error) { namespace, name, _ := strings.Cut(resource, "/") args := []string{"--kubeconfig=" + options.Kubeconfig} if options.Context != "" { args = append(args, "--context="+options.Context) } args = append(args, "--namespace="+namespace, "get", kind, name, "-o=json") return r.Kubectl.Output(ctx, "kubectl", args...) } func DefaultRunner() Runner { return Runner{Kubectl: execCommand{}, HTTP: &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}} }