检查点:保存监听时序与中文目录修复,继续处理协调请求恢复
This commit is contained in:
parent
2c0d0e8eda
commit
76b4d94ca0
@ -14,6 +14,13 @@ func (a *App) startManagementRecovery(recover func(context.Context) error) {
|
|||||||
a.managementDone = make(chan struct{})
|
a.managementDone = make(chan struct{})
|
||||||
go func() {
|
go func() {
|
||||||
defer close(a.managementDone)
|
defer close(a.managementDone)
|
||||||
|
if a.HTTP != nil {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-a.HTTP.Listening():
|
||||||
|
}
|
||||||
|
}
|
||||||
for {
|
for {
|
||||||
attempt, stop := context.WithTimeout(ctx, 100*time.Second)
|
attempt, stop := context.WithTimeout(ctx, 100*time.Second)
|
||||||
err := recover(attempt)
|
err := recover(attempt)
|
||||||
|
|||||||
116
backend/internal/app/management_recovery_test.go
Normal file
116
backend/internal/app/management_recovery_test.go
Normal file
@ -0,0 +1,116 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.pcux.cn/xuans/Scert/backend/internal/httpapi"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestManagementRecoveryCloseCancelsAndWaits(t *testing.T) {
|
||||||
|
application := &App{}
|
||||||
|
started := make(chan struct{})
|
||||||
|
canceled := make(chan struct{})
|
||||||
|
release := make(chan struct{})
|
||||||
|
var once sync.Once
|
||||||
|
releaseRecovery := func() { once.Do(func() { close(release) }) }
|
||||||
|
t.Cleanup(releaseRecovery)
|
||||||
|
application.startManagementRecovery(func(ctx context.Context) error {
|
||||||
|
if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) > 100*time.Second {
|
||||||
|
return errors.New("恢复尝试必须有超时边界")
|
||||||
|
}
|
||||||
|
close(started)
|
||||||
|
<-ctx.Done()
|
||||||
|
close(canceled)
|
||||||
|
<-release
|
||||||
|
return ctx.Err()
|
||||||
|
})
|
||||||
|
t.Cleanup(func() {
|
||||||
|
releaseRecovery()
|
||||||
|
if err := application.Close(); err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
select {
|
||||||
|
case <-started:
|
||||||
|
case <-time.After(3 * time.Second):
|
||||||
|
t.Fatal("恢复协程没有开始执行")
|
||||||
|
}
|
||||||
|
closed := make(chan error, 1)
|
||||||
|
go func() { closed <- application.Close() }()
|
||||||
|
select {
|
||||||
|
case <-canceled:
|
||||||
|
case <-time.After(3 * time.Second):
|
||||||
|
t.Fatal("关闭应用没有取消恢复尝试")
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case err := <-closed:
|
||||||
|
t.Fatal("恢复尝试退出前不能完成应用关闭", err)
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
releaseRecovery()
|
||||||
|
select {
|
||||||
|
case err := <-closed:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
case <-time.After(3 * time.Second):
|
||||||
|
t.Fatal("恢复退出后应用关闭没有完成")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestManagementRecoveryCloseInterruptsRetryWait(t *testing.T) {
|
||||||
|
application := &App{}
|
||||||
|
returned := make(chan struct{})
|
||||||
|
application.startManagementRecovery(func(context.Context) error {
|
||||||
|
close(returned)
|
||||||
|
return errors.New("注入临时恢复失败")
|
||||||
|
})
|
||||||
|
t.Cleanup(func() { _ = application.Close() })
|
||||||
|
select {
|
||||||
|
case <-returned:
|
||||||
|
case <-time.After(3 * time.Second):
|
||||||
|
t.Fatal("首次恢复未执行")
|
||||||
|
}
|
||||||
|
closed := make(chan error, 1)
|
||||||
|
go func() { closed <- application.Close() }()
|
||||||
|
select {
|
||||||
|
case err := <-closed:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
case <-time.After(3 * time.Second):
|
||||||
|
t.Fatal("关闭应用不应等待整分钟的重试间隔")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestManagementRecoveryNilDoesNotStartWorker(t *testing.T) {
|
||||||
|
application := &App{}
|
||||||
|
application.startManagementRecovery(nil)
|
||||||
|
if application.managementStop != nil || application.managementDone != nil {
|
||||||
|
t.Fatal("未配置恢复逻辑时不应启动协程")
|
||||||
|
}
|
||||||
|
if err := application.Close(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestManagementRecoveryWaitsForHTTPS(t *testing.T) {
|
||||||
|
application := &App{HTTP: httpapi.New(httpapi.Dependencies{})}
|
||||||
|
called := make(chan struct{}, 1)
|
||||||
|
application.startManagementRecovery(func(context.Context) error {
|
||||||
|
called <- struct{}{}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err := application.Close(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-called:
|
||||||
|
t.Fatal("HTTPS 尚未监听时不能开始恢复")
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -84,6 +84,16 @@ func runApplication(ctx context.Context, cfg config.Config, path string) (bool,
|
|||||||
defer service.Close()
|
defer service.Close()
|
||||||
errorsChannel := make(chan error, 1)
|
errorsChannel := make(chan error, 1)
|
||||||
go func() { errorsChannel <- service.HTTP.ListenAndServe() }()
|
go func() { errorsChannel <- service.HTTP.ListenAndServe() }()
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return false, nil
|
||||||
|
case err := <-errorsChannel:
|
||||||
|
if errors.Is(err, http.ErrServerClosed) {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
return false, err
|
||||||
|
case <-service.HTTP.Listening():
|
||||||
|
}
|
||||||
if service.Renewal != nil {
|
if service.Renewal != nil {
|
||||||
service.Renewal.Start(ctx)
|
service.Renewal.Start(ctx)
|
||||||
}
|
}
|
||||||
|
|||||||
88
backend/internal/httpapi/listener_test.go
Normal file
88
backend/internal/httpapi/listener_test.go
Normal file
@ -0,0 +1,88 @@
|
|||||||
|
package httpapi
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"crypto/tls"
|
||||||
|
"errors"
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
"strconv"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestHTTPSListeningSignalFollowsAcceptLoop(t *testing.T) {
|
||||||
|
f := newRotationFixture(t)
|
||||||
|
server := New(Dependencies{Config: f.server.config})
|
||||||
|
host, err := server.prepareHTTPServer()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-server.Listening():
|
||||||
|
t.Fatal("准备证书不代表端口已监听")
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
done <- host.ServeTLS(&servingListener{Listener: listener, ready: server.httpListening}, "", "")
|
||||||
|
}()
|
||||||
|
t.Cleanup(func() {
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
if err := server.Shutdown(ctx); err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
if !errors.Is(err, http.ErrServerClosed) {
|
||||||
|
t.Error(err)
|
||||||
|
}
|
||||||
|
case <-ctx.Done():
|
||||||
|
t.Error("HTTPS 服务没有退出")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
select {
|
||||||
|
case <-server.Listening():
|
||||||
|
case err := <-done:
|
||||||
|
t.Fatal("HTTPS 在监听前退出", err)
|
||||||
|
case <-time.After(3 * time.Second):
|
||||||
|
t.Fatal("HTTPS 接收循环未发布监听信号")
|
||||||
|
}
|
||||||
|
for range 2 {
|
||||||
|
connection, err := tls.DialWithDialer(&net.Dialer{Timeout: 3 * time.Second}, "tcp", listener.Addr().String(), &tls.Config{
|
||||||
|
RootCAs: f.pool, ServerName: "management.example.test", MinVersion: tls.VersionTLS12,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal("收到监听信号后无法建立严格 TLS 连接", err)
|
||||||
|
}
|
||||||
|
_ = connection.Close()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHTTPSBindFailureDoesNotSignalListening(t *testing.T) {
|
||||||
|
f := newRotationFixture(t)
|
||||||
|
_, port, err := net.SplitHostPort(f.address)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
cfg := f.server.config
|
||||||
|
cfg.Server.Port, err = strconv.Atoi(port)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
server := New(Dependencies{Config: cfg})
|
||||||
|
defer server.Shutdown(t.Context())
|
||||||
|
if err := server.ListenAndServe(); err == nil {
|
||||||
|
t.Fatal("被占用端口不能启动第二个监听器")
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-server.Listening():
|
||||||
|
t.Fatal("绑定失败不能报告已监听")
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -117,6 +117,7 @@ type Server struct {
|
|||||||
httpMu sync.Mutex
|
httpMu sync.Mutex
|
||||||
httpStopped bool
|
httpStopped bool
|
||||||
httpServer *http.Server
|
httpServer *http.Server
|
||||||
|
httpListening chan struct{}
|
||||||
tlsMu sync.Mutex
|
tlsMu sync.Mutex
|
||||||
tlsCertificate atomic.Pointer[tls.Certificate]
|
tlsCertificate atomic.Pointer[tls.Certificate]
|
||||||
tlsRoots *x509.CertPool
|
tlsRoots *x509.CertPool
|
||||||
@ -132,7 +133,8 @@ func New(deps Dependencies) *Server {
|
|||||||
frontendDir = filepath.Join(deps.Config.Paths.Install, "web")
|
frontendDir = filepath.Join(deps.Config.Paths.Install, "web")
|
||||||
}
|
}
|
||||||
server := &Server{
|
server := &Server{
|
||||||
config: deps.Config, configPath: deps.ConfigPath, frontendDir: filepath.Clean(frontendDir), logger: deps.Logger,
|
httpListening: make(chan struct{}),
|
||||||
|
config: deps.Config, configPath: deps.ConfigPath, frontendDir: filepath.Clean(frontendDir), logger: deps.Logger,
|
||||||
repository: deps.Repository, setupDNS: deps.SetupDNS, secrets: deps.Secrets, firewall: deps.Firewall, router: router,
|
repository: deps.Repository, setupDNS: deps.SetupDNS, secrets: deps.Secrets, firewall: deps.Firewall, router: router,
|
||||||
servers: deps.Servers,
|
servers: deps.Servers,
|
||||||
users: deps.Users,
|
users: deps.Users,
|
||||||
@ -229,7 +231,28 @@ func (s *Server) ListenAndServe() error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
return server.ListenAndServeTLS("", "")
|
listener, err := net.Listen("tcp", server.Addr)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer listener.Close()
|
||||||
|
return server.ServeTLS(&servingListener{Listener: listener, ready: s.httpListening}, "", "")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Listening 仅在 TLS 服务进入接收循环后关闭,不代表数据库或代理入口已经就绪。
|
||||||
|
func (s *Server) Listening() <-chan struct{} {
|
||||||
|
return s.httpListening
|
||||||
|
}
|
||||||
|
|
||||||
|
type servingListener struct {
|
||||||
|
net.Listener
|
||||||
|
once sync.Once
|
||||||
|
ready chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *servingListener) Accept() (net.Conn, error) {
|
||||||
|
l.once.Do(func() { close(l.ready) })
|
||||||
|
return l.Listener.Accept()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Server) prepareHTTPServer() (*http.Server, error) {
|
func (s *Server) prepareHTTPServer() (*http.Server, error) {
|
||||||
|
|||||||
@ -222,7 +222,7 @@ func (c Coordinator) withSession(ctx context.Context, operation func(*os.Root, M
|
|||||||
if c.Proxy == nil || c.Upstream == nil {
|
if c.Proxy == nil || c.Upstream == nil {
|
||||||
return errors.New("管理证书协调器尚未配置")
|
return errors.New("管理证书协调器尚未配置")
|
||||||
}
|
}
|
||||||
if err := validateDirectory(c.Directory); err != nil {
|
if err := validatePrivateDirectory(c.Directory); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
lockPath := filepath.Join(c.Directory, ".coordination.lock")
|
lockPath := filepath.Join(c.Directory, ".coordination.lock")
|
||||||
|
|||||||
@ -148,6 +148,23 @@ func TestCoordinatorCommitsOverLocalIPC(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestCoordinatorUsesLocalizedStateDirectory(t *testing.T) {
|
||||||
|
f := newCoordinatorFixture(t)
|
||||||
|
f.coordinator.Directory = filepath.Join(f.coordinator.Directory, "管理证书 状态")
|
||||||
|
if err := os.Mkdir(f.coordinator.Directory, 0o700); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
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" {
|
||||||
|
t.Fatal("中文状态目录中的协调未提交", record.Phase)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestCoordinatorRecoversLostResponses(t *testing.T) {
|
func TestCoordinatorRecoversLostResponses(t *testing.T) {
|
||||||
for _, operation := range []string{"stage", "commit"} {
|
for _, operation := range []string{"stage", "commit"} {
|
||||||
t.Run(operation, func(t *testing.T) {
|
t.Run(operation, func(t *testing.T) {
|
||||||
|
|||||||
@ -102,15 +102,19 @@ func newID() (string, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func validateDirectory(directory string) error {
|
func validateDirectory(directory string) error {
|
||||||
if !filepath.IsAbs(directory) || filepath.Clean(directory) != directory || directory == string(filepath.Separator) {
|
|
||||||
return errors.New("Nginx 激活目录必须是独立的绝对目录")
|
|
||||||
}
|
|
||||||
// 目录来自本地部署策略,仍拒绝可改变 Nginx 指令语义的字符。
|
// 目录来自本地部署策略,仍拒绝可改变 Nginx 指令语义的字符。
|
||||||
for _, char := range directory {
|
for _, char := range directory {
|
||||||
if !(char >= 'a' && char <= 'z' || char >= 'A' && char <= 'Z' || char >= '0' && char <= '9' || char == '/' || char == '-' || char == '_' || char == '.') {
|
if !(char >= 'a' && char <= 'z' || char >= 'A' && char <= 'Z' || char >= '0' && char <= '9' || char == '/' || char == '-' || char == '_' || char == '.') {
|
||||||
return errors.New("Nginx 激活目录包含不允许的字符")
|
return errors.New("Nginx 激活目录包含不允许的字符")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
return validatePrivateDirectory(directory)
|
||||||
|
}
|
||||||
|
|
||||||
|
func validatePrivateDirectory(directory string) error {
|
||||||
|
if !filepath.IsAbs(directory) || filepath.Clean(directory) != directory || directory == string(filepath.Separator) {
|
||||||
|
return errors.New("激活状态目录必须是独立的规范绝对目录")
|
||||||
|
}
|
||||||
for path := directory; ; path = filepath.Dir(path) {
|
for path := directory; ; path = filepath.Dir(path) {
|
||||||
info, err := os.Lstat(path)
|
info, err := os.Lstat(path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@ -51,6 +51,45 @@ func TestActivationDirectoryTrust(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestPrivateDirectoryKeepsTrustWithoutNginxSyntax(t *testing.T) {
|
||||||
|
base := t.TempDir()
|
||||||
|
for _, name := range []string{"管理证书", "application data", "literal;directory"} {
|
||||||
|
directory := filepath.Join(base, name)
|
||||||
|
if err := os.Mkdir(directory, 0o700); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := validatePrivateDirectory(directory); err != nil {
|
||||||
|
t.Fatalf("应用私有状态目录不应套用 Nginx 指令字符限制:%v", err)
|
||||||
|
}
|
||||||
|
if err := validateDirectory(directory); err == nil {
|
||||||
|
t.Fatal("实际写入 Nginx 配置的目录仍须限制字符")
|
||||||
|
}
|
||||||
|
if err := os.Chmod(directory, 0o750); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := validatePrivateDirectory(directory); err == nil {
|
||||||
|
t.Fatal("应用状态目录仍须保持私有权限")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
parent := filepath.Join(base, "parent")
|
||||||
|
child := filepath.Join(parent, "child")
|
||||||
|
if err := os.MkdirAll(child, 0o700); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := os.Chmod(parent, 0o777); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
link := filepath.Join(base, "link")
|
||||||
|
if err := os.Symlink(child, link); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
for _, directory := range []string{child, link, "relative", "/", base + "/../escape"} {
|
||||||
|
if err := validatePrivateDirectory(directory); err == nil {
|
||||||
|
t.Fatalf("不应接受不可信应用状态目录:%q", directory)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestActivationFilesRejectLinksAndSpecialFiles(t *testing.T) {
|
func TestActivationFilesRejectLinksAndSpecialFiles(t *testing.T) {
|
||||||
base := t.TempDir()
|
base := t.TempDir()
|
||||||
root, err := os.OpenRoot(base)
|
root, err := os.OpenRoot(base)
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user