feat: add resilient updates and access controls

internal/updateagent/controller.go

@@ -0,0 +1,133 @@
+package updateagent
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/root-kdw12/hetzner-api-dyndns-records-webinterface/internal/updateprotocol"
+)
+
+type Runner interface {
+	Update(context.Context, updateprotocol.Request, func(phase, message string)) error
+}
+
+type Controller struct {
+	runner Runner
+	logger *slog.Logger
+	mu     sync.RWMutex
+	status updateprotocol.Status
+}
+
+func NewController(runner Runner, logger *slog.Logger) *Controller {
+	return &Controller{runner: runner, logger: logger, status: updateprotocol.Status{Phase: updateprotocol.PhaseIdle, UpdatedAt: time.Now().UTC()}}
+}
+
+func (c *Controller) Handler() http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /v1/status", c.statusHandler)
+	mux.HandleFunc("POST /v1/update", c.updateHandler)
+	return mux
+}
+
+func (c *Controller) statusHandler(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.Header().Set("Cache-Control", "no-store")
+	_ = json.NewEncoder(w).Encode(c.Status())
+}
+
+func (c *Controller) updateHandler(w http.ResponseWriter, r *http.Request) {
+	r.Body = http.MaxBytesReader(w, r.Body, 32<<10)
+	defer r.Body.Close()
+	var request updateprotocol.Request
+	decoder := json.NewDecoder(io.LimitReader(r.Body, 32<<10))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&request); err != nil {
+		http.Error(w, "invalid update request", http.StatusBadRequest)
+		return
+	}
+	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
+		http.Error(w, "invalid update request", http.StatusBadRequest)
+		return
+	}
+	request.Version = strings.TrimSpace(request.Version)
+	request.RegistryUsername = strings.TrimSpace(request.RegistryUsername)
+	request.RegistryToken = strings.TrimSpace(request.RegistryToken)
+	if !updateprotocol.ValidVersion(request.Version) || request.RegistryUsername == "" || len(request.RegistryUsername) > 128 || len(request.RegistryToken) < 8 || len(request.RegistryToken) > 1024 {
+		http.Error(w, "invalid update request", http.StatusBadRequest)
+		return
+	}
+	if !c.begin(request.Version) {
+		http.Error(w, "update already running", http.StatusConflict)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(http.StatusAccepted)
+	_, _ = w.Write([]byte("{\"accepted\":true}\n"))
+	go c.run(request)
+}
+
+func (c *Controller) begin(version string) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.status.Busy() {
+		return false
+	}
+	now := time.Now().UTC()
+	c.status = updateprotocol.Status{Phase: updateprotocol.PhaseQueued, TargetVersion: version, Message: "Update eingereiht", StartedAt: now, UpdatedAt: now}
+	return true
+}
+
+func (c *Controller) run(request updateprotocol.Request) {
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
+	defer cancel()
+	err := c.runner.Update(ctx, request, c.setPhase)
+	registryToken := request.RegistryToken
+	request.RegistryToken = ""
+	if err != nil {
+		message := publicAgentError(err)
+		c.setPhase(updateprotocol.PhaseFailed, message)
+		c.logger.Error("software update failed", "version", request.Version, "error", redactSecret(err, registryToken))
+		return
+	}
+	c.setPhase(updateprotocol.PhaseComplete, "Update erfolgreich installiert")
+	c.logger.Info("software update completed", "version", request.Version)
+}
+
+func redactSecret(err error, secret string) error {
+	if err == nil || secret == "" || !strings.Contains(err.Error(), secret) {
+		return err
+	}
+	return errors.New(strings.ReplaceAll(err.Error(), secret, "[redacted]"))
+}
+
+func (c *Controller) setPhase(phase, message string) {
+	c.mu.Lock()
+	c.status.Phase = phase
+	c.status.Message = message
+	c.status.UpdatedAt = time.Now().UTC()
+	c.mu.Unlock()
+}
+
+func (c *Controller) Status() updateprotocol.Status {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.status
+}
+
+func publicAgentError(err error) string {
+	if err == nil {
+		return "Update fehlgeschlagen"
+	}
+	var public interface{ PublicMessage() string }
+	if errors.As(err, &public) {
+		return public.PublicMessage()
+	}
+	return "Update fehlgeschlagen; der bisherige Container wurde wiederhergestellt"
+}

