maidn-cli/internal/e2e/e2e.go

// 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 }}}
}