core/auth/ui/e2e/stack.go

package e2e

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"time"
)

type stackPorts struct {
	KratosPublic int `json:"kratos_public"`
	KratosAdmin  int `json:"kratos_admin"`
	HydraPublic  int `json:"hydra_public"`
	HydraAdmin   int `json:"hydra_admin"`
	UI           int `json:"ui"`
	API          int `json:"api"`
	SMTP         int `json:"smtp"`
}

type Stack struct {
	Repo           string
	Workspace      string
	ArtifactDir    string
	Ports          stackPorts
	KratosURL      string
	KratosAdmin    string
	HydraURL       string
	HydraAdmin     string
	UIURL          string
	APIURL         string
	Kratos         *supervisedProcess
	Hydra          *supervisedProcess
	AuthUI         *supervisedProcess
	startedAt      time.Time
	ready          bool
	lastStatus     string
	metadataMu     sync.Mutex
	redactions     []string
	ownershipMu    sync.Mutex
	cleanupStarted bool
	stopOnce       sync.Once
	stopErr        error
	startProcessOp func(string, string, string, string, ...string) (*supervisedProcess, error)
	stopProcessOp  func(*supervisedProcess, time.Duration) error
}

func startStack() (*Stack, error) {
	return startStackOwned(nil)
}

func startStackOwned(owner *suiteLifecycle) (*Stack, error) {
	if _, err := artifactsFor(runtime.GOOS, runtime.GOARCH); err != nil {
		return nil, err
	}
	repo, err := repositoryDir()
	if err != nil {
		return nil, err
	}
	workspace, err := os.MkdirTemp("", "auth-ui-e2e-*")
	if err != nil {
		return nil, err
	}
	runDir, err := createRunArtifactDir(repo)
	if err != nil {
		_ = os.RemoveAll(workspace)
		return nil, err
	}
	s := &Stack{Repo: repo, Workspace: workspace, ArtifactDir: runDir, startedAt: time.Now().UTC()}
	if owner != nil && !owner.setStack(s) {
		_ = s.Close(false)
		return s, fmt.Errorf("E2E watchdog fired before stack ownership registration")
	}
	if err := s.prepare(); err != nil {
		s.recordRun("setup_failed", err)
		return s, fmt.Errorf("prepare E2E stack: %w", err)
	}
	for attempt := 1; attempt <= 3; attempt++ {
		err = s.startAttempt()
		if err == nil {
			s.metadataMu.Lock()
			s.ready = true
			s.metadataMu.Unlock()
			s.recordRun("ready", nil)
			return s, nil
		}
		cleanupErr := s.stopProcessSet()
		retry, attemptErr := startupRetryDecision(attempt, err, cleanupErr)
		if !retry {
			s.recordRun("setup_failed", attemptErr)
			return s, attemptErr
		}
	}
	return s, err
}

func (s *Stack) prepare() error {
	for _, dir := range []string{"bin", "config", "logs"} {
		if err := os.MkdirAll(filepath.Join(s.Workspace, dir), 0o700); err != nil {
			return err
		}
	}
	platform, _ := artifactsFor(runtime.GOOS, runtime.GOARCH)
	offline := os.Getenv("AUTH_UI_E2E_OFFLINE") == "1"
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	defer cancel()
	for _, artifact := range []releaseArtifact{platform.Kratos, platform.Hydra} {
		archivePath, err := ensureArchive(ctx, downloadClient(), cacheRoot(s.Repo), artifact, offline)
		if err != nil {
			return err
		}
		binary, err := extractBinary(archivePath, filepath.Join(s.Workspace, "bin"), artifact.Service)
		if err != nil {
			return err
		}
		if err := verifyBinaryVersion(binary, artifact.Version); err != nil {
			return err
		}
	}
	ctx, cancel = context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	cmd := exec.CommandContext(ctx, "go", "build", "-o", filepath.Join(s.Workspace, "bin", "auth-ui"), ".")
	cmd.Dir = s.Repo
	output, err := cmd.CombinedOutput()
	if ctx.Err() != nil {
		return fmt.Errorf("build auth-ui: %w", ctx.Err())
	}
	if err != nil {
		return fmt.Errorf("build auth-ui: %w: %s", err, boundedText(output, 8192))
	}
	return nil
}

