检查点:保存双层证书协调进展并准备新主机验收

This commit is contained in:
xuans 2026-09-20 01:02:16 +08:00
parent 48cee12c4a
commit 2c0d0e8eda
15 changed files with 1190 additions and 9 deletions

View File

@ -41,6 +41,8 @@ type App struct {
HTTP *httpapi.Server
maintenanceStop context.CancelFunc
maintenanceDone chan struct{}
managementStop context.CancelFunc
managementDone chan struct{}
}
const (
@ -180,7 +182,13 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string
Tasks: taskRepo,
ReadinessCheck: db.DB.PingContext,
})
management.activator = directManagementActivator{server: server}
activator, recoverManagement, err := newManagementActivator(cfg, server)
if err != nil {
_ = db.Close()
_ = logger.Close()
return nil, fmt.Errorf("初始化管理证书协调器失败:%w", err)
}
management.activator = activator
application := &App{
Config: cfg, Logger: logger, Database: db, Repository: repo,
Servers: serverRepo,
@ -193,6 +201,7 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string
Renewal: renewal,
}
application.startAuditCleanup(auditRepo)
application.startManagementRecovery(recoverManagement)
return application, nil
}
@ -228,6 +237,9 @@ func (a *App) Close() error {
return nil
}
var first error
if a.managementStop != nil {
a.managementStop()
}
if a.Renewal != nil {
a.Renewal.Stop()
}
@ -237,6 +249,9 @@ func (a *App) Close() error {
if a.maintenanceDone != nil {
<-a.maintenanceDone
}
if a.managementDone != nil {
<-a.managementDone
}
if a.HTTP != nil {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
if err := a.HTTP.Shutdown(ctx); err != nil && first == nil {

View File

@ -0,0 +1,108 @@
package app
import (
"context"
"errors"
"os"
"path/filepath"
"reflect"
"time"
"git.pcux.cn/xuans/Scert/backend/internal/certificate"
"git.pcux.cn/xuans/Scert/backend/internal/config"
"git.pcux.cn/xuans/Scert/backend/internal/httpapi"
"git.pcux.cn/xuans/Scert/backend/internal/nginxactivation"
)
const managementActivationSocket = "/run/scert-nginx/activation.sock"
type coordinatedManagementActivator struct {
server *httpapi.Server
directory string
proxy nginxactivation.ProxyClient
}
func newManagementActivator(cfg config.Config, server *httpapi.Server) (managementActivator, func(context.Context) error, error) {
directory := filepath.Join(cfg.Paths.Data, "management-activation")
if err := os.MkdirAll(directory, 0o700); err != nil {
return nil, nil, err
}
activator := coordinatedManagementActivator{
server: server, directory: directory,
proxy: nginxactivation.Client{SocketPath: managementActivationSocket, ServiceUID: 0},
}
return activator, activator.coordinator(nil).Recover, nil
}
func (a coordinatedManagementActivator) coordinator(expected *httpapi.ManagementCertificateState) nginxactivation.Coordinator {
return nginxactivation.Coordinator{
Directory: a.directory, Proxy: a.proxy,
Upstream: managementUpstream{server: a.server, expected: expected},
}
}
func (a coordinatedManagementActivator) Ready(ctx context.Context, state httpapi.ManagementCertificateState) error {
if state.HTTPSMode == "direct" {
if err := a.coordinator(nil).Recover(ctx); err != nil {
return err
}
return (directManagementActivator{server: a.server}).Ready(ctx, state)
}
return a.coordinator(&state).Ready(ctx)
}
func (a coordinatedManagementActivator) Activate(ctx context.Context, state httpapi.ManagementCertificateState, saved certificate.SavedResource) error {
if state.HTTPSMode == "direct" {
if err := a.Ready(ctx, state); err != nil {
return err
}
return (directManagementActivator{server: a.server}).Activate(ctx, state, saved)
}
return a.coordinator(&state).Activate(ctx, nginxactivation.CertificateFiles{
CertificatePath: saved.FullChainPath, PrivateKeyPath: saved.PrivateKeyPath,
})
}
type managementUpstream struct {
server *httpapi.Server
expected *httpapi.ManagementCertificateState
}
func (u managementUpstream) WithCertificate(ctx context.Context, operation func(nginxactivation.ManagedSession) error) error {
return u.server.WithManagementCertificate(ctx, func(session *httpapi.ManagementCertificateSession) error {
return operation(&managementSession{session: session, expected: u.expected})
})
}
type managementSession struct {
session *httpapi.ManagementCertificateSession
expected *httpapi.ManagementCertificateState
previous *httpapi.ManagementSnapshot
}
func (s *managementSession) Snapshot() (nginxactivation.ManagedCertificate, error) {
current, err := s.session.Snapshot()
if err != nil {
return nginxactivation.ManagedCertificate{}, err
}
if s.previous == nil && s.expected != nil {
expected, actual := *s.expected, current.State
expected.NotAfter, actual.NotAfter = time.Time{}, time.Time{}
if !reflect.DeepEqual(expected, actual) {
return nginxactivation.ManagedCertificate{}, errors.New("签发期间管理配置已变化,拒绝开始双层激活")
}
}
s.previous = &current
return nginxactivation.ManagedCertificate{
CertificateFiles: nginxactivation.CertificateFiles{CertificatePath: current.State.CertificatePath, PrivateKeyPath: current.State.PrivateKeyPath},
Hostname: current.State.Hostname, Mode: current.State.HTTPSMode, LeafHash: current.LeafHash,
ProxyAddress: current.ProxyAddress, UpstreamAddress: current.UpstreamAddress,
}, nil
}
func (s *managementSession) Activate(files nginxactivation.CertificateFiles, hash string) error {
if s.previous == nil {
return errors.New("管理证书协调缺少切换前快照")
}
return s.session.Activate(*s.previous, certificate.SavedResource{FullChainPath: files.CertificatePath, PrivateKeyPath: files.PrivateKeyPath}, hash)
}

View File

@ -0,0 +1,14 @@
//go:build !linux
package app
import (
"context"
"git.pcux.cn/xuans/Scert/backend/internal/config"
"git.pcux.cn/xuans/Scert/backend/internal/httpapi"
)
func newManagementActivator(_ config.Config, server *httpapi.Server) (managementActivator, func(context.Context) error, error) {
return directManagementActivator{server: server}, nil, nil
}

View File

@ -0,0 +1,32 @@
package app
import (
"context"
"time"
)
func (a *App) startManagementRecovery(recover func(context.Context) error) {
if recover == nil {
return
}
ctx, cancel := context.WithCancel(context.Background())
a.managementStop = cancel
a.managementDone = make(chan struct{})
go func() {
defer close(a.managementDone)
for {
attempt, stop := context.WithTimeout(ctx, 100*time.Second)
err := recover(attempt)
stop()
if ctx.Err() != nil {
return
}
if err != nil && a.Logger != nil {
a.Logger.Error("管理证书协调恢复尚未完成,保持续期保护", "error", err)
}
if err := waitForManagementRetry(ctx, time.Minute); err != nil {
return
}
}
}()
}

View File

@ -82,11 +82,11 @@ func runApplication(ctx context.Context, cfg config.Config, path string) (bool,
return false, err
}
defer service.Close()
errorsChannel := make(chan error, 1)
go func() { errorsChannel <- service.HTTP.ListenAndServe() }()
if service.Renewal != nil {
service.Renewal.Start(ctx)
}
errorsChannel := make(chan error, 1)
go func() { errorsChannel <- service.HTTP.ListenAndServe() }()
select {
case <-ctx.Done():
return false, nil

View File

@ -0,0 +1,121 @@
package httpapi
import (
"bytes"
"context"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"net"
"strconv"
"sync"
"git.pcux.cn/xuans/Scert/backend/internal/certificate"
"git.pcux.cn/xuans/Scert/backend/internal/config"
)
// ManagementSnapshot 同时证明持久化路径、磁盘材料与当前 TLS 快照一致。
type ManagementSnapshot struct {
State ManagementCertificateState
LeafHash string
UpstreamAddress string
ProxyAddress string
}
// ManagementCertificateSession 只在独占回调内有效,不持有配置读锁跨越网络探测。
type ManagementCertificateSession struct {
mu sync.Mutex
server *Server
}
func (s *Server) WithManagementCertificate(ctx context.Context, operation func(*ManagementCertificateSession) error) error {
if err := ctx.Err(); err != nil {
return err
}
if operation == nil {
return errors.New("管理证书操作不能为空")
}
if !s.tlsMu.TryLock() {
return errors.New("管理证书或配置正在更新,请稍后重试")
}
defer s.tlsMu.Unlock()
session := &ManagementCertificateSession{server: s}
defer func() {
session.mu.Lock()
session.server = nil
session.mu.Unlock()
}()
return operation(session)
}
func (session *ManagementCertificateSession) Snapshot() (ManagementSnapshot, error) {
session.mu.Lock()
defer session.mu.Unlock()
if session.server == nil {
return ManagementSnapshot{}, errors.New("管理证书独占操作已经结束")
}
return session.server.managementSnapshotLocked()
}
func (session *ManagementCertificateSession) Activate(expected ManagementSnapshot, saved certificate.SavedResource, candidateHash string) error {
session.mu.Lock()
defer session.mu.Unlock()
if session.server == nil {
return errors.New("管理证书独占操作已经结束")
}
if candidateHash == "" {
return errors.New("管理证书协调缺少候选摘要")
}
current, err := session.server.managementSnapshotLocked()
if err != nil {
return err
}
if current.LeafHash != expected.LeafHash || current.UpstreamAddress != expected.UpstreamAddress || current.ProxyAddress != expected.ProxyAddress {
return errors.New("管理证书或监听配置在协调期间发生变化")
}
return session.server.activateTLSCertificateLocked(saved.FullChainPath, saved.PrivateKeyPath, &expected.State, candidateHash)
}
func (s *Server) managementSnapshotLocked() (ManagementSnapshot, error) {
s.configMu.RLock()
defer s.configMu.RUnlock()
var snapshot ManagementSnapshot
if !s.config.SetupReady || s.configPath == "" {
return snapshot, errors.New("管理证书协调需要已初始化的持久化配置")
}
persisted, err := config.Load(s.configPath)
if err != nil {
return snapshot, fmt.Errorf("读取管理证书持久化配置失败:%w", err)
}
state := s.managementCertificateState()
if persisted.Server.TLSCert != state.CertificatePath || persisted.Server.TLSKey != state.PrivateKeyPath {
return snapshot, errors.New("管理证书持久化路径与运行配置不一致,请先恢复服务")
}
pair, err := tls.LoadX509KeyPair(state.CertificatePath, state.PrivateKeyPath)
if err != nil {
return snapshot, fmt.Errorf("读取管理证书材料失败:%w", err)
}
live := s.tlsCertificate.Load()
if live == nil || len(live.Certificate) == 0 || !bytes.Equal(live.Certificate[0], pair.Certificate[0]) {
return snapshot, errors.New("管理证书磁盘材料与运行证书不一致,请先恢复服务")
}
leaf, err := x509.ParseCertificate(pair.Certificate[0])
if err != nil {
return snapshot, err
}
address, err := listenAddress(s.config)
if err != nil {
return snapshot, err
}
port := s.config.Server.ProxyHTTPSPort
if port == 0 {
port = 443
}
state.NotAfter = leaf.NotAfter
return ManagementSnapshot{
State: state, LeafHash: fmt.Sprintf("%x", sha256.Sum256(leaf.Raw)),
UpstreamAddress: address, ProxyAddress: net.JoinHostPort("127.0.0.1", strconv.Itoa(port)),
}, nil
}

View File

@ -0,0 +1,167 @@
package httpapi
import (
"context"
"crypto/sha256"
"crypto/tls"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"time"
"git.pcux.cn/xuans/Scert/backend/internal/certificate"
"git.pcux.cn/xuans/Scert/backend/internal/config"
)
func TestManagementSessionPinsCandidateAndExpires(t *testing.T) {
f := newRotationFixture(t)
certPath, keyPath := f.issue(t, 3, "management.example.test", time.Now().Add(-time.Hour), time.Now().Add(30*time.Hour))
pair, err := tls.LoadX509KeyPair(certPath, keyPath)
if err != nil {
t.Fatal(err)
}
hash := fmt.Sprintf("%x", sha256.Sum256(pair.Certificate[0]))
var escaped *ManagementCertificateSession
err = f.server.WithManagementCertificate(t.Context(), func(session *ManagementCertificateSession) error {
escaped = session
previous, err := session.Snapshot()
if err != nil {
return err
}
if err := f.server.WithManagementCertificate(t.Context(), func(*ManagementCertificateSession) error {
t.Fatal("独占期间不能开始第二次协调")
return nil
}); err == nil {
t.Fatal("重入协调未被拒绝")
}
saved := certificate.SavedResource{FullChainPath: certPath, PrivateKeyPath: keyPath}
if err := session.Activate(previous, saved, previous.LeafHash); err == nil {
t.Fatal("错误的候选摘要不能切换证书")
}
if err := session.Activate(previous, saved, hash); err != nil {
return err
}
current, err := session.Snapshot()
if err != nil {
return err
}
if current.LeafHash != hash || current.State.CertificatePath != certPath {
t.Fatal("切换后快照没有反映候选")
}
return nil
})
if err != nil {
t.Fatal(err)
}
if _, err := escaped.Snapshot(); err == nil {
t.Fatal("回调结束后不能读取独占快照")
}
if err := escaped.Activate(ManagementSnapshot{}, certificate.SavedResource{}, hash); err == nil {
t.Fatal("回调结束后不能切换证书")
}
if serial, err := f.serial(); err != nil || serial != 3 {
t.Fatal("真实 TLS 未切换", serial, err)
}
}
func TestManagementSnapshotRejectsDivergence(t *testing.T) {
for _, layer := range []string{"config", "material", "unstarted"} {
t.Run(layer, func(t *testing.T) {
f := newRotationFixture(t)
certPath, keyPath := f.issue(t, 3, "management.example.test", time.Now().Add(-time.Hour), time.Now().Add(30*time.Hour))
switch layer {
case "config":
cfg, err := config.Load(f.server.configPath)
if err != nil {
t.Fatal(err)
}
cfg.Server.TLSCert, cfg.Server.TLSKey = certPath, keyPath
if err := cfg.Save(f.server.configPath); err != nil {
t.Fatal(err)
}
case "material":
for destination, source := range map[string]string{f.firstCertificate: certPath, f.firstKey: keyPath} {
data, err := os.ReadFile(source)
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(destination, data, 0o600); err != nil {
t.Fatal(err)
}
}
case "unstarted":
f.server.tlsCertificate.Store(nil)
}
if err := f.server.WithManagementCertificate(t.Context(), func(session *ManagementCertificateSession) error {
_, err := session.Snapshot()
return err
}); err == nil {
t.Fatal("配置、材料与运行快照不一致时不能报告可协调")
}
})
}
}
func TestManagementSessionDoesNotBlockTLSOrReadiness(t *testing.T) {
f := newRotationFixture(t)
f.server.secrets = readinessStore(t)
f.server.configMu.Lock()
f.server.config.DNS.Provider = "aliyun"
f.server.config.DNS.CredentialRef = readinessDNSReference
f.server.configMu.Unlock()
if err := f.server.WithManagementCertificate(t.Context(), func(session *ManagementCertificateSession) error {
if _, err := session.Snapshot(); err != nil {
return err
}
if serial, err := f.serial(); err != nil || serial != 2 {
return fmt.Errorf("独占期间握手失败:%d %v", serial, err)
}
request := httptest.NewRequest(http.MethodGet, "/readyz", nil)
request.Host = "management.example.test"
response := httptest.NewRecorder()
f.server.Handler().ServeHTTP(response, request)
if response.Code != http.StatusOK {
return fmt.Errorf("独占期间就绪探测失败:%d %s", response.Code, response.Body.String())
}
return nil
}); err != nil {
t.Fatal(err)
}
}
func TestManagementSessionExcludesConfigurationWriters(t *testing.T) {
f := newRotationFixture(t)
f.server.configMu.Lock()
f.server.config.Admin.PasswordHash = httpUserPasswordHash(t, "AdminPass123")
f.server.configMu.Unlock()
cookie := loginAuthTest(t, f.server, "admin", "AdminPass123", false).Result().Cookies()[0]
if err := f.server.WithManagementCertificate(t.Context(), func(*ManagementCertificateSession) error {
response := performChangePasswordRequest(t, f.server, cookie, `{"current_password":"AdminPass123","new_password":"NewAdmin123","confirm_password":"NewAdmin123"}`)
if response.Code != http.StatusConflict {
t.Fatalf("协调期间改密应拒绝并发保存:%d %s", response.Code, response.Body.String())
}
request := httptest.NewRequest(http.MethodPut, "/api/v1/settings", strings.NewReader(`{}`))
request.Header.Set("Content-Type", "application/json")
request.AddCookie(cookie)
response = httptest.NewRecorder()
f.server.Handler().ServeHTTP(response, request)
if response.Code != http.StatusConflict {
t.Fatalf("协调期间设置应拒绝并发保存:%d %s", response.Code, response.Body.String())
}
return nil
}); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(t.Context())
cancel()
if err := f.server.WithManagementCertificate(ctx, func(*ManagementCertificateSession) error {
t.Fatal("已取消操作不能进入独占区")
return nil
}); !errors.Is(err, context.Canceled) {
t.Fatal(err)
}
}

View File

@ -41,18 +41,22 @@ func (s *Server) changePassword(c *gin.Context) {
s.writeError(c, http.StatusUnauthorized, "登录状态已失效,请重新登录", nil)
return
}
s.configMu.RLock()
cfg := s.config
s.configMu.RUnlock()
if session.UserID == "primary-super-admin" {
s.changePrimarySuperAdminPassword(c, session, cfg, request)
s.changePrimarySuperAdminPassword(c, session, request)
return
}
s.changeDatabaseUserPassword(c, session, request)
}
func (s *Server) changePrimarySuperAdminPassword(c *gin.Context, session authSession, cfg config.Config, request changePasswordRequest) {
func (s *Server) changePrimarySuperAdminPassword(c *gin.Context, session authSession, request changePasswordRequest) {
if !s.tlsMu.TryLock() {
s.writeError(c, http.StatusConflict, "管理证书或配置正在更新,请稍后重试", nil)
return
}
defer s.tlsMu.Unlock()
s.configMu.RLock()
cfg := s.config
s.configMu.RUnlock()
if _, err := auth.AuthenticateAdmin(session.Username, request.CurrentPassword, cfg.Admin.Username, cfg.Admin.Email, cfg.Admin.PasswordHash); err != nil {
s.writeError(c, http.StatusUnauthorized, "当前密码错误", nil)
return

View File

@ -123,6 +123,11 @@ func (s *Server) resetSubmit(c *gin.Context) {
}
password := c.Request.PostForm.Get("new_password")
confirmation := c.Request.PostForm.Get("confirm_password")
if !s.tlsMu.TryLock() {
s.writeResetFailure(c, http.StatusConflict, "管理证书或配置正在更新,请稍后重试")
return
}
defer s.tlsMu.Unlock()
s.configMu.RLock()
cfg := s.config
s.configMu.RUnlock()

View File

@ -180,6 +180,11 @@ func (s *Server) updateSettings(c *gin.Context) {
return
}
if !s.tlsMu.TryLock() {
s.writeError(c, http.StatusConflict, "管理证书或配置正在更新,请稍后重试", nil)
return
}
defer s.tlsMu.Unlock()
s.configMu.Lock()
previous := s.config
updated := s.config

View File

@ -237,6 +237,11 @@ func (s *Server) applySetup(c *gin.Context) {
return
}
defer s.setupMu.Unlock()
if !s.tlsMu.TryLock() {
s.writeError(c, http.StatusConflict, "管理证书或配置正在更新,请稍后重试", nil)
return
}
defer s.tlsMu.Unlock()
s.configMu.RLock()
cfg := s.config
s.configMu.RUnlock()

View File

@ -2,6 +2,7 @@ package httpapi
import (
"context"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"encoding/pem"
@ -74,6 +75,11 @@ func (s *Server) ActivateManagementCertificate(expected ManagementCertificateSta
func (s *Server) activateTLSCertificate(certificatePath, privateKeyPath string, expected *ManagementCertificateState) error {
s.tlsMu.Lock()
defer s.tlsMu.Unlock()
return s.activateTLSCertificateLocked(certificatePath, privateKeyPath, expected, "")
}
// 调用者必须持有 tlsMu,保持双层协调与配置写入之间的独占边界。
func (s *Server) activateTLSCertificateLocked(certificatePath, privateKeyPath string, expected *ManagementCertificateState, expectedHash string) error {
s.configMu.Lock()
defer s.configMu.Unlock()
if expected != nil {
@ -92,6 +98,9 @@ func (s *Server) activateTLSCertificate(certificatePath, privateKeyPath string,
if err != nil {
return fmt.Errorf("加载新的管理证书失败:%w", err)
}
if expectedHash != "" && fmt.Sprintf("%x", sha256.Sum256(pair.Certificate[0])) != expectedHash {
return errors.New("候选管理证书在协调期间发生变化,已拒绝切换")
}
leaf, err := x509.ParseCertificate(pair.Certificate[0])
if err != nil {
return fmt.Errorf("解析新的管理证书失败:%w", err)

View File

@ -0,0 +1,97 @@
package nginxactivation
import (
"bytes"
"encoding/hex"
"encoding/json"
"errors"
"io"
"os"
"path/filepath"
)
const coordinationName = "coordination.json"
type coordinationRecord struct {
Version int `json:"version"`
Phase string `json:"phase"`
BaselineID string `json:"baseline_id,omitempty"`
ProxyID string `json:"proxy_id,omitempty"`
Previous ManagedCertificate `json:"previous"`
Candidate ManagedCertificate `json:"candidate"`
}
func (value ManagedCertificate) validate() error {
hash, err := hex.DecodeString(value.LeafHash)
if err != nil || len(hash) != 32 || hex.EncodeToString(hash) != value.LeafHash || value.Mode != "proxy" {
return errors.New("管理证书协调摘要或模式无效")
}
if err := errors.Join((Endpoint{Address: value.ProxyAddress, Hostname: value.Hostname}).validate(), (Endpoint{Address: value.UpstreamAddress, Hostname: value.Hostname}).validate()); err != nil {
return err
}
if value.ProxyAddress == value.UpstreamAddress {
return errors.New("管理证书代理与上游地址不能相同")
}
for _, path := range []string{value.CertificatePath, value.PrivateKeyPath} {
if !filepath.IsAbs(path) || filepath.Clean(path) != path || path == "/" || len(path) > 4096 {
return errors.New("管理证书协调路径无效")
}
}
return nil
}
func (record coordinationRecord) validate() error {
if record.Version != 1 || (record.BaselineID != "" && !validID(record.BaselineID)) || (record.ProxyID != "" && !validID(record.ProxyID)) {
return errors.New("管理证书协调记录版本或编号无效")
}
switch record.Phase {
case "prepared":
case "staged", "upstream", "committed", "rolled_back":
if record.ProxyID == "" {
return errors.New("管理证书协调记录缺少代理事务编号")
}
default:
return errors.New("管理证书协调记录阶段无效")
}
if err := errors.Join(record.Previous.validate(), record.Candidate.validate()); err != nil {
return err
}
previous, candidate := record.Previous, record.Candidate
if previous.LeafHash == candidate.LeafHash || previous.CertificatePath == candidate.CertificatePath || previous.PrivateKeyPath == candidate.PrivateKeyPath {
return errors.New("候选证书必须使用独立的新版本材料")
}
candidate.CertificateFiles, candidate.LeafHash = previous.CertificateFiles, previous.LeafHash
if candidate != previous || (record.ProxyID != "" && record.ProxyID == record.BaselineID) {
return errors.New("管理证书协调记录包含不一致的策略或事务编号")
}
return nil
}
func readCoordination(root *os.Root) (coordinationRecord, error) {
var record coordinationRecord
data, err := readRegular(root, coordinationName, 64<<10)
if err != nil {
return record, err
}
decoder := json.NewDecoder(bytes.NewReader(data))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&record); err != nil {
return record, err
}
var extra any
if err := decoder.Decode(&extra); err != io.EOF {
return record, errors.New("管理证书协调记录包含多余内容")
}
return record, record.validate()
}
func saveCoordination(root *os.Root, record coordinationRecord) error {
if err := record.validate(); err != nil {
return err
}
data, err := json.Marshal(record)
if err != nil {
return err
}
return writeAtomic(root, coordinationName, data)
}

View File

@ -0,0 +1,268 @@
package nginxactivation
import (
"context"
"crypto/tls"
"errors"
"fmt"
"os"
"path/filepath"
"time"
"git.pcux.cn/xuans/Scert/backend/internal/processlock"
)
var ErrCoordinationPending = errors.New("管理证书协调仍未完成,已阻止再次申请证书")
type CertificateFiles struct {
CertificatePath string `json:"certificate_path"`
PrivateKeyPath string `json:"private_key_path"`
}
type ManagedCertificate struct {
CertificateFiles
Hostname string `json:"hostname"`
Mode string `json:"mode"`
LeafHash string `json:"leaf_hash"`
ProxyAddress string `json:"proxy_address"`
UpstreamAddress string `json:"upstream_address"`
}
type ManagedSession interface {
Snapshot() (ManagedCertificate, error)
Activate(CertificateFiles, string) error
}
// ManagedUpstream 的回调必须独占所有证书切换及相关配置写入,但不能阻塞 TLS 和就绪探测。
type ManagedUpstream interface {
WithCertificate(context.Context, func(ManagedSession) error) error
}
type ProxyClient interface {
Ready(context.Context) (ReadyState, error)
State(context.Context) (State, error)
Stage(context.Context, []byte, []byte) (string, error)
Commit(context.Context, string) error
Reconcile(context.Context, string) (State, error)
}
// Coordinator 的目录归应用用户所有,与 root 激活器的专用目录分离。
type Coordinator struct {
Directory string
Proxy ProxyClient
Upstream ManagedUpstream
}
func (c Coordinator) Ready(ctx context.Context) error {
return c.withSession(ctx, func(root *os.Root, session ManagedSession) error {
if _, err := c.recover(ctx, root, session); err != nil {
return err
}
current, err := session.Snapshot()
if err != nil {
return err
}
return c.ready(ctx, current)
})
}
func (c Coordinator) Recover(ctx context.Context) error {
return c.withSession(ctx, func(root *os.Root, session ManagedSession) error {
_, err := c.recover(ctx, root, session)
return err
})
}
func (c Coordinator) Activate(ctx context.Context, files CertificateFiles) error {
return c.withSession(ctx, func(root *os.Root, session ManagedSession) (resultErr error) {
if _, err := c.recover(ctx, root, session); err != nil {
return err
}
previous, err := session.Snapshot()
if err != nil {
return err
}
if err := c.ready(ctx, previous); err != nil {
return err
}
chain, err := readManagedMaterial(files.CertificatePath, 1<<20)
if err != nil {
return err
}
key, err := readManagedMaterial(files.PrivateKeyPath, 64<<10)
if err != nil {
return err
}
defer clear(key)
pair, err := tls.X509KeyPair(chain, key)
if err != nil {
return errors.New("候选管理证书与私钥不匹配")
}
candidate := previous
candidate.CertificateFiles, candidate.LeafHash = files, leafHash(pair.Certificate[0])
baseline, err := c.Proxy.State(ctx)
if err != nil {
return err
}
record := coordinationRecord{Version: 1, Phase: "prepared", BaselineID: baseline.ID, Previous: previous, Candidate: candidate}
if err := record.validate(); err != nil {
return err
}
if err := saveCoordination(root, record); err != nil {
return err
}
// 请求取消或应答丢失仍在同一独占区内恢复;不能把网络失败当成尚未执行。
defer func() {
if resultErr != nil {
recoveryCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 100*time.Second)
defer cancel()
_, recoveryErr := c.recover(recoveryCtx, root, session)
resultErr = errors.Join(resultErr, recoveryErr)
}
}()
id, err := c.Proxy.Stage(ctx, chain, key)
if err != nil {
return err
}
if !validID(id) || id == record.BaselineID {
return errors.New("代理返回的暂存事务编号无效")
}
record.ProxyID, record.Phase = id, "staged"
if err := saveCoordination(root, record); err != nil {
return err
}
if err := ctx.Err(); err != nil {
return err
}
if err := session.Activate(files, candidate.LeafHash); err != nil {
return err
}
record.Phase = "upstream"
if err := saveCoordination(root, record); err != nil {
return err
}
commitErr := c.Proxy.Commit(ctx, id)
verifyCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 100*time.Second)
defer cancel()
completed, err := c.recover(verifyCtx, root, session)
if err != nil || completed.Phase != "committed" {
return errors.Join(commitErr, err, ErrCoordinationPending)
}
// 只有同一事务终态及实时双层入口均已验证,才接受丢失提交应答后的成功。
return nil
})
}
func (c Coordinator) recover(ctx context.Context, root *os.Root, session ManagedSession) (coordinationRecord, error) {
record, err := readCoordination(root)
if errors.Is(err, os.ErrNotExist) {
return coordinationRecord{}, nil
}
if err != nil {
return record, err
}
if record.Phase == "committed" || record.Phase == "rolled_back" {
return record, nil
}
current, err := session.Snapshot()
if err != nil {
return record, errors.Join(ErrCoordinationPending, err)
}
if current != record.Previous && current != record.Candidate {
return record, errors.Join(ErrCoordinationPending, errors.New("当前上游不属于持久化协调记录,拒绝覆盖"))
}
state, err := c.Proxy.State(ctx)
if err != nil {
return record, errors.Join(ErrCoordinationPending, err)
}
if !validID(state.ID) || state.ID == record.BaselineID ||
(record.ProxyID != "" && state.ID != record.ProxyID) ||
state.PreviousHash != record.Previous.LeafHash || state.CandidateHash != record.Candidate.LeafHash {
// 尚未观测到对应事务时保留意图;延迟到达的暂存请求仍可能执行,不能清除或重放。
return record, errors.Join(ErrCoordinationPending, errors.New("尚未确认对应的代理事务,保留协调记录"))
}
record.ProxyID = state.ID
if state.Status == "pending" || state.Status == "staged" {
state, err = c.Proxy.Reconcile(ctx, state.ID)
if err != nil {
return record, errors.Join(ErrCoordinationPending, err)
}
}
if state.ID != record.ProxyID || state.PreviousHash != record.Previous.LeafHash || state.CandidateHash != record.Candidate.LeafHash {
return record, errors.Join(ErrCoordinationPending, errors.New("代理恢复结果不属于当前事务"))
}
if (state.Status != "committed" || current != record.Candidate) && (state.Status != "rolled_back" || current != record.Previous) {
return record, errors.Join(ErrCoordinationPending, errors.New("代理终态与实际上游证书不一致"))
}
if err := c.ready(ctx, current); err != nil {
return record, errors.Join(ErrCoordinationPending, err)
}
record.Phase = state.Status
return record, saveCoordination(root, record)
}
func (c Coordinator) ready(ctx context.Context, expected ManagedCertificate) error {
if err := expected.validate(); err != nil {
return err
}
ready, err := c.Proxy.Ready(ctx)
if err != nil {
return err
}
if ready.Hostname != expected.Hostname || ready.LeafHash != expected.LeafHash || ready.ProxyAddress != expected.ProxyAddress || ready.UpstreamAddress != expected.UpstreamAddress {
return errors.New("本地激活策略或双层入口与应用管理配置不一致")
}
return nil
}
func (c Coordinator) withSession(ctx context.Context, operation func(*os.Root, ManagedSession) error) error {
if err := ctx.Err(); err != nil {
return err
}
if c.Proxy == nil || c.Upstream == nil {
return errors.New("管理证书协调器尚未配置")
}
if err := validateDirectory(c.Directory); err != nil {
return err
}
lockPath := filepath.Join(c.Directory, ".coordination.lock")
if info, err := os.Lstat(lockPath); err == nil {
if err := validateStateFile(info, 4096); err != nil {
return err
}
} else if !errors.Is(err, os.ErrNotExist) {
return err
}
lock, err := processlock.Acquire(lockPath)
if err != nil {
return err
}
defer lock.Close()
root, err := os.OpenRoot(c.Directory)
if err != nil {
return err
}
defer root.Close()
return c.Upstream.WithCertificate(ctx, func(session ManagedSession) error {
return operation(root, session)
})
}
func readManagedMaterial(path string, limit int64) ([]byte, error) {
if !filepath.IsAbs(path) || filepath.Clean(path) != path {
return nil, errors.New("管理证书材料必须使用规范绝对路径")
}
if err := validateTrustedParents(filepath.Dir(path), uint32(os.Geteuid())); err != nil {
return nil, err
}
root, err := os.OpenRoot(filepath.Dir(path))
if err != nil {
return nil, err
}
defer root.Close()
data, err := readRegular(root, filepath.Base(path), limit)
if err != nil {
return nil, fmt.Errorf("读取候选管理证书材料失败:%w", err)
}
return data, nil
}

View File

@ -0,0 +1,331 @@
package nginxactivation
import (
"context"
"errors"
"os"
"path/filepath"
"sync"
"testing"
"time"
)
type coordinatedTestUpstream struct {
mu sync.Mutex
current ManagedCertificate
readErr error
activate func(CertificateFiles, string) error
}
func (u *coordinatedTestUpstream) WithCertificate(ctx context.Context, operation func(ManagedSession) error) error {
if !u.mu.TryLock() {
return errors.New("测试上游正被独占")
}
defer u.mu.Unlock()
if err := ctx.Err(); err != nil {
return err
}
return operation(u)
}
func (u *coordinatedTestUpstream) Snapshot() (ManagedCertificate, error) {
return u.current, u.readErr
}
func (u *coordinatedTestUpstream) Activate(files CertificateFiles, hash string) error {
return u.activate(files, hash)
}
type coordinatedTestProxy struct {
ProxyClient
stage func(context.Context, []byte, []byte) (string, error)
commit func(context.Context, string) error
}
func (p coordinatedTestProxy) Stage(ctx context.Context, chain, key []byte) (string, error) {
if p.stage != nil {
return p.stage(ctx, chain, key)
}
return p.ProxyClient.Stage(ctx, chain, key)
}
func (p coordinatedTestProxy) Commit(ctx context.Context, id string) error {
if p.commit != nil {
return p.commit(ctx, id)
}
return p.ProxyClient.Commit(ctx, id)
}
type coordinatorFixture struct {
*rotationFixture
coordinator Coordinator
session *coordinatedTestUpstream
previous ManagedCertificate
candidate ManagedCertificate
}
func newCoordinatorFixture(t *testing.T) *coordinatorFixture {
t.Helper()
f := &coordinatorFixture{rotationFixture: newRotationFixture(t)}
materials := t.TempDir()
files := CertificateFiles{CertificatePath: filepath.Join(materials, "candidate.pem"), PrivateKeyPath: filepath.Join(materials, "candidate.key")}
writeTestFile(t, files.CertificatePath, f.rotationFixture.candidate.chain)
writeTestFile(t, files.PrivateKeyPath, f.rotationFixture.candidate.key)
f.previous = ManagedCertificate{
CertificateFiles: CertificateFiles{CertificatePath: filepath.Join(materials, "old.pem"), PrivateKeyPath: filepath.Join(materials, "old.key")},
Hostname: testHostname, Mode: "proxy", LeafHash: leafHash(f.old.pair.Certificate[0]),
ProxyAddress: f.proxy.Address, UpstreamAddress: f.upstream.Address,
}
writeTestFile(t, f.previous.CertificatePath, f.old.chain)
writeTestFile(t, f.previous.PrivateKeyPath, f.old.key)
f.candidate = f.previous
f.candidate.CertificateFiles, f.candidate.LeafHash = files, leafHash(f.rotationFixture.candidate.pair.Certificate[0])
f.session = &coordinatedTestUpstream{current: f.previous}
f.session.activate = func(actual CertificateFiles, hash string) error {
if actual != files || hash != f.candidate.LeafHash {
return errors.New("协调器未固定候选材料")
}
f.session.current = f.candidate
f.upstream.active.Store(&f.rotationFixture.candidate.pair)
return nil
}
directory := t.TempDir()
if err := os.Chmod(directory, 0o700); err != nil {
t.Fatal(err)
}
f.coordinator = Coordinator{Directory: directory, Proxy: f.rotation, Upstream: f.session}
return f
}
func (f *coordinatorFixture) record(t *testing.T) coordinationRecord {
t.Helper()
root, err := os.OpenRoot(f.coordinator.Directory)
if err != nil {
t.Fatal(err)
}
defer root.Close()
value, err := readCoordination(root)
if err != nil {
t.Fatal(err)
}
return value
}
func (f *coordinatorFixture) save(t *testing.T, record coordinationRecord) {
t.Helper()
root, err := os.OpenRoot(f.coordinator.Directory)
if err != nil {
t.Fatal(err)
}
defer root.Close()
if err := saveCoordination(root, record); err != nil {
t.Fatal(err)
}
}
func TestCoordinatorCommitsOverLocalIPC(t *testing.T) {
f := newCoordinatorFixture(t)
client, cancel, done := startLocalService(t, f.rotation, uint32(os.Geteuid()))
defer stopLocalService(t, cancel, done)
f.coordinator.Proxy = client
if err := f.coordinator.Ready(t.Context()); err != nil {
t.Fatal(err)
}
if err := f.coordinator.Activate(t.Context(), f.candidate.CertificateFiles); err != nil {
t.Fatal(err)
}
if record := f.record(t); record.Phase != "committed" || record.ProxyID == "" {
t.Fatal("双层提交未持久化", record.Phase)
}
if err := f.coordinator.Ready(t.Context()); err != nil {
t.Fatal("提交后的双层入口未就绪", err)
}
if err := f.coordinator.Recover(t.Context()); err != nil {
t.Fatal(err)
}
if f.reloader.reloadCount != 1 {
t.Fatal("已完成事务不应再次重载")
}
}
func TestCoordinatorRecoversLostResponses(t *testing.T) {
for _, operation := range []string{"stage", "commit"} {
t.Run(operation, func(t *testing.T) {
f := newCoordinatorFixture(t)
proxy := coordinatedTestProxy{ProxyClient: f.rotation}
if operation == "stage" {
proxy.stage = func(ctx context.Context, chain, key []byte) (string, error) {
if _, err := f.rotation.Stage(ctx, chain, key); err != nil {
return "", err
}
return "", errors.New("注入暂存应答丢失")
}
} else {
proxy.commit = func(ctx context.Context, id string) error {
if err := f.rotation.Commit(ctx, id); err != nil {
return err
}
return errors.New("注入提交应答丢失")
}
}
f.coordinator.Proxy = proxy
err := f.coordinator.Activate(t.Context(), f.candidate.CertificateFiles)
want := "committed"
if operation == "stage" {
want = "rolled_back"
if err == nil || f.session.current != f.previous {
t.Fatal("暂存应答丢失后不应切换上游", err)
}
} else if err != nil || f.session.current != f.candidate {
t.Fatal("应验证已提交的同一事务", err)
}
if record := f.record(t); record.Phase != want || record.ProxyID == "" {
t.Fatal("应答丢失后的终态错误", record.Phase)
}
})
}
}
func TestCoordinatorRecoversRestartWindows(t *testing.T) {
for _, phase := range []string{"prepared", "staged", "saved_before_snapshot", "upstream", "commit_before_record"} {
t.Run(phase, func(t *testing.T) {
f := newCoordinatorFixture(t)
record := coordinationRecord{Version: 1, Phase: "prepared", Previous: f.previous, Candidate: f.candidate}
f.save(t, record)
id, err := f.rotation.Stage(t.Context(), f.rotationFixture.candidate.chain, f.rotationFixture.candidate.key)
if err != nil {
t.Fatal(err)
}
want := "rolled_back"
if phase != "prepared" {
record.Phase, record.ProxyID = "staged", id
f.save(t, record)
}
if phase == "saved_before_snapshot" {
f.session.readErr = errors.New("配置已保存但运行快照仍旧")
if err := f.coordinator.Recover(t.Context()); !errors.Is(err, ErrCoordinationPending) {
t.Fatal("混合状态下不能回退或提交", err)
}
f.assertStatus(t, "staged")
f.session.readErr = nil
}
if phase == "saved_before_snapshot" || phase == "upstream" || phase == "commit_before_record" {
if err := f.session.Activate(f.candidate.CertificateFiles, f.candidate.LeafHash); err != nil {
t.Fatal(err)
}
want = "committed"
}
if phase == "upstream" || phase == "commit_before_record" {
record.Phase = "upstream"
f.save(t, record)
}
if phase == "commit_before_record" {
if err := f.rotation.Commit(t.Context(), id); err != nil {
t.Fatal(err)
}
}
// 新实例只依靠同一目录的持久化记录和重启后实际提供的证书。
restarted := Coordinator{Directory: f.coordinator.Directory, Proxy: f.rotation, Upstream: f.session}
if err := restarted.Recover(t.Context()); err != nil {
t.Fatal(err)
}
if record := f.record(t); record.Phase != want || record.ProxyID != id {
t.Fatal("重启恢复终态错误", record.Phase)
}
})
}
}
func TestCoordinatorKeepsUnobservedStagePending(t *testing.T) {
f := newCoordinatorFixture(t)
calls := 0
f.coordinator.Proxy = coordinatedTestProxy{ProxyClient: f.rotation, stage: func(context.Context, []byte, []byte) (string, error) {
calls++
return "", errors.New("暂存请求是否到达尚不可知")
}}
if err := f.coordinator.Activate(t.Context(), f.candidate.CertificateFiles); !errors.Is(err, ErrCoordinationPending) {
t.Fatal(err)
}
for range 2 {
if err := f.coordinator.Ready(t.Context()); !errors.Is(err, ErrCoordinationPending) {
t.Fatal("未观测到事务不能允许再次申请", err)
}
if err := f.coordinator.Activate(t.Context(), f.candidate.CertificateFiles); !errors.Is(err, ErrCoordinationPending) {
t.Fatal("未观测到事务不能重放暂存", err)
}
}
if calls != 1 || f.record(t).Phase != "prepared" {
t.Fatal("不能重放请求或清除不确定意图")
}
if _, err := f.rotation.Stage(t.Context(), f.rotationFixture.candidate.chain, f.rotationFixture.candidate.key); err != nil {
t.Fatal(err)
}
if err := f.coordinator.Recover(t.Context()); err != nil {
t.Fatal("迟到事务应在观测后恢复", err)
}
if f.record(t).Phase != "rolled_back" {
t.Fatal("迟到暂存没有回退")
}
}
func TestCoordinatorRefusesUnrelatedStateAndPolicy(t *testing.T) {
for _, change := range []string{"hostname", "proxy", "upstream", "leaf", "mode"} {
t.Run(change, func(t *testing.T) {
f := newCoordinatorFixture(t)
switch change {
case "hostname":
f.session.current.Hostname = "other.example.test"
case "proxy":
f.session.current.ProxyAddress = "127.0.0.1:1"
case "upstream":
f.session.current.UpstreamAddress = "127.0.0.1:2"
case "leaf":
f.session.current.LeafHash = f.candidate.LeafHash
case "mode":
f.session.current.Mode = "direct"
}
if err := f.coordinator.Ready(t.Context()); err == nil {
t.Fatal("错误策略不能通过签发前检查")
}
})
}
f := newCoordinatorFixture(t)
f.save(t, coordinationRecord{Version: 1, Phase: "prepared", Previous: f.previous, Candidate: f.candidate})
f.session.current.Mode = "direct"
if err := f.coordinator.Recover(t.Context()); !errors.Is(err, ErrCoordinationPending) {
t.Fatal("不能因模式切换忽略未决事务", err)
}
}
func TestCoordinatorSerializesWholeLifecycle(t *testing.T) {
f := newCoordinatorFixture(t)
started, release := make(chan struct{}), make(chan struct{})
defer close(release)
f.reloader.check = func(number int) error {
if number == 2 {
close(started)
<-release
}
return nil
}
done := make(chan error, 1)
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
defer cancel()
go func() { done <- f.coordinator.Activate(ctx, f.candidate.CertificateFiles) }()
select {
case <-started:
case <-time.After(3 * time.Second):
t.Fatal("协调未开始")
}
second := f.coordinator
if err := second.Recover(ctx); err == nil {
t.Fatal("独立协调实例不能在生命周期中途恢复")
}
if err := second.Ready(ctx); err == nil {
t.Fatal("独立协调实例不能开始下一次申请")
}
go func() { release <- struct{}{} }()
if err := <-done; err != nil {
t.Fatal(err)
}
}