internal/updateagent/controller_test.go

@@ -0,0 +1,95 @@
+package updateagent
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/root-kdw12/hetzner-api-dyndns-records-webinterface/internal/updateprotocol"
+)
+
+func TestRedactSecretRemovesCredentialFromErrors(t *testing.T) {
+	t.Parallel()
+	const secret = "sensitive-registry-token"
+	redacted := redactSecret(errors.New("registry rejected "+secret), secret)
+	if strings.Contains(redacted.Error(), secret) || !strings.Contains(redacted.Error(), "[redacted]") {
+		t.Fatalf("credential was not redacted: %q", redacted)
+	}
+}
+
+type blockingRunner struct {
+	started chan updateprotocol.Request
+	release chan struct{}
+}
+
+func (r *blockingRunner) Update(ctx context.Context, request updateprotocol.Request, progress func(phase, message string)) error {
+	progress(updateprotocol.PhasePulling, "image is being pulled")
+	r.started <- request
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-r.release:
+		return nil
+	}
+}
+
+func TestControllerValidatesRequestsAndRunsOnlyOneUpdate(t *testing.T) {
+	t.Parallel()
+	runner := &blockingRunner{started: make(chan updateprotocol.Request, 1), release: make(chan struct{})}
+	controller := NewController(runner, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	handler := controller.Handler()
+
+	invalid := httptest.NewRecorder()
+	handler.ServeHTTP(invalid, httptest.NewRequest(http.MethodPost, "/v1/update", strings.NewReader(`{"version":"stable","registry_username":"user","registry_token":"secret-token"}`)))
+	if invalid.Code != http.StatusBadRequest {
+		t.Fatalf("invalid version accepted: status=%d", invalid.Code)
+	}
+	trailing := httptest.NewRecorder()
+	handler.ServeHTTP(trailing, httptest.NewRequest(http.MethodPost, "/v1/update", strings.NewReader(`{"version":"v1.2.3","registry_username":"user","registry_token":"secret-token"} {}`)))
+	if trailing.Code != http.StatusBadRequest {
+		t.Fatalf("request with trailing JSON accepted: status=%d", trailing.Code)
+	}
+
+	const secret = "registry-token-must-not-leak"
+	body := []byte(`{"version":"v1.2.3","registry_username":"user","registry_token":"` + secret + `"}`)
+	accepted := httptest.NewRecorder()
+	handler.ServeHTTP(accepted, httptest.NewRequest(http.MethodPost, "/v1/update", bytes.NewReader(body)))
+	if accepted.Code != http.StatusAccepted {
+		t.Fatalf("valid update rejected: status=%d body=%s", accepted.Code, accepted.Body.String())
+	}
+	select {
+	case request := <-runner.started:
+		if request.Version != "v1.2.3" || request.RegistryToken != secret {
+			t.Fatalf("runner received an altered request: %#v", request)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("update runner did not start")
+	}
+
+	busy := httptest.NewRecorder()
+	handler.ServeHTTP(busy, httptest.NewRequest(http.MethodPost, "/v1/update", bytes.NewReader(body)))
+	if busy.Code != http.StatusConflict {
+		t.Fatalf("parallel update was not rejected: status=%d", busy.Code)
+	}
+	status := httptest.NewRecorder()
+	handler.ServeHTTP(status, httptest.NewRequest(http.MethodGet, "/v1/status", nil))
+	if status.Code != http.StatusOK || strings.Contains(status.Body.String(), secret) || !strings.Contains(status.Body.String(), `"phase":"pulling"`) {
+		t.Fatalf("unsafe or invalid public status: status=%d body=%s", status.Code, status.Body.String())
+	}
+
+	close(runner.release)
+	deadline := time.Now().Add(time.Second)
+	for controller.Status().Phase != updateprotocol.PhaseComplete && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if controller.Status().Phase != updateprotocol.PhaseComplete {
+		t.Fatalf("update did not complete: %#v", controller.Status())
+	}
+}

internal/updateagent/docker.go

@@ -0,0 +1,452 @@
+package updateagent
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/root-kdw12/hetzner-api-dyndns-records-webinterface/internal/updateprotocol"
+)
+
+const defaultImageRepository = "git.j-mahler.de/admin/domain-control-webinterface-prod"
+
+type DockerRunner struct {
+	docker          *dockerClient
+	imageRepository string
+	targetLabel     string
+	logger          *slog.Logger
+}
+
+type updateError struct {
+	message string
+	err     error
+}
+
+func (e *updateError) Error() string         { return e.err.Error() }
+func (e *updateError) Unwrap() error         { return e.err }
+func (e *updateError) PublicMessage() string { return e.message }
+
+func NewDockerRunner(socket, imageRepository, targetLabel string, logger *slog.Logger) *DockerRunner {
+	if imageRepository == "" {
+		imageRepository = defaultImageRepository
+	}
+	return &DockerRunner{docker: newDockerClient(socket), imageRepository: imageRepository, targetLabel: targetLabel, logger: logger}
+}
+
+func (r *DockerRunner) Update(ctx context.Context, request updateprotocol.Request, progress func(phase, message string)) (resultErr error) {
+	if !updateprotocol.ValidVersion(request.Version) {
+		return errors.New("invalid target version")
+	}
+	if err := r.docker.negotiate(ctx); err != nil {
+		return &updateError{message: "Containerverwaltung ist nicht erreichbar; der Dienst läuft unverändert weiter", err: err}
+	}
+	progress(updateprotocol.PhasePulling, "Container-Image wird geladen")
+	if err := r.docker.pullImage(ctx, r.imageRepository, request.Version, request.RegistryUsername, request.RegistryToken); err != nil {
+		return &updateError{message: "Image konnte nicht geladen werden; der Dienst läuft unverändert weiter", err: err}
+	}
+	target, err := r.docker.findTarget(ctx, r.targetLabel)
+	if err != nil {
+		return &updateError{message: "Laufender Anwendungscontainer wurde nicht eindeutig gefunden", err: err}
+	}
+	inspected, err := r.docker.inspectContainer(ctx, target.ID)
+	if err != nil {
+		return err
+	}
+	originalName := strings.TrimPrefix(inspected.Name, "/")
+	if originalName == "" {
+		return errors.New("target container has no name")
+	}
+	backupName := originalName + "-rollback-" + fmt.Sprint(time.Now().UTC().Unix())
+	var oldStopped, oldRenamed bool
+	var newID string
+	defer func() {
+		if resultErr == nil || (!oldStopped && !oldRenamed && newID == "") {
+			return
+		}
+		rollbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+		defer cancel()
+		if rollbackErr := r.rollback(rollbackCtx, target.ID, originalName, oldStopped, oldRenamed, newID); rollbackErr != nil {
+			r.logger.Error("software update rollback failed", "error", rollbackErr)
+			resultErr = &updateError{message: "Update und automatische Wiederherstellung sind fehlgeschlagen; Containerstatus am Host prüfen", err: fmt.Errorf("%w; rollback: %v", resultErr, rollbackErr)}
+		}
+	}()
+
+	progress(updateprotocol.PhaseRestarting, "Anwendung wird kontrolliert neu gestartet")
+	if err := r.docker.stopContainer(ctx, target.ID, 20); err != nil {
+		return err
+	}
+	oldStopped = true
+	if err := r.docker.renameContainer(ctx, target.ID, backupName); err != nil {
+		return err
+	}
+	oldRenamed = true
+	newID, err = r.docker.createReplacement(ctx, inspected, originalName, r.imageRepository+":"+request.Version)
+	if err != nil {
+		return err
+	}
+	if err := r.docker.startContainer(ctx, newID); err != nil {
+		return err
+	}
+	progress(updateprotocol.PhaseVerifying, "Neuer Container wird geprüft")
+	if err := r.docker.waitHealthy(ctx, newID, 100*time.Second); err != nil {
+		return err
+	}
+	if err := r.docker.removeContainer(ctx, target.ID, false); err != nil {
+		r.logger.Warn("old update rollback container could not be removed", "container", target.ID, "error", err)
+	}
+	oldStopped, oldRenamed, newID = false, false, ""
+	return nil
+}
+
+func (r *DockerRunner) rollback(ctx context.Context, oldID, originalName string, oldStopped, oldRenamed bool, newID string) error {
+	var failures []string
+	if newID != "" {
+		if err := r.docker.stopContainer(ctx, newID, 5); err != nil && !isDockerNotModified(err) {
+			failures = append(failures, "stop replacement: "+err.Error())
+		}
+		if err := r.docker.removeContainer(ctx, newID, true); err != nil {
+			failures = append(failures, "remove replacement: "+err.Error())
+		}
+	}
+	if oldRenamed {
+		if err := r.docker.renameContainer(ctx, oldID, originalName); err != nil {
+			failures = append(failures, "restore name: "+err.Error())
+		}
+	}
+	if oldStopped {
+		if err := r.docker.startContainer(ctx, oldID); err != nil && !isDockerNotModified(err) {
+			failures = append(failures, "restart previous container: "+err.Error())
+		}
+	}
+	if len(failures) > 0 {
+		return errors.New(strings.Join(failures, "; "))
+	}
+	return nil
+}
+
+type dockerClient struct {
+	socket    string
+	client    *http.Client
+	apiPrefix string
+}
+
+type containerSummary struct {
+	ID    string   `json:"Id"`
+	Names []string `json:"Names"`
+	State string   `json:"State"`
+}
+
+type containerInspect struct {
+	ID              string          `json:"Id"`
+	Name            string          `json:"Name"`
+	Config          json.RawMessage `json:"Config"`
+	HostConfig      json.RawMessage `json:"HostConfig"`
+	NetworkSettings struct {
+		Networks map[string]endpointSettings `json:"Networks"`
+	} `json:"NetworkSettings"`
+	State containerState `json:"State"`
+}
+
+type endpointSettings struct {
+	IPAMConfig json.RawMessage   `json:"IPAMConfig,omitempty"`
+	Links      []string          `json:"Links,omitempty"`
+	Aliases    []string          `json:"Aliases,omitempty"`
+	MacAddress string            `json:"MacAddress,omitempty"`
+	DriverOpts map[string]string `json:"DriverOpts,omitempty"`
+	GwPriority int               `json:"GwPriority,omitempty"`
+}
+
+type containerState struct {
+	Running  bool `json:"Running"`
+	ExitCode int  `json:"ExitCode"`
+	Health   *struct {
+		Status string `json:"Status"`
+	} `json:"Health"`
+}
+
+func newDockerClient(socket string) *dockerClient {
+	transport := &http.Transport{
+		DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+			return (&net.Dialer{Timeout: 2 * time.Second}).DialContext(ctx, "unix", socket)
+		},
+		MaxIdleConns:        2,
+		MaxIdleConnsPerHost: 2,
+		IdleConnTimeout:     30 * time.Second,
+	}
+	return &dockerClient{socket: socket, client: &http.Client{Transport: transport}}
+}
+
+func (c *dockerClient) negotiate(ctx context.Context) error {
+	response, err := c.request(ctx, http.MethodGet, "/version", nil, nil, false)
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return dockerStatusError(response)
+	}
+	var version struct {
+		APIVersion string `json:"ApiVersion"`
+	}
+	if err := json.NewDecoder(io.LimitReader(response.Body, 64<<10)).Decode(&version); err != nil {
+		return fmt.Errorf("decode Docker API version: %w", err)
+	}
+	for _, character := range version.APIVersion {
+		if (character < '0' || character > '9') && character != '.' {
+			return errors.New("Docker returned an invalid API version")
+		}
+	}
+	if version.APIVersion == "" {
+		return errors.New("Docker returned an empty API version")
+	}
+	c.apiPrefix = "/v" + version.APIVersion
+	return nil
+}
+
+func (c *dockerClient) findTarget(ctx context.Context, label string) (containerSummary, error) {
+	filters, _ := json.Marshal(map[string][]string{"label": {label}})
+	query := url.Values{"all": {"1"}, "filters": {string(filters)}}
+	response, err := c.request(ctx, http.MethodGet, "/containers/json?"+query.Encode(), nil, nil, true)
+	if err != nil {
+		return containerSummary{}, err
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return containerSummary{}, dockerStatusError(response)
+	}
+	var containers []containerSummary
+	if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&containers); err != nil {
+		return containerSummary{}, err
+	}
+	var running []containerSummary
+	for _, container := range containers {
+		if container.State == "running" {
+			running = append(running, container)
+		}
+	}
+	if len(running) != 1 {
+		return containerSummary{}, fmt.Errorf("expected exactly one running target container, found %d", len(running))
+	}
+	return running[0], nil
+}
+
+func (c *dockerClient) pullImage(ctx context.Context, repository, tag, username, token string) error {
+	authJSON, _ := json.Marshal(map[string]string{"username": username, "password": token, "serveraddress": strings.Split(repository, "/")[0]})
+	headers := http.Header{"X-Registry-Auth": {base64.URLEncoding.EncodeToString(authJSON)}}
+	query := url.Values{"fromImage": {repository}, "tag": {tag}}
+	response, err := c.request(ctx, http.MethodPost, "/images/create?"+query.Encode(), nil, headers, true)
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return dockerStatusError(response)
+	}
+	scanner := bufio.NewScanner(response.Body)
+	scanner.Buffer(make([]byte, 64<<10), 1<<20)
+	for scanner.Scan() {
+		var event struct {
+			Error string `json:"error"`
+		}
+		if err := json.Unmarshal(scanner.Bytes(), &event); err == nil && event.Error != "" {
+			return errors.New(event.Error)
+		}
+	}
+	return scanner.Err()
+}
+
+func (c *dockerClient) inspectContainer(ctx context.Context, id string) (containerInspect, error) {
+	response, err := c.request(ctx, http.MethodGet, "/containers/"+url.PathEscape(id)+"/json", nil, nil, true)
+	if err != nil {
+		return containerInspect{}, err
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return containerInspect{}, dockerStatusError(response)
+	}
+	var inspected containerInspect
+	if err := json.NewDecoder(io.LimitReader(response.Body, 4<<20)).Decode(&inspected); err != nil {
+		return containerInspect{}, err
+	}
+	return inspected, nil
+}
+
+func (c *dockerClient) stopContainer(ctx context.Context, id string, seconds int) error {
+	query := url.Values{"t": {fmt.Sprint(seconds)}}
+	response, err := c.request(ctx, http.MethodPost, "/containers/"+url.PathEscape(id)+"/stop?"+query.Encode(), nil, nil, true)
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusNoContent || response.StatusCode == http.StatusNotModified {
+		return nil
+	}
+	return dockerStatusError(response)
+}
+
+func (c *dockerClient) renameContainer(ctx context.Context, id, name string) error {
+	query := url.Values{"name": {name}}
+	response, err := c.request(ctx, http.MethodPost, "/containers/"+url.PathEscape(id)+"/rename?"+query.Encode(), nil, nil, true)
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusNoContent {
+		return nil
+	}
+	return dockerStatusError(response)
+}
+
+func (c *dockerClient) createReplacement(ctx context.Context, inspected containerInspect, name, image string) (string, error) {
+	var body map[string]json.RawMessage
+	if err := json.Unmarshal(inspected.Config, &body); err != nil {
+		return "", fmt.Errorf("decode container config: %w", err)
+	}
+	imageJSON, _ := json.Marshal(image)
+	body["Image"] = imageJSON
+	body["HostConfig"] = inspected.HostConfig
+	endpoints := make(map[string]endpointSettings, len(inspected.NetworkSettings.Networks))
+	oldIDPrefix := inspected.ID
+	if len(oldIDPrefix) > 12 {
+		oldIDPrefix = oldIDPrefix[:12]
+	}
+	for network, endpoint := range inspected.NetworkSettings.Networks {
+		aliases := endpoint.Aliases[:0]
+		for _, alias := range endpoint.Aliases {
+			if alias != inspected.ID && alias != oldIDPrefix {
+				aliases = append(aliases, alias)
+			}
+		}
+		endpoint.Aliases = aliases
+		endpoints[network] = endpoint
+	}
+	networkingJSON, _ := json.Marshal(map[string]any{"EndpointsConfig": endpoints})
+	body["NetworkingConfig"] = networkingJSON
+	encoded, err := json.Marshal(body)
+	if err != nil {
+		return "", err
+	}
+	query := url.Values{"name": {name}}
+	response, err := c.request(ctx, http.MethodPost, "/containers/create?"+query.Encode(), bytes.NewReader(encoded), http.Header{"Content-Type": {"application/json"}}, true)
+	if err != nil {
+		return "", err
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusCreated {
+		return "", dockerStatusError(response)
+	}
+	var created struct {
+		ID string `json:"Id"`
+	}
+	if err := json.NewDecoder(io.LimitReader(response.Body, 64<<10)).Decode(&created); err != nil {
+		return "", err
+	}
+	if created.ID == "" {
+		return "", errors.New("Docker created a replacement without an ID")
+	}
+	return created.ID, nil
+}
+
+func (c *dockerClient) startContainer(ctx context.Context, id string) error {
+	response, err := c.request(ctx, http.MethodPost, "/containers/"+url.PathEscape(id)+"/start", nil, nil, true)
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusNoContent || response.StatusCode == http.StatusNotModified {
+		return nil
+	}
+	return dockerStatusError(response)
+}
+
+func (c *dockerClient) removeContainer(ctx context.Context, id string, force bool) error {
+	query := url.Values{"force": {fmt.Sprint(force)}, "v": {"0"}}
+	response, err := c.request(ctx, http.MethodDelete, "/containers/"+url.PathEscape(id)+"?"+query.Encode(), nil, nil, true)
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusNoContent || response.StatusCode == http.StatusNotFound {
+		return nil
+	}
+	return dockerStatusError(response)
+}
+
+func (c *dockerClient) waitHealthy(ctx context.Context, id string, timeout time.Duration) error {
+	deadline := time.NewTimer(timeout)
+	ticker := time.NewTicker(time.Second)
+	defer deadline.Stop()
+	defer ticker.Stop()
+	consecutiveHealthy := 0
+	startedAt := time.Now()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-deadline.C:
+			return errors.New("replacement container did not become healthy in time")
+		case <-ticker.C:
+			inspected, err := c.inspectContainer(ctx, id)
+			if err != nil {
+				return err
+			}
+			if !inspected.State.Running {
+				return fmt.Errorf("replacement container stopped with exit code %d", inspected.State.ExitCode)
+			}
+			if inspected.State.Health == nil {
+				if time.Since(startedAt) >= 8*time.Second {
+					return nil
+				}
+				continue
+			}
+			if inspected.State.Health.Status == "healthy" {
+				consecutiveHealthy++
+				if consecutiveHealthy >= 3 {
+					return nil
+				}
+			} else {
+				consecutiveHealthy = 0
+			}
+		}
+	}
+}
+
+func (c *dockerClient) request(ctx context.Context, method, path string, body io.Reader, headers http.Header, versioned bool) (*http.Response, error) {
+	if versioned {
+		path = c.apiPrefix + path
+	}
+	request, err := http.NewRequestWithContext(ctx, method, "http://docker"+path, body)
+	if err != nil {
+		return nil, err
+	}
+	for name, values := range headers {
+		for _, value := range values {
+			request.Header.Add(name, value)
+		}
+	}
+	return c.client.Do(request)
+}
+
+func dockerStatusError(response *http.Response) error {
+	message, _ := io.ReadAll(io.LimitReader(response.Body, 8<<10))
+	trimmed := strings.TrimSpace(string(message))
+	if trimmed == "" {
+		trimmed = http.StatusText(response.StatusCode)
+	}
+	return fmt.Errorf("Docker API HTTP %d: %s", response.StatusCode, trimmed)
+}
+
+func isDockerNotModified(err error) bool {
+	return err != nil && strings.Contains(err.Error(), "HTTP 304")
+}