From 822e204ae0c9c38e1d107c1dbf20159d3c6a800a Mon Sep 17 00:00:00 2001 From: xuans Date: Sun, 20 Sep 2026 01:37:49 +0800 Subject: [PATCH] =?UTF-8?q?=E6=A3=80=E6=9F=A5=E7=82=B9=EF=BC=9A=E4=BF=9D?= =?UTF-8?q?=E5=AD=98=E8=AF=B7=E6=B1=82=E5=8F=96=E6=B6=88=E6=A0=85=E6=A0=8F?= =?UTF-8?q?=E4=B8=8E=E7=9B=91=E5=90=AC=E4=BF=9D=E6=8A=A4=EF=BC=8C=E5=87=86?= =?UTF-8?q?=E5=A4=87=E8=A1=A5=E9=BD=90Linux=E5=8F=91=E5=B8=83=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/management_activation_linux.go | 6 + .../internal/app/management_recovery_test.go | 17 ++ backend/internal/app/management_renewal.go | 16 +- .../internal/app/management_renewal_test.go | 6 +- backend/internal/nginxactivation/README.md | 14 +- .../internal/nginxactivation/client_linux.go | 19 ++ .../coordination_record_linux.go | 8 + .../nginxactivation/coordinator_linux.go | 36 +++- .../nginxactivation/coordinator_linux_test.go | 192 ++++++++++++++++-- backend/internal/nginxactivation/ipc_linux.go | 12 +- .../nginxactivation/nginx_linux_test.go | 108 ++++++++-- .../nginxactivation/reconcile_linux.go | 2 +- .../internal/nginxactivation/request_linux.go | 123 +++++++++++ .../nginxactivation/request_linux_test.go | 164 +++++++++++++++ .../nginxactivation/rotation_linux.go | 12 +- backend/internal/nginxactivation/status.go | 1 + 16 files changed, 685 insertions(+), 51 deletions(-) create mode 100644 backend/internal/nginxactivation/request_linux.go create mode 100644 backend/internal/nginxactivation/request_linux_test.go diff --git a/backend/internal/app/management_activation_linux.go b/backend/internal/app/management_activation_linux.go index 39c1585..929625e 100644 --- a/backend/internal/app/management_activation_linux.go +++ b/backend/internal/app/management_activation_linux.go @@ -42,6 +42,9 @@ func (a coordinatedManagementActivator) coordinator(expected *httpapi.Management } func (a coordinatedManagementActivator) Ready(ctx context.Context, state httpapi.ManagementCertificateState) error { + if err := requireManagementListener(ctx, a.server); err != nil { + return err + } if state.HTTPSMode == "direct" { if err := a.coordinator(nil).Recover(ctx); err != nil { return err @@ -52,6 +55,9 @@ func (a coordinatedManagementActivator) Ready(ctx context.Context, state httpapi } func (a coordinatedManagementActivator) Activate(ctx context.Context, state httpapi.ManagementCertificateState, saved certificate.SavedResource) error { + if err := requireManagementListener(ctx, a.server); err != nil { + return err + } if state.HTTPSMode == "direct" { if err := a.Ready(ctx, state); err != nil { return err diff --git a/backend/internal/app/management_recovery_test.go b/backend/internal/app/management_recovery_test.go index 100c523..22d027b 100644 --- a/backend/internal/app/management_recovery_test.go +++ b/backend/internal/app/management_recovery_test.go @@ -114,3 +114,20 @@ func TestManagementRecoveryWaitsForHTTPS(t *testing.T) { default: } } + +func TestManagementActivationRequiresHTTPSListener(t *testing.T) { + for _, server := range []*httpapi.Server{nil, httpapi.New(httpapi.Dependencies{})} { + if err := requireManagementListener(t.Context(), server); err == nil { + t.Fatal("没有监听器的桌面或嵌入式运行时不能申请管理证书") + } + activator := directManagementActivator{server: server} + if err := activator.Ready(t.Context(), httpapi.ManagementCertificateState{HTTPSMode: "direct"}); err == nil { + t.Fatal("直接模式不能绕过监听检查") + } + } + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if err := requireManagementListener(ctx, nil); !errors.Is(err, context.Canceled) { + t.Fatal("监听检查应保留取消原因", err) + } +} diff --git a/backend/internal/app/management_renewal.go b/backend/internal/app/management_renewal.go index 7dcbdbd..6d94fa2 100644 --- a/backend/internal/app/management_renewal.go +++ b/backend/internal/app/management_renewal.go @@ -240,10 +240,24 @@ type directManagementActivator struct { server *httpapi.Server } -func (a directManagementActivator) Ready(ctx context.Context, state httpapi.ManagementCertificateState) error { +func requireManagementListener(ctx context.Context, server *httpapi.Server) error { if err := ctx.Err(); err != nil { return err } + if server != nil { + select { + case <-server.Listening(): + return nil + default: + } + } + return errors.New("当前进程尚未启动管理 HTTPS 监听，已阻止申请管理证书") +} + +func (a directManagementActivator) Ready(ctx context.Context, state httpapi.ManagementCertificateState) error { + if err := requireManagementListener(ctx, a.server); err != nil { + return err + } if state.HTTPSMode != "direct" { return errors.New("管理证书代理激活器尚未配置，已阻止申请和上游单独切换") } diff --git a/backend/internal/app/management_renewal_test.go b/backend/internal/app/management_renewal_test.go index fc31bea..02a9eee 100644 --- a/backend/internal/app/management_renewal_test.go +++ b/backend/internal/app/management_renewal_test.go @@ -156,13 +156,15 @@ func TestManagementRenewalSkipsFreshAndIPCertificates(t *testing.T) { } func TestManagementRenewalPreflightRejectsBeforeIssuance(t *testing.T) { - for _, name := range []string{"dns", "proxy"} { + for _, name := range []string{"dns", "proxy", "no_listener"} { t.Run(name, func(t *testing.T) { f := newManagementFixture(t) if name == "dns" { f.state.DNSVerified = false } else { - f.state.HTTPSMode = "proxy" + if name == "proxy" { + f.state.HTTPSMode = "proxy" + } f.m.activator = directManagementActivator{server: f.app.HTTP} } report, err := f.m.RunRenewal(t.Context(), f.policy, f.now) diff --git a/backend/internal/nginxactivation/README.md b/backend/internal/nginxactivation/README.md index 13906dc..aca7f1b 100644 --- a/backend/internal/nginxactivation/README.md +++ b/backend/internal/nginxactivation/README.md @@ -1,6 +1,6 @@ # Nginx 本地激活组件 -本组件负责受限的 Nginx 管理证书事务，不是完整的管理证书自动续期协调器。当前不能仅凭组件测试通过就解除应用内代理续期保护。 +本组件包含受限的 Nginx 管理证书事务及应用侧双层协调器。应用已接入协调与周期恢复，但正式部署、真实 Nginx 协调及完整自动续期仍须单独验收，不能仅凭组件测试或交叉编译通过就正式切流。 ## 权限边界 @@ -19,11 +19,17 @@ | `ready` | 检查不存在未决事务、双层 TLS 和专用证书材料一致、Nginx 配置合法 | | `state` | 返回事务编号、状态和两张叶证书的 SHA-256，不返回私钥或磁盘路径 | | `stage` | 验证候选，持久化事务并仅切换 Nginx，成功后返回随机事务编号 | +| `stage_request` | 接受应用预先保存的随机请求编号，先持久化占位再暂存；同一编号不允许重放 | +| `resolve_request` | 与暂存使用同一把锁，返回关联事务，或先持久化取消栅栏再返回未执行终态 | | `commit` | 确认双层已使用候选证书，再记录提交 | | `reconcile` | 依据实际上游证书回退或提交；无关证书、上游不可用或外部变更时保留未决事务 | `stage` 的应答丢失不等于没有执行。调用方必须用 `state` 查询，并结合持久化的候选摘要核对事务，不能直接重新申请证书或重试暂存。重复 `commit` 不返回幂等成功，调用方须确认查询到的是同一事务的提交终态。`reconcile` 对已完成的同一事务返回原终态，不代表重新验证了当下入口状态。 +应用协调器只使用 `stage_request`，并在签发前要求 `ready.request_fencing=true`。请求编号为 128 位随机数的规范十六进制文本，不接受任意路径。激活器保留每个请求的占位或取消记录，后续事务覆盖主事务日志也不会解除旧编号的重放保护。尚未进入事务的请求可由 `resolve_request` 取消；并发暂存持锁时恢复请求失败并等待下轮，不抢先报告取消。取消应答丢失后可查询同一编号，不重放变更操作。 + +协调记录仍兼容旧版本没有请求编号的记录，但不能为旧请求追溯建立可靠栅栏。旧记录无法确认对应事务时继续保持未决，需要维护人员核对，禁止直接清除。新应用与激活器必须配套升级，旧客户端可能拒绝新增的就绪字段。请求记录目前不自动清理，容量规划须包含这些小文件。 + 激活等待要求三次间隔 100 毫秒的新连接探测连续通过，每次严格核验证书链、域名、叶证书及 Scert 就绪响应。该检查降低平滑重载过程中偶然命中新工作进程导致的误判，不构成对未来所有连接的保证。 停止服务会取消活动请求并等待有界回退；监听器异常关闭报告错误。请求持有的私钥切片及帧缓冲在返回路径清理，但不声明能够擦除 Go 运行时或 JSON 解码器产生的所有内部副本。 @@ -34,10 +40,14 @@ 2. 应用须持久化旧状态、候选摘要、激活阶段和关联事务；完整生命周期必须串行化，不能在恢复检查上游之后并发切换上游。 3. 覆盖代理暂存、上游配置持久化与内存切换、代理提交之间的中断及应答丢失，并在启动时先恢复再允许申请下一张证书。 4. 不可将本组件的 `Recover` 视作双层恢复。它只回退代理；`Reconcile` 也必须由了解应用持久化状态的协调者调用。 -5. 当前没有自动恢复循环、请求编号幂等机制或租约超时策略，不能宣称已具备无人值守续期闭环。 +5. 应用等待自身 HTTPS 进入接收循环后开始周期恢复，每次最多 100 秒，结束后等待一分钟；关闭应用会取消并等待恢复。只有 `Handler()` 的桌面或嵌入式运行时不能充当管理 HTTPS 上游，签发前检查明确拒绝。 +6. 协调记录归应用用户所有，允许中文或空格目录，同时保留 0700、所有者、父路径及链接检查；实际写入 Nginx 指令的激活目录仍限制字符。 +7. 本协议不提供租约超时后强制覆盖、自动清理旧版本或通用 Nginx 管理接口。损坏记录、保存失败、无关证书或配置变动保留保护，不宣称已经完成无人值守续期验收。 ## 验证范围 Windows 测试只能覆盖通用协议和 TLS 探测。Linux 测试覆盖目录、文件、Unix 身份、策略、事务及继承监听器；真实 Nginx 用例需显式指定 `SCERT_TEST_NGINX_BINARY`。不同 UID 测试需在隔离 root 单元中设置 `SCERT_TEST_ROOT_IPC=1`，只在临时目录中复制测试程序并降权子进程，不修改真实用户或正式 socket。 +`TestCoordinatorWithRealNginxAndIPC` 使用独立 Nginx 进程和真实 Unix IPC，覆盖正常提交以及暂存、提交应答丢失。上游仍为测试 HTTPS 夹具，不等同于真实应用配置持久化、ACME 签发或正式 systemd 安装验收。2026-09-20 新增用例已交叉编译，当前目标主机身份尚待确认，未执行本轮 Linux 用例。 + 实机执行命令、程序摘要和失败修复记录见仓库 `doc/Linux生产流程测试记录.md`。 diff --git a/backend/internal/nginxactivation/client_linux.go b/backend/internal/nginxactivation/client_linux.go index bf65097..15b78e5 100644 --- a/backend/internal/nginxactivation/client_linux.go +++ b/backend/internal/nginxactivation/client_linux.go @@ -48,6 +48,25 @@ func (c Client) Stage(ctx context.Context, chain, key []byte) (string, error) { return result.ID, err } +func (c Client) StageRequest(ctx context.Context, id string, chain, key []byte) (string, error) { + result, err := c.call(ctx, request{Action: "stage_request", ID: id, Chain: chain, Key: key}) + if err == nil && result.ID != id { + return "", errors.New("本地激活服务返回的编号与请求不一致") + } + return result.ID, err +} + +func (c Client) ResolveRequest(ctx context.Context, id string) (State, error) { + result, err := c.call(ctx, request{Action: "resolve_request", ID: id}) + if err != nil { + return State{}, err + } + if result.State == nil || result.State.ID != id { + return State{}, errors.New("本地激活服务没有返回对应请求状态") + } + return *result.State, nil +} + func (c Client) Commit(ctx context.Context, id string) error { _, err := c.call(ctx, request{Action: "commit", ID: id}) return err diff --git a/backend/internal/nginxactivation/coordination_record_linux.go b/backend/internal/nginxactivation/coordination_record_linux.go index fb9c927..d70a76d 100644 --- a/backend/internal/nginxactivation/coordination_record_linux.go +++ b/backend/internal/nginxactivation/coordination_record_linux.go @@ -16,6 +16,7 @@ type coordinationRecord struct { Version int `json:"version"` Phase string `json:"phase"` BaselineID string `json:"baseline_id,omitempty"` + RequestID string `json:"request_id,omitempty"` ProxyID string `json:"proxy_id,omitempty"` Previous ManagedCertificate `json:"previous"` Candidate ManagedCertificate `json:"candidate"` @@ -44,8 +45,15 @@ func (record coordinationRecord) validate() error { if record.Version != 1 || (record.BaselineID != "" && !validID(record.BaselineID)) || (record.ProxyID != "" && !validID(record.ProxyID)) { return errors.New("管理证书协调记录版本或编号无效") } + if record.RequestID != "" && (!validID(record.RequestID) || record.RequestID == record.BaselineID || (record.ProxyID != "" && record.ProxyID != record.RequestID)) { + return errors.New("管理证书协调请求编号不一致") + } switch record.Phase { case "prepared": + case "canceled": + if record.RequestID == "" || record.ProxyID != "" { + return errors.New("取消的管理证书协调必须保留未执行请求编号") + } case "staged", "upstream", "committed", "rolled_back": if record.ProxyID == "" { return errors.New("管理证书协调记录缺少代理事务编号") diff --git a/backend/internal/nginxactivation/coordinator_linux.go b/backend/internal/nginxactivation/coordinator_linux.go index 0dc6800..3c1ed70 100644 --- a/backend/internal/nginxactivation/coordinator_linux.go +++ b/backend/internal/nginxactivation/coordinator_linux.go @@ -41,7 +41,8 @@ type ManagedUpstream interface { type ProxyClient interface { Ready(context.Context) (ReadyState, error) State(context.Context) (State, error) - Stage(context.Context, []byte, []byte) (string, error) + StageRequest(context.Context, string, []byte, []byte) (string, error) + ResolveRequest(context.Context, string) (State, error) Commit(context.Context, string) error Reconcile(context.Context, string) (State, error) } @@ -104,7 +105,11 @@ func (c Coordinator) Activate(ctx context.Context, files CertificateFiles) error if err != nil { return err } - record := coordinationRecord{Version: 1, Phase: "prepared", BaselineID: baseline.ID, Previous: previous, Candidate: candidate} + requestID, err := newID() + if err != nil { + return err + } + record := coordinationRecord{Version: 1, Phase: "prepared", BaselineID: baseline.ID, RequestID: requestID, Previous: previous, Candidate: candidate} if err := record.validate(); err != nil { return err } @@ -120,11 +125,11 @@ func (c Coordinator) Activate(ctx context.Context, files CertificateFiles) error resultErr = errors.Join(resultErr, recoveryErr) } }() - id, err := c.Proxy.Stage(ctx, chain, key) + id, err := c.Proxy.StageRequest(ctx, requestID, chain, key) if err != nil { return err } - if !validID(id) || id == record.BaselineID { + if id != requestID { return errors.New("代理返回的暂存事务编号无效") } record.ProxyID, record.Phase = id, "staged" @@ -161,7 +166,7 @@ func (c Coordinator) recover(ctx context.Context, root *os.Root, session Managed if err != nil { return record, err } - if record.Phase == "committed" || record.Phase == "rolled_back" { + if record.Phase == "committed" || record.Phase == "rolled_back" || record.Phase == "canceled" { return record, nil } current, err := session.Snapshot() @@ -171,11 +176,27 @@ func (c Coordinator) recover(ctx context.Context, root *os.Root, session Managed if current != record.Previous && current != record.Candidate { return record, errors.Join(ErrCoordinationPending, errors.New("当前上游不属于持久化协调记录，拒绝覆盖")) } - state, err := c.Proxy.State(ctx) + var state State + if record.RequestID != "" { + state, err = c.Proxy.ResolveRequest(ctx, record.RequestID) + } else { + state, err = c.Proxy.State(ctx) + } if err != nil { return record, errors.Join(ErrCoordinationPending, err) } + if record.RequestID != "" && state.ID == record.RequestID && state.Status == "canceled" { + if record.Phase != "prepared" || record.ProxyID != "" || 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 = "canceled" + return record, saveCoordination(root, record) + } if !validID(state.ID) || state.ID == record.BaselineID || + (record.RequestID != "" && state.ID != record.RequestID) || (record.ProxyID != "" && state.ID != record.ProxyID) || state.PreviousHash != record.Previous.LeafHash || state.CandidateHash != record.Candidate.LeafHash { // 尚未观测到对应事务时保留意图；延迟到达的暂存请求仍可能执行，不能清除或重放。 @@ -209,6 +230,9 @@ func (c Coordinator) ready(ctx context.Context, expected ManagedCertificate) err if err != nil { return err } + if !ready.RequestFencing { + return errors.New("本地激活服务不支持请求取消栅栏，请先升级激活器") + } if ready.Hostname != expected.Hostname || ready.LeafHash != expected.LeafHash || ready.ProxyAddress != expected.ProxyAddress || ready.UpstreamAddress != expected.UpstreamAddress { return errors.New("本地激活策略或双层入口与应用管理配置不一致") } diff --git a/backend/internal/nginxactivation/coordinator_linux_test.go b/backend/internal/nginxactivation/coordinator_linux_test.go index 8e8f332..2d14f4f 100644 --- a/backend/internal/nginxactivation/coordinator_linux_test.go +++ b/backend/internal/nginxactivation/coordinator_linux_test.go @@ -38,15 +38,31 @@ func (u *coordinatedTestUpstream) Activate(files CertificateFiles, hash string) type coordinatedTestProxy struct { ProxyClient - stage func(context.Context, []byte, []byte) (string, error) - commit func(context.Context, string) error + stage func(context.Context, string, []byte, []byte) (string, error) + commit func(context.Context, string) error + ready func(context.Context) (ReadyState, error) + resolve func(context.Context, string) (State, error) } -func (p coordinatedTestProxy) Stage(ctx context.Context, chain, key []byte) (string, error) { - if p.stage != nil { - return p.stage(ctx, chain, key) +func (p coordinatedTestProxy) ResolveRequest(ctx context.Context, id string) (State, error) { + if p.resolve != nil { + return p.resolve(ctx, id) } - return p.ProxyClient.Stage(ctx, chain, key) + return p.ProxyClient.ResolveRequest(ctx, id) +} + +func (p coordinatedTestProxy) Ready(ctx context.Context) (ReadyState, error) { + if p.ready != nil { + return p.ready(ctx) + } + return p.ProxyClient.Ready(ctx) +} + +func (p coordinatedTestProxy) StageRequest(ctx context.Context, id string, chain, key []byte) (string, error) { + if p.stage != nil { + return p.stage(ctx, id, chain, key) + } + return p.ProxyClient.StageRequest(ctx, id, chain, key) } func (p coordinatedTestProxy) Commit(ctx context.Context, id string) error { @@ -171,8 +187,8 @@ func TestCoordinatorRecoversLostResponses(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 { + proxy.stage = func(ctx context.Context, id string, chain, key []byte) (string, error) { + if _, err := f.rotation.StageRequest(ctx, id, chain, key); err != nil { return "", err } return "", errors.New("注入暂存应答丢失") @@ -256,13 +272,12 @@ func TestCoordinatorRecoversRestartWindows(t *testing.T) { 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) { + f.coordinator.Proxy = coordinatedTestProxy{ProxyClient: f.rotation, stage: func(context.Context, string, []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) - } + // 旧版本没有请求编号，不能追溯创建栅栏并假定旧请求不会到达。 + f.save(t, coordinationRecord{Version: 1, Phase: "prepared", Previous: f.previous, Candidate: f.candidate}) for range 2 { if err := f.coordinator.Ready(t.Context()); !errors.Is(err, ErrCoordinationPending) { t.Fatal("未观测到事务不能允许再次申请", err) @@ -271,7 +286,7 @@ func TestCoordinatorKeepsUnobservedStagePending(t *testing.T) { t.Fatal("未观测到事务不能重放暂存", err) } } - if calls != 1 || f.record(t).Phase != "prepared" { + if calls != 0 || f.record(t).Phase != "prepared" { t.Fatal("不能重放请求或清除不确定意图") } if _, err := f.rotation.Stage(t.Context(), f.rotationFixture.candidate.chain, f.rotationFixture.candidate.key); err != nil { @@ -285,6 +300,157 @@ func TestCoordinatorKeepsUnobservedStagePending(t *testing.T) { } } +func TestCoordinatorFencesUndeliveredRequestOverIPC(t *testing.T) { + f := newCoordinatorFixture(t) + client, cancel, done := startLocalService(t, f.rotation, uint32(os.Geteuid())) + defer stopLocalService(t, cancel, done) + var delayedID string + f.coordinator.Proxy = coordinatedTestProxy{ProxyClient: client, stage: func(_ context.Context, id string, _, _ []byte) (string, error) { + delayedID = id + return "", errors.New("注入请求未送达") + }} + if err := f.coordinator.Activate(t.Context(), f.candidate.CertificateFiles); err == nil || errors.Is(err, ErrCoordinationPending) { + t.Fatal("失败应返回，但已确认取消后不应永久未决", err) + } + if record := f.record(t); record.Phase != "canceled" || record.RequestID != delayedID || record.ProxyID != "" { + t.Fatal("未送达请求没有持久化取消终态", record) + } + if f.reloader.reloadCount != 0 || f.session.current != f.previous { + t.Fatal("取消未送达请求不能切换证书") + } + 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 _, err := client.StageRequest(t.Context(), delayedID, f.rotationFixture.candidate.chain, f.rotationFixture.candidate.key); err == nil { + t.Fatal("旧请求迟到不能重新执行") + } + if f.reloader.reloadCount != 1 || f.record(t).Phase != "committed" { + t.Fatal("迟到请求不能影响新事务") + } +} + +func TestCoordinatorRequestRecoveryAfterRestart(t *testing.T) { + for _, phase := range []string{"unsent", "reserved", "staged", "upstream", "committed"} { + t.Run(phase, func(t *testing.T) { + f := newCoordinatorFixture(t) + id, err := newID() + if err != nil { + t.Fatal(err) + } + f.save(t, coordinationRecord{Version: 1, Phase: "prepared", RequestID: id, Previous: f.previous, Candidate: f.candidate}) + want := "canceled" + if phase == "reserved" { + if _, err := f.rotation.StageRequest(t.Context(), id, []byte("invalid"), []byte("invalid")); err == nil { + t.Fatal("无效证书必须拒绝") + } + } + if phase == "staged" || phase == "upstream" || phase == "committed" { + if _, err := f.rotation.StageRequest(t.Context(), id, f.rotationFixture.candidate.chain, f.rotationFixture.candidate.key); err != nil { + t.Fatal(err) + } + want = "rolled_back" + } + if phase == "upstream" || phase == "committed" { + if err := f.session.Activate(f.candidate.CertificateFiles, f.candidate.LeafHash); err != nil { + t.Fatal(err) + } + want = "committed" + } + if phase == "committed" { + 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.RequestID != id { + t.Fatal("重启后请求恢复终态错误", record) + } + }) + } +} + +func TestCoordinatorRejectsHelperWithoutRequestFencing(t *testing.T) { + f := newCoordinatorFixture(t) + f.coordinator.Proxy = coordinatedTestProxy{ProxyClient: f.rotation, ready: func(ctx context.Context) (ReadyState, error) { + state, err := f.rotation.Ready(ctx) + state.RequestFencing = false + return state, err + }} + if err := f.coordinator.Ready(t.Context()); err == nil { + t.Fatal("旧激活器不能通过签发前检查") + } + if err := f.coordinator.Activate(t.Context(), f.candidate.CertificateFiles); err == nil { + t.Fatal("旧激活器不能开始协调") + } + if _, err := os.Stat(filepath.Join(f.coordinator.Directory, coordinationName)); !errors.Is(err, os.ErrNotExist) { + t.Fatal("不支持栅栏时不能保存新的协调意图", err) + } +} + +func TestCoordinatorRecoversLostFenceResponse(t *testing.T) { + f := newCoordinatorFixture(t) + var id string + f.coordinator.Proxy = coordinatedTestProxy{ + ProxyClient: f.rotation, + stage: func(_ context.Context, requestID string, _, _ []byte) (string, error) { + id = requestID + return "", errors.New("注入请求未送达") + }, + resolve: func(ctx context.Context, requestID string) (State, error) { + if _, err := f.rotation.ResolveRequest(ctx, requestID); err != nil { + return State{}, err + } + return State{}, errors.New("注入取消栅栏应答丢失") + }, + } + if err := f.coordinator.Activate(t.Context(), f.candidate.CertificateFiles); !errors.Is(err, ErrCoordinationPending) { + t.Fatal("栅栏应答丢失时必须保留未决意图", err) + } + if f.record(t).Phase != "prepared" { + t.Fatal("未确认栅栏前不能标记取消") + } + if _, err := f.rotation.StageRequest(t.Context(), id, f.rotationFixture.candidate.chain, f.rotationFixture.candidate.key); err == nil { + t.Fatal("应答丢失不应撤销已持久化的栅栏") + } + restarted := Coordinator{Directory: f.coordinator.Directory, Proxy: f.rotation, Upstream: f.session} + if err := restarted.Recover(t.Context()); err != nil { + t.Fatal(err) + } + if f.record(t).Phase != "canceled" { + t.Fatal("重启后应确认同一请求的取消结果") + } +} + +func TestCoordinatorRejectsBrokenIntentBeforeStage(t *testing.T) { + for _, failure := range []string{"corrupt", "directory"} { + t.Run(failure, func(t *testing.T) { + f := newCoordinatorFixture(t) + path := filepath.Join(f.coordinator.Directory, coordinationName) + if failure == "corrupt" { + writeTestFile(t, path, []byte(`{"version":1,"phase":"prepared"}`)) + } else if err := os.Mkdir(path, 0o700); err != nil { + t.Fatal(err) + } + if err := f.coordinator.Activate(t.Context(), f.candidate.CertificateFiles); err == nil { + t.Fatal("无法可信保存协调意图时不能暂存") + } + if state, err := f.rotation.State(t.Context()); err != nil || state.Status != "idle" { + t.Fatal("协调记录损坏不应改变代理", state, err) + } + if f.reloader.reloadCount != 0 || f.session.current != f.previous { + t.Fatal("协调记录损坏不应改变双层证书") + } + }) + } +} + func TestCoordinatorRefusesUnrelatedStateAndPolicy(t *testing.T) { for _, change := range []string{"hostname", "proxy", "upstream", "leaf", "mode"} { t.Run(change, func(t *testing.T) { diff --git a/backend/internal/nginxactivation/ipc_linux.go b/backend/internal/nginxactivation/ipc_linux.go index efea1f5..2a12da1 100644 --- a/backend/internal/nginxactivation/ipc_linux.go +++ b/backend/internal/nginxactivation/ipc_linux.go @@ -112,7 +112,9 @@ func validRequest(message request) bool { return message.ID == "" && len(message.Chain) == 0 && len(message.Key) == 0 case "stage": return message.ID == "" && len(message.Chain) > 0 && len(message.Chain) <= 1<<20 && len(message.Key) > 0 && len(message.Key) <= 64<<10 - case "commit", "reconcile": + case "stage_request": + return validID(message.ID) && len(message.Chain) > 0 && len(message.Chain) <= 1<<20 && len(message.Key) > 0 && len(message.Key) <= 64<<10 + case "commit", "reconcile", "resolve_request": return validID(message.ID) && len(message.Chain) == 0 && len(message.Key) == 0 default: return false @@ -137,6 +139,14 @@ func (s LocalService) execute(ctx context.Context, message request) response { } case "stage": result.ID, err = s.Rotation.Stage(ctx, message.Chain, message.Key) + case "stage_request": + result.ID, err = s.Rotation.StageRequest(ctx, message.ID, message.Chain, message.Key) + case "resolve_request": + value, resolveErr := s.Rotation.ResolveRequest(ctx, message.ID) + err = resolveErr + if err == nil { + result.State = &value + } case "commit": err = s.Rotation.Commit(ctx, message.ID) case "reconcile": diff --git a/backend/internal/nginxactivation/nginx_linux_test.go b/backend/internal/nginxactivation/nginx_linux_test.go index 46a926b..4a053ae 100644 --- a/backend/internal/nginxactivation/nginx_linux_test.go +++ b/backend/internal/nginxactivation/nginx_linux_test.go @@ -15,6 +15,36 @@ import ( // 实机验收显式指定 Nginx 路径；仅启动独立进程，不使用系统站点、PID 或服务单元。 func TestRotationWithRealNginx(t *testing.T) { + f := newRotationFixture(t) + useRealNginx(t, f) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + id, err := f.rotation.Stage(ctx, f.candidate.chain, f.candidate.key) + if err != nil { + t.Fatal(err) + } + f.upstream.active.Store(&f.candidate.pair) + if err := f.rotation.Commit(ctx, id); err != nil { + t.Fatal(err) + } + f.assertStatus(t, "committed") + next := f.authority.issue(t, testHostname, time.Now().Add(-time.Hour), time.Now().Add(72*time.Hour)) + if _, err := f.rotation.Stage(ctx, next.chain, next.key); err != nil { + t.Fatal(err) + } + // 模拟调用方退出后由新组件恢复，真实入口必须重新提供上一张已提交证书。 + restarted := f.rotation + if err := restarted.Recover(ctx); err != nil { + t.Fatal(err) + } + f.assertStatus(t, "rolled_back") + if _, err := restarted.Proxy.Check(ctx, f.candidate.pair.Certificate[0]); err != nil { + t.Fatal(err) + } +} + +func useRealNginx(t *testing.T, f *rotationFixture) { + t.Helper() binary := os.Getenv("SCERT_TEST_NGINX_BINARY") if binary == "" { t.Skip("未指定隔离验收使用的 Nginx 可执行文件") @@ -22,7 +52,6 @@ func TestRotationWithRealNginx(t *testing.T) { if !filepath.IsAbs(binary) { t.Fatal("测试 Nginx 路径必须是绝对路径") } - f := newRotationFixture(t) prefix := t.TempDir() listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { @@ -106,27 +135,62 @@ http { if err := f.rotation.Proxy.Wait(ctx, f.old.pair.Certificate[0]); err != nil { t.Fatal(err) } - id, err := f.rotation.Stage(ctx, f.candidate.chain, f.candidate.key) - if err != nil { - t.Fatal(err) - } - f.upstream.active.Store(&f.candidate.pair) - if err := f.rotation.Commit(ctx, id); err != nil { - t.Fatal(err) - } - f.assertStatus(t, "committed") - next := f.authority.issue(t, testHostname, time.Now().Add(-time.Hour), time.Now().Add(72*time.Hour)) - if _, err := f.rotation.Stage(ctx, next.chain, next.key); err != nil { - t.Fatal(err) - } - // 模拟调用方退出后由新组件恢复，真实入口必须重新提供上一张已提交证书。 - restarted := f.rotation - if err := restarted.Recover(ctx); err != nil { - t.Fatal(err) - } - f.assertStatus(t, "rolled_back") - if _, err := restarted.Proxy.Check(ctx, f.candidate.pair.Certificate[0]); err != nil { - t.Fatal(err) +} + +func TestCoordinatorWithRealNginxAndIPC(t *testing.T) { + for _, failure := range []string{"none", "stage_response", "commit_response"} { + t.Run(failure, func(t *testing.T) { + f := newCoordinatorFixture(t) + useRealNginx(t, f.rotationFixture) + f.previous.ProxyAddress = f.rotation.Proxy.Address + f.candidate.ProxyAddress = f.rotation.Proxy.Address + f.session.current = f.previous + client, cancel, done := startLocalService(t, f.rotation, uint32(os.Geteuid())) + defer stopLocalService(t, cancel, done) + proxy := coordinatedTestProxy{ProxyClient: client} + if failure == "stage_response" { + proxy.stage = func(ctx context.Context, id string, chain, key []byte) (string, error) { + if _, err := client.StageRequest(ctx, id, chain, key); err != nil { + return "", err + } + return "", fmt.Errorf("注入暂存应答丢失") + } + } + if failure == "commit_response" { + proxy.commit = func(ctx context.Context, id string) error { + if err := client.Commit(ctx, id); err != nil { + return err + } + return fmt.Errorf("注入提交应答丢失") + } + } + f.coordinator.Proxy = proxy + ctx, stop := context.WithTimeout(t.Context(), 20*time.Second) + defer stop() + err := f.coordinator.Activate(ctx, f.candidate.CertificateFiles) + want := "committed" + leaf := f.rotationFixture.candidate.pair.Certificate[0] + if failure == "stage_response" { + want, leaf = "rolled_back", f.old.pair.Certificate[0] + if err == nil { + t.Fatal("暂存应答丢失应回退并报告失败") + } + } else if err != nil { + t.Fatal(err) + } + if record := f.record(t); record.Phase != want || record.RequestID != record.ProxyID { + t.Fatal("真实 Nginx 协调终态错误", record) + } + if _, err := f.rotation.Proxy.Check(ctx, leaf); err != nil { + t.Fatal(err) + } + if _, err := f.rotation.Upstream.Check(ctx, leaf); err != nil { + t.Fatal(err) + } + if err := f.coordinator.Ready(ctx); err != nil { + t.Fatal(err) + } + }) } } diff --git a/backend/internal/nginxactivation/reconcile_linux.go b/backend/internal/nginxactivation/reconcile_linux.go index b8c9a9a..df75f0e 100644 --- a/backend/internal/nginxactivation/reconcile_linux.go +++ b/backend/internal/nginxactivation/reconcile_linux.go @@ -70,7 +70,7 @@ func (r Rotation) Ready(ctx context.Context) (ReadyState, error) { if err := requireInclude(root, include); err != nil { return ReadyState{}, err } - return ReadyState{Hostname: r.Proxy.Hostname, ProxyAddress: r.Proxy.Address, UpstreamAddress: r.Upstream.Address, LeafHash: leafHash(leaf)}, nil + return ReadyState{Hostname: r.Proxy.Hostname, ProxyAddress: r.Proxy.Address, UpstreamAddress: r.Upstream.Address, LeafHash: leafHash(leaf), RequestFencing: true}, nil } // Reconcile 仅根据严格验证的实际上游完成未决事务，不推断配置保存是否成功。 diff --git a/backend/internal/nginxactivation/request_linux.go b/backend/internal/nginxactivation/request_linux.go new file mode 100644 index 0000000..03aafb4 --- /dev/null +++ b/backend/internal/nginxactivation/request_linux.go @@ -0,0 +1,123 @@ +package nginxactivation + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "os" +) + +// 请求记录永久保留，防止已取消或已结束的请求在后续续期时迟到执行。 +type stageRequest struct { + ID string `json:"id"` + BaselineID string `json:"baseline_id,omitempty"` + Status string `json:"status"` +} + +func readStageRequest(root *os.Root, id string) (stageRequest, error) { + var value stageRequest + data, err := readRegular(root, "request-"+id+".json", 4096) + if err != nil { + return value, err + } + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&value); err != nil { + return value, err + } + var extra any + if err := decoder.Decode(&extra); err != io.EOF { + return value, errors.New("暂存请求记录包含多余内容") + } + if value.ID != id || (value.BaselineID != "" && !validID(value.BaselineID)) || value.BaselineID == id || (value.Status != "reserved" && value.Status != "canceled") { + return value, errors.New("暂存请求记录无效") + } + return value, nil +} + +func saveStageRequest(root *os.Root, value stageRequest) error { + data, err := json.Marshal(value) + if err != nil { + return err + } + return writeAtomic(root, "request-"+value.ID+".json", data) +} + +// StageRequest 先持久化请求占位再执行暂存；同一编号不能重放，即使首次未完成。 +func (r Rotation) StageRequest(ctx context.Context, id string, chain, key []byte) (string, error) { + if !validID(id) { + return "", errors.New("暂存请求编号无效") + } + root, lock, err := r.open(ctx) + if err != nil { + return "", err + } + defer root.Close() + defer lock.Close() + if _, err := readStageRequest(root, id); !errors.Is(err, os.ErrNotExist) { + return "", errors.Join(errors.New("暂存请求已处理或记录不可信，拒绝重放"), err) + } + previous, err := readTransaction(root) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return "", err + } + if previous.ID == id || previous.Status == "pending" || previous.Status == "staged" { + return "", errors.New("暂存请求与已有事务冲突") + } + if err := saveStageRequest(root, stageRequest{ID: id, BaselineID: previous.ID, Status: "reserved"}); err != nil { + return "", err + } + return r.stage(ctx, root, id, chain, key) +} + +// ResolveRequest 与暂存持有同一把锁；没有进入事务的请求先落盘取消栅栏再返回。 +// 已进入事务时仅返回可核对的状态，双层恢复仍由协调器驱动。 +func (r Rotation) ResolveRequest(ctx context.Context, id string) (State, error) { + if !validID(id) { + return State{}, errors.New("暂存请求编号无效") + } + root, lock, err := r.open(ctx) + if err != nil { + return State{}, err + } + defer root.Close() + defer lock.Close() + value, err := readStageRequest(root, id) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return State{}, err + } + current, stateErr := readTransaction(root) + if stateErr != nil && !errors.Is(stateErr, os.ErrNotExist) { + return State{}, stateErr + } + if value.Status == "canceled" { + if current.ID == id { + return State{}, errors.New("取消栅栏与代理事务冲突，拒绝推断") + } + // 先前写入可能在重命名后同步失败；重试必须重新确认目录持久化。 + if err := saveStageRequest(root, value); err != nil { + return State{}, err + } + return State{ID: id, Status: "canceled"}, nil + } + if err == nil { + if current.ID == id { + if err := saveTransaction(root, current); err != nil { + return State{}, err + } + return transactionState(current), nil + } + if current.ID != value.BaselineID { + return State{}, errors.New("暂存请求的基准事务发生变化，保留记录") + } + } else if current.ID == id { + return State{}, errors.New("代理事务缺少对应请求记录，拒绝推断") + } + value = stageRequest{ID: id, Status: "canceled"} + if err := saveStageRequest(root, value); err != nil { + return State{}, err + } + return State{ID: id, Status: "canceled"}, nil +} diff --git a/backend/internal/nginxactivation/request_linux_test.go b/backend/internal/nginxactivation/request_linux_test.go new file mode 100644 index 0000000..09613ca --- /dev/null +++ b/backend/internal/nginxactivation/request_linux_test.go @@ -0,0 +1,164 @@ +package nginxactivation + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/processlock" +) + +func requestIDForTest(t *testing.T) string { + t.Helper() + id, err := newID() + if err != nil { + t.Fatal(err) + } + return id +} + +func TestRequestFencePersistsAcrossRestartAndNextTransaction(t *testing.T) { + f := newRotationFixture(t) + id := requestIDForTest(t) + for range 2 { + state, err := f.rotation.ResolveRequest(t.Context(), id) + if err != nil || state.ID != id || state.Status != "canceled" { + t.Fatal("未送达请求应持久化取消栅栏", state, err) + } + } + nextID := requestIDForTest(t) + if _, err := f.rotation.StageRequest(t.Context(), nextID, f.candidate.chain, f.candidate.key); err != nil { + t.Fatal(err) + } + f.upstream.active.Store(&f.candidate.pair) + if err := f.rotation.Commit(t.Context(), nextID); err != nil { + t.Fatal(err) + } + restarted := f.rotation + if _, err := restarted.StageRequest(t.Context(), id, f.candidate.chain, f.candidate.key); err == nil { + t.Fatal("重启且后续事务提交后仍须拒绝旧请求") + } + if _, err := restarted.StageRequest(t.Context(), nextID, f.candidate.chain, f.candidate.key); err == nil { + t.Fatal("已经提交的请求不能重放") + } + if f.reloader.reloadCount != 1 { + t.Fatal("重复请求触发了额外重载") + } +} + +func TestRequestReservationFailureDoesNotActivate(t *testing.T) { + for _, invalid := range []string{"symlink", "corrupt", "directory"} { + t.Run(invalid, func(t *testing.T) { + f := newRotationFixture(t) + id := requestIDForTest(t) + path := filepath.Join(f.rotation.Directory, "request-"+id+".json") + switch invalid { + case "symlink": + if err := os.Symlink("missing", path); err != nil { + t.Fatal(err) + } + case "directory": + if err := os.Mkdir(path, 0o700); err != nil { + t.Fatal(err) + } + case "corrupt": + writeTestFile(t, path, []byte(`{"status":"canceled"}`)) + } + if _, err := f.rotation.StageRequest(t.Context(), id, f.candidate.chain, f.candidate.key); err == nil { + t.Fatal("无法可信保存请求时不能激活") + } + if _, err := f.rotation.ResolveRequest(t.Context(), id); err == nil { + t.Fatal("不能把损坏请求记录当成尚未执行") + } + if f.reloader.reloadCount != 0 { + t.Fatal("请求记录故障不能重载代理") + } + }) + } +} + +func TestRequestResolveCannotOvertakeActiveStage(t *testing.T) { + f := newRotationFixture(t) + id := requestIDForTest(t) + started, release := make(chan struct{}), make(chan struct{}) + f.reloader.check = func(number int) error { + if number == 1 { + close(started) + <-release + } + return nil + } + done := make(chan error, 1) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + go func() { + _, err := f.rotation.StageRequest(ctx, id, f.candidate.chain, f.candidate.key) + done <- err + }() + select { + case <-started: + case <-time.After(3 * time.Second): + close(release) + t.Fatal("暂存没有开始") + } + _, err := f.rotation.ResolveRequest(ctx, id) + close(release) + stageErr := <-done + if !errors.Is(err, processlock.ErrLocked) || stageErr != nil { + t.Fatal("恢复不能越过正在执行的暂存", err, stageErr) + } + state, err := f.rotation.ResolveRequest(ctx, id) + if err != nil || state.ID != id || state.Status != "staged" { + t.Fatal("已执行请求应返回对应事务", state, err) + } +} + +func TestRequestReservationBeforeTransactionCanBeCanceled(t *testing.T) { + f := newRotationFixture(t) + id := requestIDForTest(t) + if _, err := f.rotation.StageRequest(t.Context(), id, []byte("invalid"), []byte("invalid")); err == nil { + t.Fatal("无效材料必须拒绝") + } + state, err := f.rotation.ResolveRequest(t.Context(), id) + if err != nil || state.Status != "canceled" { + t.Fatal("已占位但未写入事务的请求应允许取消", state, err) + } + if _, err := f.rotation.StageRequest(t.Context(), id, f.candidate.chain, f.candidate.key); err == nil { + t.Fatal("同一编号不能在失败后替换材料重试") + } +} + +func TestRequestResolveRefusesChangedBaseline(t *testing.T) { + f := newRotationFixture(t) + id := requestIDForTest(t) + if _, err := f.rotation.StageRequest(t.Context(), id, []byte("invalid"), []byte("invalid")); err == nil { + t.Fatal("无效材料必须拒绝") + } + if _, err := f.rotation.Stage(t.Context(), f.candidate.chain, f.candidate.key); err != nil { + t.Fatal(err) + } + if _, err := f.rotation.ResolveRequest(t.Context(), id); err == nil { + t.Fatal("基准事务已变化时不能推断旧占位尚未执行") + } + f.assertStatus(t, "staged") + if f.reloader.reloadCount != 1 { + t.Fatal("恢复查询不能改动无关事务") + } +} + +func TestRequestProtocolRejectsMissingIDAndExtraMaterial(t *testing.T) { + id := requestIDForTest(t) + for _, message := range []request{ + {Version: protocolVersion, Action: "stage_request", Chain: []byte("chain"), Key: []byte("key")}, + {Version: protocolVersion, Action: "stage_request", ID: "../request", Chain: []byte("chain"), Key: []byte("key")}, + {Version: protocolVersion, Action: "resolve_request"}, + {Version: protocolVersion, Action: "resolve_request", ID: id, Chain: []byte("chain")}, + } { + if validRequest(message) { + t.Fatal("无效请求不应通过协议检查", message.Action) + } + } +} diff --git a/backend/internal/nginxactivation/rotation_linux.go b/backend/internal/nginxactivation/rotation_linux.go index c05070d..0eca9fc 100644 --- a/backend/internal/nginxactivation/rotation_linux.go +++ b/backend/internal/nginxactivation/rotation_linux.go @@ -55,6 +55,10 @@ func (r Rotation) Stage(ctx context.Context, chain, key []byte) (id string, resu } defer root.Close() defer lock.Close() + return r.stage(ctx, root, "", chain, key) +} + +func (r Rotation) stage(ctx context.Context, root *os.Root, id string, chain, key []byte) (_ string, resultErr error) { previous, err := readTransaction(root) if err != nil && !errors.Is(err, os.ErrNotExist) { return "", err @@ -87,9 +91,11 @@ func (r Rotation) Stage(ctx context.Context, chain, key []byte) (id string, resu if err := r.Reloader.Check(ctx); err != nil { return "", fmt.Errorf("Nginx 原配置检查失败：%w", err) } - id, err = newID() - if err != nil { - return "", err + if id == "" { + id, err = newID() + if err != nil { + return "", err + } } if err := root.Mkdir(id, 0o700); err != nil { return "", err diff --git a/backend/internal/nginxactivation/status.go b/backend/internal/nginxactivation/status.go index 32c7ba8..766b0b8 100644 --- a/backend/internal/nginxactivation/status.go +++ b/backend/internal/nginxactivation/status.go @@ -13,4 +13,5 @@ type ReadyState struct { ProxyAddress string `json:"proxy_address"` UpstreamAddress string `json:"upstream_address"` LeafHash string `json:"leaf_hash"` + RequestFencing bool `json:"request_fencing"` }