xuans/Scert
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

检查点：保存请求取消栅栏与监听保护，准备补齐Linux发布包

Browse Source
This commit is contained in:
xuans 2026-09-20 01:37:49 +08:00
parent 76b4d94ca0
commit 822e204ae0
16 changed files with 685 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
backend/internal/app/management_activation_linux.go
View File

@ -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

17
backend/internal/app/management_recovery_test.go
View File

@ -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)
}
}

16
backend/internal/app/management_renewal.go
View File

@ -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("管理证书代理激活器尚未配置，已阻止申请和上游单独切换")
}

6
backend/internal/app/management_renewal_test.go
View File

@ -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)

14
backend/internal/nginxactivation/README.md
View File

@ -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`。

19
backend/internal/nginxactivation/client_linux.go
View File

@ -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

8
backend/internal/nginxactivation/coordination_record_linux.go
View File

@ -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("管理证书协调记录缺少代理事务编号")

36
backend/internal/nginxactivation/coordinator_linux.go
View File

@ -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("本地激活策略或双层入口与应用管理配置不一致")
}

192
backend/internal/nginxactivation/coordinator_linux_test.go
View File

@ -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) {

12
backend/internal/nginxactivation/ipc_linux.go
View File

@ -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":

108
backend/internal/nginxactivation/nginx_linux_test.go
View File

@ -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)
}
})
}
}

2
backend/internal/nginxactivation/reconcile_linux.go
View File

@ -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 仅根据严格验证的实际上游完成未决事务，不推断配置保存是否成功。

123
backend/internal/nginxactivation/request_linux.go Normal file
View File

@ -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
}

164
backend/internal/nginxactivation/request_linux_test.go Normal file
View File

@ -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)
}
}
}

12
backend/internal/nginxactivation/rotation_linux.go
View File

@ -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

1
backend/internal/nginxactivation/status.go
View File

@ -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"`
}