func (s *Stack) startAttempt() error {
	reservations, ports, err := reservePorts(7)
	if err != nil {
		return err
	}
	defer func() {
		for _, reservation := range reservations {
			_ = reservation.Close()
		}
	}()
	s.metadataMu.Lock()
	s.Ports = stackPorts{ports[0], ports[1], ports[2], ports[3], ports[4], ports[5], ports[6]}
	s.KratosURL = loopbackURL(s.Ports.KratosPublic)
	s.KratosAdmin = loopbackURL(s.Ports.KratosAdmin)
	s.HydraURL = loopbackURL(s.Ports.HydraPublic)
	s.HydraAdmin = loopbackURL(s.Ports.HydraAdmin)
	s.UIURL = loopbackURL(s.Ports.UI)
	s.APIURL = loopbackURL(s.Ports.API)
	s.metadataMu.Unlock()
	cookie, err := randomSecret(16)
	if err != nil {
		return err
	}
	cipher, err := randomSecret(16)
	if err != nil {
		return err
	}
	hydra, err := randomSecret(32)
	if err != nil {
		return err
	}
	s.metadataMu.Lock()
	s.redactions = []string{cookie, cipher, hydra}
	s.metadataMu.Unlock()
	values := fixtureValues{
		KratosPublicURL: s.KratosURL, KratosAdminURL: s.KratosAdmin,
		KratosPublicPort: s.Ports.KratosPublic, KratosAdminPort: s.Ports.KratosAdmin,
		HydraPublicURL: s.HydraURL, HydraAdminURL: s.HydraAdmin,
		HydraPublicPort: s.Ports.HydraPublic, HydraAdminPort: s.Ports.HydraAdmin,
		UIURL: s.UIURL, SMTPPort: s.Ports.SMTP,
		CookieSecret: cookie, CipherSecret: cipher, HydraSecret: hydra,
	}
	if err := renderFixtures(s.Repo, filepath.Join(s.Workspace, "config"), values); err != nil {
		return err
	}
	_ = reservations[6].Close() // No courier is started; only a valid unused SMTP address is needed.

	_ = reservations[0].Close()
	_ = reservations[1].Close()
	process, err := s.startOwnedProcess(&s.Kratos, "kratos", s.logPath("kratos"), s.Workspace, s.binary("kratos"), "serve", "-c", s.config("kratos.yml"), "--dev", "--sqa-opt-out")
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	err = waitHTTPReady(ctx, process, s.KratosURL+"/health/ready", s.KratosAdmin+"/health/ready")
	cancel()
	if err != nil {
		return err
	}

	_ = reservations[2].Close()
	_ = reservations[3].Close()
	process, err = s.startOwnedProcess(&s.Hydra, "hydra", s.logPath("hydra"), s.Workspace, s.binary("hydra"), "serve", "-c", s.config("hydra.yml"), "all", "--dev", "--sqa-opt-out")
	if err != nil {
		return err
	}
	ctx, cancel = context.WithTimeout(context.Background(), 30*time.Second)
	err = waitHTTPReady(ctx, process, s.HydraURL+"/health/ready", s.HydraAdmin+"/health/ready")
	cancel()
	if err != nil {
		return err
	}

	args := []string{"-port", strconv.Itoa(s.Ports.UI), "-api-port", strconv.Itoa(s.Ports.API), "-kratos", s.KratosURL, "-kratos-api", s.KratosAdmin, "-hydra", "127.0.0.1:" + strconv.Itoa(s.Ports.HydraAdmin), "-enable-registration=true", "-email-domain=example.test"}
	_ = reservations[4].Close()
	_ = reservations[5].Close()
	process, err = s.startOwnedProcess(&s.AuthUI, "auth-ui", s.logPath("auth-ui"), s.Repo, s.binary("auth-ui"), args...)
	if err != nil {
		return err
	}
	ctx, cancel = context.WithTimeout(context.Background(), 30*time.Second)
	err = waitTCPReady(ctx, process,
		"127.0.0.1:"+strconv.Itoa(s.Ports.UI),
		"127.0.0.1:"+strconv.Itoa(s.Ports.API),
	)
	cancel()
	return err
}

func reservePorts(count int) ([]net.Listener, []int, error) {
	listeners := make([]net.Listener, 0, count)
	ports := make([]int, 0, count)
	for len(ports) < count {
		listener, err := net.Listen("tcp", "127.0.0.1:0")
		if err != nil {
			for _, reserved := range listeners {
				_ = reserved.Close()
			}
			return nil, nil, err
		}
		listeners = append(listeners, listener)
		ports = append(ports, listener.Addr().(*net.TCPAddr).Port)
	}
	return listeners, ports, nil
}

func loopbackURL(port int) string           { return "http://127.0.0.1:" + strconv.Itoa(port) }
func (s *Stack) binary(name string) string  { return filepath.Join(s.Workspace, "bin", name) }
func (s *Stack) config(name string) string  { return filepath.Join(s.Workspace, "config", name) }
func (s *Stack) logPath(name string) string { return filepath.Join(s.Workspace, "logs", name+".log") }

