diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index 842d866..3a5822f 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -143,7 +143,9 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string tasks: taskRepo, audit: auditRepo, notifier: notifier, notifyError: func(err error) { logger.Error("管理证书续期或通知失败", "error", err) }, } - renewal := certificate.NewRenewalSchedulerWithManagement(repo, workflow, certificate.RenewalPolicy{ + trackedWorkflow := &certificateTaskWorkflow{workflow: workflow, tasks: taskRepo, audit: auditRepo} + trackedWorkflow.logError = func(err error) { logger.Error("证书续期或部署任务失败", "error", err) } + renewal := certificate.NewRenewalSchedulerWithManagement(repo, trackedWorkflow, certificate.RenewalPolicy{ CheckInterval: time.Duration(cfg.ACME.RenewalCheckIntervalDays) * 24 * time.Hour, NearExpiryThreshold: time.Duration(cfg.ACME.RenewalNearExpiryDays) * 24 * time.Hour, RenewalThreshold: time.Duration(cfg.ACME.RenewalThresholdDays) * 24 * time.Hour, @@ -166,7 +168,7 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string return result.Certificate, result.Deployment, err }, DeployCertificate: func(ctx context.Context, domainID string) (certificate.IssueResult, deployment.Report, error) { - result, err := workflow.DeployCurrent(ctx, domainID) + result, err := trackedWorkflow.DeployCurrent(ctx, domainID) return result.Certificate, result.Deployment, err }, RestoreCertificateVersion: func(ctx context.Context, domainID, version string) (certificate.IssueResult, error) { diff --git a/backend/internal/app/certificate_tasks.go b/backend/internal/app/certificate_tasks.go new file mode 100644 index 0000000..699a575 --- /dev/null +++ b/backend/internal/app/certificate_tasks.go @@ -0,0 +1,126 @@ +package app + +import ( + "context" + "errors" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/certificate" + "git.pcux.cn/xuans/Scert/backend/internal/domain" + "git.pcux.cn/xuans/Scert/backend/internal/repository" +) + +type certificateTaskWorkflow struct { + workflow *CertificateWorkflow + tasks repository.TaskRepository + audit repository.AuditRepository + logError func(error) +} + +func (w *certificateTaskWorkflow) Issue(ctx context.Context, id string) (certificate.IssueResult, error) { + return w.workflow.Issue(ctx, id) +} + +func (w *certificateTaskWorkflow) Renew(ctx context.Context, id string, urgent bool, policy certificate.RenewalPolicy) error { + attempts := 1 + if urgent && policy.MaxRetries > 1 { + attempts = policy.MaxRetries + } + _, err := w.run(ctx, id, "renewal_run", "certificate_issue", attempts, policy.RetryInterval, w.workflow.IssueWithReport) + return err +} + +func (w *certificateTaskWorkflow) DeployCurrent(ctx context.Context, id string) (CertificateWorkflowResult, error) { + return w.run(ctx, id, "certificate_deploy", "certificate_deploy", 1, 0, w.workflow.DeployCurrent) +} + +func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage string, attempts int, interval time.Duration, execute func(context.Context, string) (CertificateWorkflowResult, error)) (result CertificateWorkflowResult, resultErr error) { + if w.tasks == nil || w.audit == nil { + return result, errors.New("证书任务记录服务未初始化") + } + started := time.Now().UTC() + task := repository.Task{ + ID: domain.NewID(), DomainID: id, TaskType: taskType, + Status: "running", Stage: stage, Message: "正在执行证书工作流", + StartedAt: &started, CreatedAt: started, + } + if err := w.tasks.CreateTask(ctx, task); err != nil { + return result, err + } + defer func() { + finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + finished := time.Now().UTC() + status, message := certificateTaskOutcome(resultErr) + auditErr := w.audit.CreateAuditLog(finishCtx, repository.AuditLog{ + Action: taskType, ResourceType: "domain", ResourceID: id, + Status: status, Message: message, + }) + if auditErr != nil { + status, message = certificateTaskOutcome(auditErr) + stage = "record" + } + taskErr := w.tasks.UpdateTask(finishCtx, task.ID, status, stage, message, &started, &finished) + resultErr = errors.Join(resultErr, taskErr, auditErr) + if resultErr != nil && w.logError != nil { + w.logError(resultErr) + } + }() + if err := w.audit.CreateAuditLog(ctx, repository.AuditLog{ + Action: taskType, ResourceType: "domain", ResourceID: id, + Status: "running", Message: "开始执行证书工作流", + }); err != nil { + return result, err + } + for number := 1; number <= attempts; number++ { + if err := ctx.Err(); err != nil { + return result, err + } + if err := w.tasks.UpdateTask(ctx, task.ID, "running", stage, "正在执行证书工作流", &started, nil); err != nil { + return result, err + } + attemptStarted := time.Now().UTC() + attempt := repository.TaskAttempt{ + ID: domain.NewID(), TaskID: task.ID, AttemptNumber: number, + Stage: stage, Status: "running", Message: "正在执行证书工作流", + StartedAt: &attemptStarted, CreatedAt: attemptStarted, + } + if err := w.tasks.CreateAttempt(ctx, attempt); err != nil { + return result, err + } + issueCtx, cancel := context.WithTimeout(ctx, 30*time.Minute) + result, resultErr = execute(issueCtx, id) + cancel() + finishCtx, finishCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + finished := time.Now().UTC() + status, message := certificateTaskOutcome(resultErr) + recordErr := w.tasks.FinishAttempt(finishCtx, attempt.ID, status, message, &finished) + finishCancel() + if recordErr != nil { + return result, errors.Join(resultErr, recordErr) + } + if resultErr == nil || number == attempts || result.Certificate.Saved.CertificatePath != "" || errors.Is(resultErr, context.Canceled) || errors.Is(resultErr, context.DeadlineExceeded) { + return result, resultErr + } + if err := w.tasks.UpdateTask(ctx, task.ID, "running", "retry_wait", "证书申请失败，等待重试", &started, nil); err != nil { + return result, errors.Join(resultErr, err) + } + if err := waitForManagementRetry(ctx, interval); err != nil { + return result, errors.Join(resultErr, err) + } + } + return result, resultErr +} + +func certificateTaskOutcome(err error) (string, string) { + if errors.Is(err, context.Canceled) { + return "cancelled", "证书工作流已取消" + } + if err != nil { + return "failed", "证书工作流未完成，请检查受保护的服务日志" + } + return "success", "证书工作流执行完成" +} + +var _ certificate.CertificateIssuer = (*certificateTaskWorkflow)(nil) +var _ certificate.RenewalExecutor = (*certificateTaskWorkflow)(nil) diff --git a/backend/internal/app/certificate_tasks_test.go b/backend/internal/app/certificate_tasks_test.go new file mode 100644 index 0000000..cd27cfe --- /dev/null +++ b/backend/internal/app/certificate_tasks_test.go @@ -0,0 +1,254 @@ +package app + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/certificate" + "git.pcux.cn/xuans/Scert/backend/internal/deployment" + "git.pcux.cn/xuans/Scert/backend/internal/domain" + "git.pcux.cn/xuans/Scert/backend/internal/repository" +) + +func newCertificateTaskFixture(t *testing.T) (*certificateTaskWorkflow, *App) { + t.Helper() + a, err := New(t.Context(), readyApplicationConfig(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = a.Close() }) + item, err := domain.NewDomain("task.example.test", domain.CertificateSingle, nil) + if err != nil { + t.Fatal(err) + } + item.ID = "task-domain" + if err := a.Repository.Create(t.Context(), item); err != nil { + t.Fatal(err) + } + return &certificateTaskWorkflow{ + tasks: a.Tasks, audit: repository.NewSQLAuditRepository(a.Database.DB), + }, a +} + +func assertCertificateTask(t *testing.T, a *App, status string, statuses ...string) repository.Task { + t.Helper() + items, err := a.Tasks.ListTasks(t.Context(), "task-domain", 10) + if err != nil || len(items) != 1 { + t.Fatalf("任务数量错误：%+v %v", items, err) + } + item := items[0] + if item.Status != status || item.StartedAt == nil || item.FinishedAt == nil { + t.Fatalf("任务未正确收尾：%+v", item) + } + attempts, err := a.Tasks.ListAttempts(t.Context(), item.ID) + if err != nil || len(attempts) != len(statuses) { + t.Fatalf("尝试数量错误：%+v %v", attempts, err) + } + for i, attempt := range attempts { + if attempt.AttemptNumber != i+1 || attempt.Status != statuses[i] || attempt.FinishedAt == nil { + t.Fatalf("尝试结果错误：%+v", attempt) + } + if strings.Contains(attempt.Message, "secret-path") { + t.Fatal("尝试记录泄露内部错误") + } + } + var count int + if err := a.Database.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM audit_logs WHERE action = ? AND resource_id = ? AND status = ?", item.TaskType, item.DomainID, status).Scan(&count); err != nil || count != 1 { + t.Fatalf("审计记录缺失：%d %v", count, err) + } + return item +} + +func TestCertificateTasksRecordSuccessFailureAndCancellation(t *testing.T) { + for _, status := range []string{"success", "failed", "cancelled"} { + t.Run(status, func(t *testing.T) { + w, a := newCertificateTaskFixture(t) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + _, err := w.run(ctx, "task-domain", "renewal_run", "certificate_issue", 1, 0, func(context.Context, string) (CertificateWorkflowResult, error) { + if status == "cancelled" { + cancel() + return CertificateWorkflowResult{}, context.Canceled + } + if status == "failed" { + return CertificateWorkflowResult{}, errors.New("secret-path") + } + return CertificateWorkflowResult{}, nil + }) + if (err != nil) != (status != "success") { + t.Fatalf("执行结果错误：%v", err) + } + assertCertificateTask(t, a, status, status) + }) + } +} + +func TestCertificateTasksKeepRetriesInOneTask(t *testing.T) { + w, a := newCertificateTaskFixture(t) + calls := 0 + _, err := w.run(t.Context(), "task-domain", "renewal_run", "certificate_issue", 3, time.Nanosecond, func(context.Context, string) (CertificateWorkflowResult, error) { + calls++ + if calls < 3 { + return CertificateWorkflowResult{}, errors.New("secret-path") + } + return CertificateWorkflowResult{}, nil + }) + if err != nil || calls != 3 { + t.Fatalf("重试次数错误：%d %v", calls, err) + } + assertCertificateTask(t, a, "success", "failed", "failed", "success") +} + +func TestCertificateTasksDoNotReissueAfterDeploymentFailure(t *testing.T) { + w, a := newCertificateTaskFixture(t) + calls := 0 + _, err := w.run(t.Context(), "task-domain", "renewal_run", "certificate_issue", 3, time.Nanosecond, func(context.Context, string) (CertificateWorkflowResult, error) { + calls++ + return CertificateWorkflowResult{Certificate: certificate.IssueResult{Saved: certificate.SavedResource{CertificatePath: "cert.pem"}}}, errors.New("secret-path") + }) + if err == nil || calls != 1 { + t.Fatalf("部署失败不应重复签发：%d %v", calls, err) + } + assertCertificateTask(t, a, "failed", "failed") +} + +type failingCertificateTasks struct { + repository.TaskRepository + failCreate bool + failFinish bool +} + +func (r failingCertificateTasks) CreateTask(ctx context.Context, task repository.Task) error { + if r.failCreate { + return errors.New("task-store-unavailable") + } + return r.TaskRepository.CreateTask(ctx, task) +} + +func (r failingCertificateTasks) FinishAttempt(ctx context.Context, id, status, message string, at *time.Time) error { + if r.failFinish { + return errors.New("attempt-store-unavailable") + } + return r.TaskRepository.FinishAttempt(ctx, id, status, message, at) +} + +func TestCertificateTaskPersistenceFailuresStopExecutionOrRetries(t *testing.T) { + for _, failCreate := range []bool{true, false} { + w, a := newCertificateTaskFixture(t) + w.tasks = failingCertificateTasks{TaskRepository: a.Tasks, failCreate: failCreate, failFinish: !failCreate} + calls := 0 + _, err := w.run(t.Context(), "task-domain", "renewal_run", "certificate_issue", 3, time.Nanosecond, func(context.Context, string) (CertificateWorkflowResult, error) { + calls++ + return CertificateWorkflowResult{}, errors.New("secret-path") + }) + if err == nil || (failCreate && calls != 0) || (!failCreate && calls != 1) { + t.Fatalf("记录失败未阻止副作用或重复执行：%d %v", calls, err) + } + } +} + +func TestCertificateTaskEntryPointsRecordRenewalAndDeployment(t *testing.T) { + for _, renew := range []bool{true, false} { + w, a := newCertificateTaskFixture(t) + service := &workflowCertificateService{result: certificate.IssueResult{ + DomainID: "task-domain", Saved: certificate.SavedResource{CertificatePath: "cert.pem"}, + }} + w.workflow = &CertificateWorkflow{ + Certificates: service, Domains: &workflowDomainState{}, + Servers: workflowServerLister{servers: []deployment.Server{{ID: "server"}}}, + Deployment: &workflowDeployer{report: deployment.Report{Total: 1, Success: 1, Status: "success"}}, + CurrentCertificate: func(context.Context, string) (certificate.IssueResult, error) { return service.result, nil }, + } + var err error + if renew { + err = w.Renew(t.Context(), "task-domain", false, certificate.RenewalPolicy{}) + } else { + _, err = w.DeployCurrent(t.Context(), "task-domain") + } + if err != nil || service.called != renew { + t.Fatalf("任务入口执行错误：%v", err) + } + item := assertCertificateTask(t, a, "success", "success") + if (renew && item.TaskType != "renewal_run") || (!renew && item.TaskType != "certificate_deploy") { + t.Fatalf("任务类型错误：%+v", item) + } + } +} + +type failingCertificateAudit struct { + repository.AuditRepository + failStatus string + failure error +} + +func (r failingCertificateAudit) CreateAuditLog(ctx context.Context, item repository.AuditLog) error { + if item.Status == r.failStatus { + return r.failure + } + return r.AuditRepository.CreateAuditLog(ctx, item) +} + +type failingCertificateTerminal struct { + repository.TaskRepository + failure error +} + +func (r failingCertificateTerminal) UpdateTask(ctx context.Context, id, status, stage, message string, started, finished *time.Time) error { + if finished != nil { + return r.failure + } + return r.TaskRepository.UpdateTask(ctx, id, status, stage, message, started, finished) +} + +func TestCertificateAuditFailuresPersistFailedTaskWithoutReissue(t *testing.T) { + for _, failStatus := range []string{"running", "success"} { + t.Run(failStatus, func(t *testing.T) { + w, a := newCertificateTaskFixture(t) + failure := errors.New("audit-store-unavailable") + w.audit = failingCertificateAudit{AuditRepository: w.audit, failStatus: failStatus, failure: failure} + calls := 0 + var logged error + w.logError = func(err error) { logged = err } + _, err := w.run(t.Context(), "task-domain", "renewal_run", "certificate_issue", 3, time.Nanosecond, func(context.Context, string) (CertificateWorkflowResult, error) { + calls++ + return CertificateWorkflowResult{}, nil + }) + expectedCalls := 1 + if failStatus == "running" { + expectedCalls = 0 + } + if !errors.Is(err, failure) || !errors.Is(logged, failure) || calls != expectedCalls { + t.Fatalf("calls=%d error=%v logged=%v", calls, err, logged) + } + tasks, listErr := a.Tasks.ListTasks(t.Context(), "task-domain", 10) + if listErr != nil || len(tasks) != 1 || tasks[0].Status != "failed" || tasks[0].FinishedAt == nil { + t.Fatalf("tasks=%+v error=%v", tasks, listErr) + } + if failStatus == "success" && tasks[0].Stage != "record" { + t.Fatalf("stage=%s", tasks[0].Stage) + } + if strings.Contains(tasks[0].Message, failure.Error()) { + t.Fatal("audit-store-unavailable") + } + }) + } +} + +func TestCertificateTerminalPersistenceFailureIsReturnedWithoutReissue(t *testing.T) { + w, a := newCertificateTaskFixture(t) + failure := errors.New("task-terminal-unavailable") + w.tasks = failingCertificateTerminal{TaskRepository: a.Tasks, failure: failure} + calls := 0 + var logged error + w.logError = func(err error) { logged = err } + _, err := w.run(t.Context(), "task-domain", "renewal_run", "certificate_issue", 3, time.Nanosecond, func(context.Context, string) (CertificateWorkflowResult, error) { + calls++ + return CertificateWorkflowResult{}, nil + }) + if calls != 1 || !errors.Is(err, failure) || !errors.Is(logged, failure) { + t.Fatalf("calls=%d error=%v logged=%v", calls, err, logged) + } +} diff --git a/backend/internal/certificate/renewal.go b/backend/internal/certificate/renewal.go index 28bb902..1707d41 100644 --- a/backend/internal/certificate/renewal.go +++ b/backend/internal/certificate/renewal.go @@ -237,6 +237,9 @@ func (s *RenewalScheduler) Stop() { } func (s *RenewalScheduler) issueWithRetry(ctx context.Context, domainID string, urgent bool, policy RenewalPolicy) error { + if executor, ok := s.issuer.(RenewalExecutor); ok { + return executor.Renew(ctx, domainID, urgent, policy) + } attempts := 1 if urgent { attempts = policy.MaxRetries diff --git a/backend/internal/certificate/renewal_executor.go b/backend/internal/certificate/renewal_executor.go new file mode 100644 index 0000000..b1db1b5 --- /dev/null +++ b/backend/internal/certificate/renewal_executor.go @@ -0,0 +1,7 @@ +package certificate + +import "context" + +type RenewalExecutor interface { + Renew(context.Context, string, bool, RenewalPolicy) error +} diff --git a/backend/internal/certificate/renewal_executor_test.go b/backend/internal/certificate/renewal_executor_test.go new file mode 100644 index 0000000..cf103cd --- /dev/null +++ b/backend/internal/certificate/renewal_executor_test.go @@ -0,0 +1,35 @@ +package certificate + +import ( + "context" + "testing" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/domain" +) + +type recordedRenewalExecutor struct { + calls int + urgent bool + policy RenewalPolicy +} + +func (r *recordedRenewalExecutor) Issue(context.Context, string) (IssueResult, error) { + panic("unexpected fallback") +} + +func (r *recordedRenewalExecutor) Renew(_ context.Context, _ string, urgent bool, policy RenewalPolicy) error { + r.calls++ + r.urgent, r.policy = urgent, policy + return nil +} + +func TestRenewalSchedulerUsesRecordedExecutor(t *testing.T) { + expires := time.Now().Add(time.Hour) + executor := &recordedRenewalExecutor{} + scheduler := NewRenewalScheduler(renewalLister{items: []domain.Domain{{ID: "domain", AutoRenew: true, ExpiresAt: &expires}}}, executor, 6) + report, err := scheduler.RunOnce(t.Context()) + if err != nil || report.Succeeded != 1 || executor.calls != 1 || !executor.urgent || executor.policy.MaxRetries != 3 { + t.Fatalf("调度器未使用带记录的续期入口：%+v %+v %v", report, executor, err) + } +} diff --git a/backend/internal/httpapi/task_api.go b/backend/internal/httpapi/task_api.go index efe7bbe..7c9690a 100644 --- a/backend/internal/httpapi/task_api.go +++ b/backend/internal/httpapi/task_api.go @@ -201,6 +201,10 @@ func (s *Server) cancelTask(c *gin.Context) { c.JSON(http.StatusConflict, gin.H{"error": "任务当前无法取消", "task": item}) return } + if item.TaskType == "renewal_run" || item.TaskType == "certificate_deploy" { + c.JSON(http.StatusConflict, gin.H{"error": "该任务由证书工作流控制，不能单独修改取消状态", "task": item}) + return + } if item.TaskType == "management_renewal" { c.JSON(http.StatusConflict, gin.H{"error": "管理证书任务由续期调度控制，不能单独修改取消状态", "task": item}) return diff --git a/backend/internal/httpapi/workflow_task_cancel_test.go b/backend/internal/httpapi/workflow_task_cancel_test.go new file mode 100644 index 0000000..80d104a --- /dev/null +++ b/backend/internal/httpapi/workflow_task_cancel_test.go @@ -0,0 +1,28 @@ +package httpapi + +import ( + "net/http" + "net/http/httptest" + "testing" + + "git.pcux.cn/xuans/Scert/backend/internal/repository" +) + +func TestTaskAPICannotFalselyCancelWorkflowTasks(t *testing.T) { + for _, kind := range []string{"renewal_run", "certificate_deploy"} { + t.Run(kind, func(t *testing.T) { + server, tasks, _, closeDB := newTaskAPITestServer(t, nil) + defer closeDB() + item := repository.Task{ID: "workflow-task", DomainID: "domain-1", TaskType: kind, Status: "running", Stage: "certificate_issue"} + if err := tasks.CreateTask(t.Context(), item); err != nil { + t.Fatal(err) + } + response := httptest.NewRecorder() + serveAdminHTTP(server, response, httptest.NewRequest(http.MethodDelete, "/api/v1/tasks/"+item.ID, nil)) + stored, err := tasks.GetTask(t.Context(), item.ID) + if response.Code != http.StatusConflict || err != nil || stored.Status != "running" { + t.Fatalf("工作流任务被伪造为已取消：%d %+v %v", response.Code, stored, err) + } + }) + } +} diff --git a/frontend/src/views/Dashboard/Dashboard.test.js b/frontend/src/views/Dashboard/Dashboard.test.js index a375c7e..1e4eb65 100644 --- a/frontend/src/views/Dashboard/Dashboard.test.js +++ b/frontend/src/views/Dashboard/Dashboard.test.js @@ -873,3 +873,43 @@ test("dashboardLayout.usesSharedTooltipForUserManagementDescription", async () = assert.match(template, /[\s\S]*?<\/HelpTip>/); assert.doesNotMatch(template, /

/); }); + + +test("workflowTasks.renderOnlySupportedCancellation", async () => { + const { readFile } = await import("node:fs/promises"); + const { compileTemplate } = await import("@vue/compiler-sfc"); + const { renderToString } = await import("@vue/server-renderer"); + const Vue = await import("vue"); + const source = await readFile(new URL("./RuntimeOverview.vue", import.meta.url), "utf8"); + const start = source.indexOf('", start) + "".length; + const compiled = compileTemplate({ + source: `${source.slice(start, end)}
`, + filename: "RuntimeTaskActions.vue", + id: "runtime-task-actions", + compilerOptions: { mode: "function" }, + }); + assert.deepEqual(compiled.errors, []); + const render = new Function("Vue", compiled.code)(Vue); + for (const taskType of ["certificate_issue", "renewal_run", "certificate_deploy", "management_renewal", "unknown"]) { + for (const status of ["queued", "running", "success", "failed", "cancelled"]) { + const html = await renderToString(Vue.createSSRApp({ + render, + setup: () => ({ + visibleTasks: [{ id: "task", task_type: taskType, status }], + taskTypeLabel: (value) => value, + taskDomain: () => "example.test", + statusClass: () => "warning", + displayLabel: (_, value) => value, + taskStatusLabels: {}, + runtimeTime: () => "2026-09-26", + openTaskDetails: () => {}, + cancelTask: () => {}, + }), + })); + const canCancel = taskType === "certificate_issue" && ["queued", "running"].includes(status); + assert.equal((html.match(/