diff --git a/.dockerignore b/.dockerignore index 567085a..f9ce846 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,5 +1,6 @@ .git .env +subscription.url coverage.out tests README.md diff --git a/internal/bootstrap/candidate.go b/internal/bootstrap/candidate.go index 630f7b9..1428984 100644 --- a/internal/bootstrap/candidate.go +++ b/internal/bootstrap/candidate.go @@ -13,6 +13,7 @@ import ( "os" "path/filepath" "strings" + "syscall" "time" "gopkg.in/yaml.v3" @@ -24,12 +25,16 @@ const ( defaultControllerURL = "http://127.0.0.1:9090" ) +var errPersistence = errors.New("candidate persistence failure") + type CandidateConfig struct { - SecretPath string - DataDir string - TemplatePath string - MihomoBinary string - Client *http.Client + SecretPath string + DataDir string + TemplatePath string + MihomoBinary string + Client *http.Client + replaceLastGood func(string, string) error + directorySync func(string) error } type LifecycleConfig struct { @@ -42,16 +47,17 @@ type LifecycleConfig struct { // PublishCandidate performs one Stage 1 update. Starting or reloading Mihomo is // deliberately left to the lifecycle stage. func PublishCandidate(ctx context.Context, config CandidateConfig) error { - dataDir := filepath.Clean(config.DataDir) - if !filepath.IsAbs(dataDir) || dataDir == string(filepath.Separator) { - return fmt.Errorf("unsafe data directory %q", config.DataDir) - } - if err := ensureDirectory(dataDir); err != nil { - return err - } + return withDataDirLock(config.DataDir, func(dataDir string) error { + config.DataDir = dataDir + return publishCandidateLocked(ctx, config) + }) +} + +func publishCandidateLocked(ctx context.Context, config CandidateConfig) error { + dataDir := config.DataDir generations := filepath.Join(dataDir, "generations") if err := ensureDirectory(generations); err != nil { - return err + return persistenceError("prepare generations directory", err) } for path, label := range map[string]string{ config.TemplatePath: "Mihomo template", @@ -97,20 +103,20 @@ func PublishCandidate(ctx context.Context, config CandidateConfig) error { } candidate, err := os.MkdirTemp(generations, ".candidate-") if err != nil { - return fmt.Errorf("create candidate generation: %w", err) + return persistenceError("create candidate generation", err) } if err := os.Chmod(candidate, 0o700); err != nil { _ = os.RemoveAll(candidate) - return fmt.Errorf("secure candidate generation: %w", err) + return persistenceError("secure candidate generation", err) } defer os.RemoveAll(candidate) configPath := filepath.Join(candidate, "config.yaml") if err := writePrivateFile(configPath, generated); err != nil { - return fmt.Errorf("write candidate config: %w", err) + return persistenceError("write candidate config", err) } if err := writePrivateFile(filepath.Join(candidate, "subscription.yaml"), subscription); err != nil { - return fmt.Errorf("write candidate subscription: %w", err) + return persistenceError("write candidate subscription", err) } if err := validateMihomoConfig(ctx, config.MihomoBinary, candidate, configPath); err != nil { return errors.New("candidate configuration failed Mihomo validation") @@ -118,24 +124,20 @@ func PublishCandidate(ctx context.Context, config CandidateConfig) error { slot := filepath.Join(dataDir, filepath.FromSlash(next)) if err := os.RemoveAll(slot); err != nil { - return fmt.Errorf("clear inactive generation: %w", err) + return persistenceError("clear inactive generation", err) } if err := os.Rename(candidate, slot); err != nil { - return fmt.Errorf("publish candidate generation: %w", err) + return persistenceError("publish candidate generation", err) } - if err := syncDirectory(generations); err != nil { - return err + if err := config.syncDirectory(generations); err != nil { + return persistenceError("sync generations directory", err) } - if err := replaceSymlink(filepath.Join(dataDir, "last-good"), next); err != nil { - return err + if err := config.replacePointer(filepath.Join(dataDir, "last-good"), next); err != nil { + return persistenceError("publish last-good pointer", err) } - if err := syncDirectory(dataDir); err != nil { - if current == "" { - _ = os.Remove(filepath.Join(dataDir, "last-good")) - } else { - _ = replaceSymlink(filepath.Join(dataDir, "last-good"), current) - } - return err + if err := config.syncDirectory(dataDir); err != nil { + rollbackErr := restoreLastGood(config, dataDir, current) + return errors.Join(persistenceError("sync data directory", err), rollbackErr) } return nil } @@ -153,13 +155,24 @@ func Run(ctx context.Context, config LifecycleConfig) error { return err } - generation, valid := validLastGood(ctx, config.Candidate) + var generation string + valid := withDataDirLock(config.Candidate.DataDir, func(dataDir string) error { + config.Candidate.DataDir = dataDir + var err error + generation, err = validLastGoodLocked(ctx, config.Candidate) + return err + }) warm := valid == nil if !warm { if err := PublishCandidate(ctx, config.Candidate); err != nil { return fmt.Errorf("cold-start candidate failed: %w", err) } - generation, valid = validLastGood(ctx, config.Candidate) + valid = withDataDirLock(config.Candidate.DataDir, func(dataDir string) error { + config.Candidate.DataDir = dataDir + var err error + generation, err = validLastGoodLocked(ctx, config.Candidate) + return err + }) if valid != nil { return fmt.Errorf("published candidate is invalid: %w", valid) } @@ -181,9 +194,16 @@ func Run(ctx context.Context, config LifecycleConfig) error { return err } log.Printf("bootstrap: Mihomo started config=%s update_interval=%s", filepath.Base(generation), config.UpdateInterval) + failClosed := func(err error) error { + cancel() + <-exit + return fmt.Errorf("fatal update stopped Mihomo: %w", err) + } if warm { - updateAndReload(ctx, client, controllerURL, config.Candidate) + if err := updateAndReload(ctx, client, controllerURL, config.Candidate); err != nil { + return failClosed(err) + } } ticker := time.NewTicker(config.UpdateInterval) defer ticker.Stop() @@ -199,14 +219,18 @@ func Run(ctx context.Context, config LifecycleConfig) error { } return fmt.Errorf("Mihomo exited: %w", err) case <-ticker.C: - updateAndReload(ctx, client, controllerURL, config.Candidate) + if err := updateAndReload(ctx, client, controllerURL, config.Candidate); err != nil { + return failClosed(err) + } case <-config.Trigger: - updateAndReload(ctx, client, controllerURL, config.Candidate) + if err := updateAndReload(ctx, client, controllerURL, config.Candidate); err != nil { + return failClosed(err) + } } } } -func validLastGood(ctx context.Context, config CandidateConfig) (string, error) { +func validLastGoodLocked(ctx context.Context, config CandidateConfig) (string, error) { target, err := currentGeneration(filepath.Join(config.DataDir, "last-good")) if err != nil { return "", err @@ -229,34 +253,40 @@ func validLastGood(ctx context.Context, config CandidateConfig) (string, error) return directory, nil } -func updateAndReload(ctx context.Context, client *http.Client, controllerURL string, config CandidateConfig) { - previous, err := currentGeneration(filepath.Join(config.DataDir, "last-good")) - if err != nil || previous == "" { - log.Print("bootstrap: update skipped; last-good is unavailable") - return - } - if err := PublishCandidate(ctx, config); err != nil { - log.Printf("bootstrap: update rejected; keeping last-good: %v", err) - return - } - next, err := currentGeneration(filepath.Join(config.DataDir, "last-good")) - if err == nil { - err = reloadMihomo(ctx, client, controllerURL, filepath.Join(config.DataDir, filepath.FromSlash(next), "config.yaml")) - } - if err == nil { - log.Print("bootstrap: configuration updated and reloaded") - return - } - if rollbackErr := replaceSymlink(filepath.Join(config.DataDir, "last-good"), previous); rollbackErr != nil { - log.Print("bootstrap: reload rejected; could not restore last-good pointer") - return - } - _ = syncDirectory(config.DataDir) - if rollbackErr := reloadMihomo(ctx, client, controllerURL, filepath.Join(config.DataDir, filepath.FromSlash(previous), "config.yaml")); rollbackErr != nil { - log.Print("bootstrap: reload rejected; last-good restored but reload confirmation failed") - return - } - log.Printf("bootstrap: reload rejected; restored and reloaded last-good: %v", err) +func updateAndReload(ctx context.Context, client *http.Client, controllerURL string, config CandidateConfig) error { + return withDataDirLock(config.DataDir, func(dataDir string) error { + config.DataDir = dataDir + previous, err := currentGeneration(filepath.Join(dataDir, "last-good")) + if err != nil || previous == "" { + return errors.Join(errors.New("last-good is unavailable during update"), err) + } + if err := publishCandidateLocked(ctx, config); err != nil { + if errors.Is(err, errPersistence) { + return err + } + log.Printf("bootstrap: update rejected; keeping last-good: %v", err) + return nil + } + next, err := currentGeneration(filepath.Join(dataDir, "last-good")) + if err == nil { + err = reloadMihomo(ctx, client, controllerURL, filepath.Join(dataDir, filepath.FromSlash(next), "config.yaml")) + } + if err == nil { + log.Print("bootstrap: configuration updated and reloaded") + return nil + } + if rollbackErr := config.replacePointer(filepath.Join(dataDir, "last-good"), previous); rollbackErr != nil { + return fmt.Errorf("restore last-good pointer after reload failure: %w", rollbackErr) + } + if rollbackErr := config.syncDirectory(dataDir); rollbackErr != nil { + return fmt.Errorf("persist restored last-good pointer after reload failure: %w", rollbackErr) + } + if rollbackErr := reloadMihomo(ctx, client, controllerURL, filepath.Join(dataDir, filepath.FromSlash(previous), "config.yaml")); rollbackErr != nil { + return fmt.Errorf("reload restored last-good after reload failure: %w", rollbackErr) + } + log.Printf("bootstrap: reload rejected; restored and reloaded last-good: %v", err) + return nil + }) } func reloadMihomo(ctx context.Context, client *http.Client, controllerURL, configPath string) error { @@ -318,6 +348,45 @@ func validateControllerURL(raw string) error { return nil } +func withDataDirLock(dataDir string, operation func(string) error) error { + dataDir = filepath.Clean(dataDir) + if !filepath.IsAbs(dataDir) || dataDir == string(filepath.Separator) { + return fmt.Errorf("unsafe data directory %q", dataDir) + } + if err := ensureDirectory(dataDir); err != nil { + return err + } + lock, err := os.OpenFile(filepath.Join(dataDir, ".bootstrap.lock"), os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return fmt.Errorf("open data directory lock: %w", err) + } + if err := lock.Chmod(0o600); err != nil { + _ = lock.Close() + return fmt.Errorf("secure data directory lock: %w", err) + } + for { + err = syscall.Flock(int(lock.Fd()), syscall.LOCK_EX) + if !errors.Is(err, syscall.EINTR) { + break + } + } + if err != nil { + _ = lock.Close() + return fmt.Errorf("lock data directory: %w", err) + } + + operationErr := operation(dataDir) + unlockErr := syscall.Flock(int(lock.Fd()), syscall.LOCK_UN) + closeErr := lock.Close() + if unlockErr != nil { + unlockErr = fmt.Errorf("unlock data directory: %w", unlockErr) + } + if closeErr != nil { + closeErr = fmt.Errorf("close data directory lock: %w", closeErr) + } + return errors.Join(operationErr, unlockErr, closeErr) +} + func ensureDirectory(path string) error { info, err := os.Lstat(path) if errors.Is(err, os.ErrNotExist) { @@ -335,6 +404,44 @@ func ensureDirectory(path string) error { return nil } +func (config CandidateConfig) replacePointer(path, target string) error { + if config.replaceLastGood != nil { + return config.replaceLastGood(path, target) + } + return replaceSymlink(path, target) +} + +func (config CandidateConfig) syncDirectory(path string) error { + if config.directorySync != nil { + return config.directorySync(path) + } + return syncDirectory(path) +} + +func restoreLastGood(config CandidateConfig, dataDir, target string) error { + path := filepath.Join(dataDir, "last-good") + var err error + if target == "" { + err = os.Remove(path) + if errors.Is(err, os.ErrNotExist) { + err = nil + } + } else { + err = config.replacePointer(path, target) + } + if err != nil { + return persistenceError("restore last-good pointer", err) + } + if err := config.syncDirectory(dataDir); err != nil { + return persistenceError("sync restored last-good pointer", err) + } + return nil +} + +func persistenceError(action string, err error) error { + return fmt.Errorf("%w: %s: %w", errPersistence, action, err) +} + func readSubscriptionSecret(path string) (string, error) { info, err := os.Lstat(path) if err != nil { diff --git a/internal/bootstrap/config_contract_test.go b/internal/bootstrap/config_contract_test.go index 4d7275a..85c0755 100644 --- a/internal/bootstrap/config_contract_test.go +++ b/internal/bootstrap/config_contract_test.go @@ -101,3 +101,15 @@ func TestMihomoTemplateAndRuntimeAssetsArePinned(t *testing.T) { } } } + +func TestDockerBuildContextExcludesSubscriptionSecret(t *testing.T) { + t.Parallel() + + content, err := os.ReadFile("../../.dockerignore") + if err != nil { + t.Fatalf("read .dockerignore: %v", err) + } + if !strings.Contains("\n"+string(content)+"\n", "\nsubscription.url\n") { + t.Fatal(".dockerignore does not exclude subscription.url") + } +} diff --git a/internal/bootstrap/lifecycle_test.go b/internal/bootstrap/lifecycle_test.go index 5552ea2..6f4c9da 100644 --- a/internal/bootstrap/lifecycle_test.go +++ b/internal/bootstrap/lifecycle_test.go @@ -81,7 +81,7 @@ func TestRunReloadFailureRestoresLastGood(t *testing.T) { } wantTarget := readLastGood(t, fixture.config.DataDir) fixture.setSubscription("rejected-node", http.StatusOK) - fixture.failNextReload.Store(true) + fixture.reloadFailures.Store(1) ctx, cancel := context.WithCancel(context.Background()) result := make(chan error, 1) @@ -98,23 +98,125 @@ func TestRunReloadFailureRestoresLastGood(t *testing.T) { } } +func TestRunAndCandidateSerializeDataDirUpdates(t *testing.T) { + fixture := newLifecycleFixture(t) + if err := PublishCandidate(context.Background(), fixture.config); err != nil { + t.Fatal(err) + } + fixture.setSubscription("second-node", http.StatusOK) + fixture.blockNextSubscription.Store(true) + + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- Run(ctx, fixture.lifecycle(nil)) }() + fixture.waitStarted(t) + select { + case <-fixture.subscriptionEntered: + case <-time.After(5 * time.Second): + close(fixture.subscriptionRelease) + t.Fatal("Run did not enter the locked candidate update") + } + + candidateResult := make(chan error, 1) + go func() { candidateResult <- PublishCandidate(context.Background(), fixture.config) }() + select { + case err := <-candidateResult: + close(fixture.subscriptionRelease) + t.Fatalf("concurrent candidate bypassed the data lock: %v", err) + case <-time.After(100 * time.Millisecond): + } + close(fixture.subscriptionRelease) + if err := <-candidateResult; err != nil { + t.Fatalf("concurrent PublishCandidate() error = %v", err) + } + assertContains(t, filepath.Join(fixture.config.DataDir, "last-good", "subscription.yaml"), "second-node") + + cancel() + if err := <-result; !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want context cancellation", err) + } +} + +func TestRunRollbackPersistenceFailureStopsMihomo(t *testing.T) { + fixture := newLifecycleFixture(t) + if err := PublishCandidate(context.Background(), fixture.config); err != nil { + t.Fatal(err) + } + wantTarget := readLastGood(t, fixture.config.DataDir) + fixture.setSubscription("rejected-node", http.StatusOK) + fixture.reloadFailures.Store(1) + var syncCalls atomic.Int32 + fixture.config.directorySync = func(path string) error { + if syncCalls.Add(1) == 3 { + return errors.New("injected rollback sync failure") + } + return syncDirectory(path) + } + + result := make(chan error, 1) + go func() { result <- Run(context.Background(), fixture.lifecycle(nil)) }() + fixture.waitStarted(t) + err := waitResult(t, result) + if !strings.Contains(err.Error(), "persist restored last-good pointer") { + t.Fatalf("Run() error = %v, want rollback persistence failure", err) + } + if got := readLastGood(t, fixture.config.DataDir); got != wantTarget { + t.Fatalf("last-good = %q, want restored %q", got, wantTarget) + } + fixture.waitStopped(t) +} + +func TestRunSecondReloadFailureStopsMihomo(t *testing.T) { + fixture := newLifecycleFixture(t) + if err := PublishCandidate(context.Background(), fixture.config); err != nil { + t.Fatal(err) + } + wantTarget := readLastGood(t, fixture.config.DataDir) + fixture.setSubscription("rejected-node", http.StatusOK) + fixture.reloadFailures.Store(2) + + result := make(chan error, 1) + go func() { result <- Run(context.Background(), fixture.lifecycle(nil)) }() + fixture.waitStarted(t) + err := waitResult(t, result) + if !strings.Contains(err.Error(), "reload restored last-good") { + t.Fatalf("Run() error = %v, want second reload failure", err) + } + if got := readLastGood(t, fixture.config.DataDir); got != wantTarget { + t.Fatalf("last-good = %q, want restored %q", got, wantTarget) + } + fixture.waitStopped(t) +} + type lifecycleFixture struct { - config CandidateConfig - controllerURL string - startedPath string - lock sync.RWMutex - response string - status int - subscriptionRequests atomic.Int32 - reloadRequests atomic.Int32 - failNextReload atomic.Bool + config CandidateConfig + controllerURL string + startedPath string + lock sync.RWMutex + response string + status int + subscriptionRequests atomic.Int32 + reloadRequests atomic.Int32 + reloadFailures atomic.Int32 + blockNextSubscription atomic.Bool + subscriptionEntered chan struct{} + subscriptionRelease chan struct{} } func newLifecycleFixture(t *testing.T) *lifecycleFixture { t.Helper() - fixture := &lifecycleFixture{response: fullSubscription("first-node"), status: http.StatusOK} + fixture := &lifecycleFixture{ + response: fullSubscription("first-node"), + status: http.StatusOK, + subscriptionEntered: make(chan struct{}), + subscriptionRelease: make(chan struct{}), + } subscription := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { fixture.subscriptionRequests.Add(1) + if fixture.blockNextSubscription.CompareAndSwap(true, false) { + close(fixture.subscriptionEntered) + <-fixture.subscriptionRelease + } fixture.lock.RLock() defer fixture.lock.RUnlock() writer.WriteHeader(fixture.status) @@ -127,7 +229,7 @@ func newLifecycleFixture(t *testing.T) *lifecycleFixture { writer.WriteHeader(http.StatusOK) case request.Method == http.MethodPut && request.URL.Path == "/configs": fixture.reloadRequests.Add(1) - if fixture.failNextReload.CompareAndSwap(true, false) { + if fixture.consumeReloadFailure() { http.Error(writer, "rejected", http.StatusInternalServerError) return } @@ -159,7 +261,7 @@ if [ "${1:-}" = -t ]; then ! grep -F reject-validation "$directory/subscription.yaml" >/dev/null exit 0 fi -printf started > %q +printf '%%s' $$ > %q trap 'exit 0' TERM INT while :; do sleep 1; done `, fixture.startedPath)) @@ -200,6 +302,47 @@ func (fixture *lifecycleFixture) waitStarted(t *testing.T) { }) } +func (fixture *lifecycleFixture) waitStopped(t *testing.T) { + t.Helper() + waitFor(t, func() bool { + pidBytes, err := os.ReadFile(fixture.startedPath) + if err != nil { + return false + } + var pid int + if _, err := fmt.Sscanf(string(pidBytes), "%d", &pid); err != nil { + return false + } + return errors.Is(syscall.Kill(pid, 0), syscall.ESRCH) + }) +} + +func (fixture *lifecycleFixture) consumeReloadFailure() bool { + for { + remaining := fixture.reloadFailures.Load() + if remaining == 0 { + return false + } + if fixture.reloadFailures.CompareAndSwap(remaining, remaining-1) { + return true + } + } +} + +func waitResult(t *testing.T, result <-chan error) error { + t.Helper() + select { + case err := <-result: + if err == nil { + t.Fatal("Run() error = nil") + } + return err + case <-time.After(5 * time.Second): + t.Fatal("Run did not fail closed") + return nil + } +} + func waitFor(t *testing.T, condition func() bool) { t.Helper() deadline := time.Now().Add(5 * time.Second) diff --git a/tests/container-smoke.sh b/tests/container-smoke.sh index 5e6d04c..a48f2c6 100755 --- a/tests/container-smoke.sh +++ b/tests/container-smoke.sh @@ -163,6 +163,16 @@ docker kill --signal HUP "$container" >/dev/null wait_for_last_good second-node wait_for_log 'configuration updated and reloaded' host_curl "http://127.0.0.1:${controller_port}/providers/proxies/subscription" >/dev/null +selection_group='%F0%9F%9A%80%20%E8%8A%82%E7%82%B9%E9%80%89%E6%8B%A9' +group=$(host_curl "http://127.0.0.1:${controller_port}/proxies/${selection_group}") +printf '%s\n' "$group" | grep -F '"DIRECT"' >/dev/null +host_curl \ + --request PUT \ + --header 'Content-Type: application/json' \ + --data '{"name":"DIRECT"}' \ + "http://127.0.0.1:${controller_port}/proxies/${selection_group}" >/dev/null +host_curl "http://127.0.0.1:${controller_port}/proxies/${selection_group}" \ + | grep -F '"now":"DIRECT"' >/dev/null docker exec "$provider" /bin/sh -c 'printf "proxies: [" > /tmp/web/provider.yaml' docker restart "$container" >/dev/null @@ -174,4 +184,4 @@ if docker logs "$container" 2>&1 | grep -F "$secret" >/dev/null; then fi docker exec "$container" /usr/local/bin/mihomo -t -d /data/last-good -f /data/last-good/config.yaml >/dev/null -echo "container smoke test passed: cold fail-closed, warm recovery, HUP update, rollback, 7890, 9090, and ExternalUI" +echo "container smoke test passed: cold fail-closed, warm recovery, HUP update, rollback, 7890, 9090, ExternalUI, and proxy switching"