func isBindConflict(err error) bool {
	text := strings.ToLower(err.Error())
	return strings.Contains(text, "address already in use") || strings.Contains(text, "address in use")
}

func startupRetryDecision(attempt int, startupErr, cleanupErr error) (bool, error) {
	if cleanupErr != nil {
		return false, fmt.Errorf("startup attempt %d failed: %v; retry cleanup failed: %w", attempt, startupErr, cleanupErr)
	}
	if !isBindConflict(startupErr) || attempt >= 3 {
		return false, startupErr
	}
	return true, nil
}

func (s *Stack) startOwnedProcess(slot **supervisedProcess, name, logPath, dir, executable string, args ...string) (*supervisedProcess, error) {
	// Cleanup and process creation share this lock from the terminal-state check
	// through cmd.Start and slot publication. StopServices can therefore either
	// prevent the spawn or observe the complete ownership; it cannot snapshot a
	// nil slot while a process is being created.
	s.ownershipMu.Lock()
	defer s.ownershipMu.Unlock()
	if s.cleanupStarted {
		return nil, fmt.Errorf("E2E cleanup started before launching %s", name)
	}
	start := s.startProcessOp
	if start == nil {
		start = startProcess
	}
	process, err := start(name, logPath, dir, executable, args...)
	if err != nil {
		return process, err
	}
	if process == nil {
		return nil, fmt.Errorf("started %s process ownership is unavailable", name)
	}
	process.redactions = append([]string(nil), s.redactions...)
	*slot = process
	return process, nil
}

func (s *Stack) ownedProcessGroups() []int {
	if s == nil {
		return nil
	}
	s.ownershipMu.Lock()
	defer s.ownershipMu.Unlock()
	return s.ownedProcessGroupsLocked()
}

func (s *Stack) emergencyProcessGroups() []int {
	if s == nil || !s.ownershipMu.TryLock() {
		// Descendant discovery remains available when an OS start is holding the
		// ownership lock, so emergency exit must never block on this mutex.
		return nil
	}
	defer s.ownershipMu.Unlock()
	return s.ownedProcessGroupsLocked()
}

func (s *Stack) ownedProcessGroupsLocked() []int {
	var groups []int
	for _, process := range []*supervisedProcess{s.AuthUI, s.Hydra, s.Kratos} {
		if process != nil && process.pgid > 0 {
			groups = append(groups, process.pgid)
		}
	}
	return groups
}

func (s *Stack) stopOwnedProcess(process *supervisedProcess, timeout time.Duration) error {
	if s.stopProcessOp != nil {
		return s.stopProcessOp(process, timeout)
	}
	return process.stopAndWait(timeout)
}

func (s *Stack) stopProcessSet() error {
	s.ownershipMu.Lock()
	processes := []*supervisedProcess{s.AuthUI, s.Hydra, s.Kratos}
	s.ownershipMu.Unlock()
	var errs []string
	for _, process := range processes {
		if process == nil {
			continue
		}
		stopErr := s.stopOwnedProcess(process, 5*time.Second)
		if stopErr != nil {
			errs = append(errs, stopErr.Error())
		}
		// Do not discard ownership merely because cleanup returned. Retain the
		// process until its complete owned group is confirmed absent.
		if ownedProcessGroupAlive(process.pgid) {
			if stopErr == nil {
				errs = append(errs, fmt.Sprintf("%s process group %d survived cleanup", process.name, process.pgid))
			}
		} else {
			s.ownershipMu.Lock()
			if s.AuthUI == process {
				s.AuthUI = nil
			}
			if s.Hydra == process {
				s.Hydra = nil
			}
			if s.Kratos == process {
				s.Kratos = nil
			}
			s.ownershipMu.Unlock()
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("stop service processes: %s", strings.Join(errs, "; "))
	}
	return nil
}

func (s *Stack) StopServices() error {
	if s == nil {
		return nil
	}
	s.ownershipMu.Lock()
	s.cleanupStarted = true
	s.ownershipMu.Unlock()
	s.stopOnce.Do(func() { s.stopErr = s.stopProcessSet() })
	return s.stopErr
}

func (s *Stack) Close(success bool) error {
	if s == nil {
		return nil
	}
	stopErr := s.StopServices()
	finalizeErr := s.Finalize(success)
	if stopErr != nil && finalizeErr != nil {
		return fmt.Errorf("%v; %w", stopErr, finalizeErr)
	}
	if stopErr != nil {
		return stopErr
	}
	return finalizeErr
}

func (s *Stack) Finalize(success bool) error {
	s.metadataMu.Lock()
	ready, lastStatus := s.ready, s.lastStatus
	redactions := append([]string(nil), s.redactions...)
	s.metadataMu.Unlock()
	if ready {
		if success {
			s.recordRun("passed", nil)
		} else if lastStatus == "ready" {
			s.recordRun("failed", nil)
		}
	}
	var errs []string
	services := filepath.Join(s.ArtifactDir, "services")
	if err := os.MkdirAll(services, 0o700); err != nil {
		errs = append(errs, err.Error())
	}
	for _, name := range []string{"auth-ui", "hydra", "kratos"} {
		if err := copyFileBounded(s.logPath(name), filepath.Join(services, name+".log"), 1<<20, redactions); err != nil && !errors.Is(err, os.ErrNotExist) {
			errs = append(errs, err.Error())
		}
	}
	keep := !success || os.Getenv("AUTH_UI_E2E_KEEP_TMP") == "1"
	if keep {
		fmt.Fprintf(os.Stderr, "E2E workspace retained at %s\nE2E artifacts retained at %s\n", s.Workspace, s.ArtifactDir)
	} else if err := os.RemoveAll(s.Workspace); err != nil {
		errs = append(errs, err.Error())
	}
	if len(errs) > 0 {
		return fmt.Errorf("stack cleanup: %s", strings.Join(errs, "; "))
	}
	return nil
}

func createRunArtifactDir(repo string) (string, error) {
	root := os.Getenv("AUTH_UI_E2E_ARTIFACT_DIR")
	if root == "" {
		root = filepath.Join(repo, "e2e", "artifacts")
	}
	if err := os.MkdirAll(root, 0o700); err != nil {
		return "", err
	}
	random, err := randomSecret(4)
	if err != nil {
		return "", err
	}
	name := fmt.Sprintf("%s-%d-%s", time.Now().UTC().Format("20060102T150405.000000000Z"), os.Getpid(), random)
	path := filepath.Join(root, name)
	return path, os.Mkdir(path, 0o700)
}

func (s *Stack) recordRun(status string, setupErr error) {
	s.metadataMu.Lock()
	defer s.metadataMu.Unlock()
	if s.lastStatus == "watchdog_timeout" && status != "watchdog_timeout" {
		return
	}
	s.lastStatus = status
	metadata := map[string]any{"started_at": s.startedAt, "updated_at": time.Now().UTC(), "status": status, "workspace": s.Workspace, "platform": runtime.GOOS + "/" + runtime.GOARCH, "kratos_version": kratosVersion, "hydra_version": hydraVersion, "playwright_binding_version": playwrightVersion, "playwright_cli_version": playwrightCLIVersion, "chromium_revision": chromiumRevision, "chromium_version": chromiumVersion, "ffmpeg_revision": ffmpegRevision, "ports": s.Ports}
	if setupErr != nil {
		metadata["error"] = redactDiagnostic(setupErr.Error())
	}
	data, err := json.MarshalIndent(metadata, "", "  ")
	if err == nil {
		_ = os.WriteFile(filepath.Join(s.ArtifactDir, "run.json"), append(data, '\n'), 0o600)
	}
}

func redactDiagnostic(value string) string {
	// Text logs intentionally omit generated secrets. Strip query strings from URLs
	// as a final guard against opaque flow/challenge/token values.
	words := strings.Fields(value)
	for i, word := range words {
		if strings.Contains(word, "http://") && strings.Contains(word, "?") {
			words[i] = strings.SplitN(word, "?", 2)[0]
		}
	}
	return strings.Join(words, " ")
}

func copyFileBounded(source, destination string, limit int64, redactions []string) error {
	f, err := os.Open(source)
	if err != nil {
		return err
	}
	defer f.Close()
	info, err := f.Stat()
	if err != nil {
		return err
	}
	if info.Size() > limit {
		if _, err := f.Seek(info.Size()-limit, io.SeekStart); err != nil {
			return err
		}
	}
	data, err := io.ReadAll(io.LimitReader(f, limit))
	if err != nil {
		return err
	}
	text := string(data)
	for _, secret := range redactions {
		if secret != "" {
			text = strings.ReplaceAll(text, secret, "[REDACTED]")
		}
	}
	return os.WriteFile(destination, []byte(text), 0o600)
}

func (s *Stack) healthy(endpoint string) error {
	client := &http.Client{Timeout: 2 * time.Second}
	defer client.CloseIdleConnections()
	resp, err := client.Get(endpoint)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("health status %s", resp.Status)
	}
	return nil
}