开发修复:收紧域名读取并保护任务创建竞态
更新域名改用项目可见资源读取;任务创建在事务中锁定所属域名,和删除流程共用资源锁,防止删除检查后的在途任务竞态。
This commit is contained in:
parent
68bf2b9838
commit
82448c9ad1
@ -30,7 +30,12 @@ func TestDomainResourceScopeAPI(t *testing.T) {
|
||||
}
|
||||
}
|
||||
repo := repository.NewSQLDomainRepository(db.DB)
|
||||
server := New(Dependencies{Config: cfg, Repository: repo, ProjectDomains: repository.NewSQLProjectMemberRepository(db.DB)})
|
||||
projectResources := repository.NewSQLProjectMemberRepository(db.DB)
|
||||
server := New(Dependencies{
|
||||
Config: cfg, Repository: repo,
|
||||
ProjectResources: projectResources,
|
||||
ProjectDomains: projectResources,
|
||||
})
|
||||
request := func(method, path, body, user string) *httptest.ResponseRecorder {
|
||||
req := httptest.NewRequest(method, path, strings.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
@ -150,7 +150,11 @@ func TestDomainAPIValidatesSANAndHidesInternalErrors(t *testing.T) {
|
||||
func TestDeleteMissingDomainReturnsNotFound(t *testing.T) {
|
||||
cfg := readyTestConfig(t.TempDir())
|
||||
repo := &stubDomainRepository{deleteErr: sql.ErrNoRows}
|
||||
server := New(Dependencies{Config: cfg, Repository: repo, ProjectDomains: projectDomainRepositoryStub{DomainRepository: repo}})
|
||||
server := New(Dependencies{
|
||||
Config: cfg, Repository: repo,
|
||||
ProjectResources: projectResourceReaderStub{domains: repo},
|
||||
ProjectDomains: projectDomainRepositoryStub{DomainRepository: repo},
|
||||
})
|
||||
response := httptest.NewRecorder()
|
||||
serveAdminHTTP(server, response, httptest.NewRequest(http.MethodDelete, "/api/v1/domains/missing", nil))
|
||||
if response.Code != http.StatusNotFound || strings.Contains(response.Body.String(), "detail") {
|
||||
@ -162,7 +166,11 @@ func TestUpdateDomainAPI(t *testing.T) {
|
||||
cfg := readyTestConfig(t.TempDir())
|
||||
loaded := domain.Domain{ID: "domain-1", Name: "example.com", CertificateType: domain.CertificateSingle, SANs: []string{"example.com"}, Status: "active"}
|
||||
repo := &stubDomainRepository{loaded: loaded}
|
||||
server := New(Dependencies{Config: cfg, Repository: repo, ProjectDomains: projectDomainRepositoryStub{DomainRepository: repo}})
|
||||
server := New(Dependencies{
|
||||
Config: cfg, Repository: repo,
|
||||
ProjectResources: projectResourceReaderStub{domains: repo},
|
||||
ProjectDomains: projectDomainRepositoryStub{DomainRepository: repo},
|
||||
})
|
||||
request := httptest.NewRequest(http.MethodPut, "/api/v1/domains/domain-1", strings.NewReader(`{"name":"example.com","certificate_type":"san","sans":["www.example.com"],"auto_renew":false,"acme_environment":"production","dns_provider":"cloudflare"}`))
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
response := httptest.NewRecorder()
|
||||
@ -192,7 +200,11 @@ func TestUpdateDomainRejectsInvalidEnvironment(t *testing.T) {
|
||||
cfg := readyTestConfig(t.TempDir())
|
||||
loaded := domain.Domain{ID: "domain-1", Name: "example.com"}
|
||||
repo := &stubDomainRepository{loaded: loaded}
|
||||
server := New(Dependencies{Config: cfg, Repository: repo, ProjectDomains: projectDomainRepositoryStub{DomainRepository: repo}})
|
||||
server := New(Dependencies{
|
||||
Config: cfg, Repository: repo,
|
||||
ProjectResources: projectResourceReaderStub{domains: repo},
|
||||
ProjectDomains: projectDomainRepositoryStub{DomainRepository: repo},
|
||||
})
|
||||
request := httptest.NewRequest(http.MethodPut, "/api/v1/domains/domain-1", strings.NewReader(`{"name":"example.com","certificate_type":"single","acme_environment":"invalid"}`))
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
@ -602,7 +602,7 @@ func (s *Server) createDomain(c *gin.Context) {
|
||||
}
|
||||
|
||||
func (s *Server) updateDomain(c *gin.Context) {
|
||||
if !s.requireProjectDomainRepository(c) || !s.requireDomainRepository(c) {
|
||||
if !s.requireProjectDomainRepository(c) || !s.requireProjectResourceReader(c) {
|
||||
return
|
||||
}
|
||||
var request domainRequest
|
||||
@ -610,13 +610,9 @@ func (s *Server) updateDomain(c *gin.Context) {
|
||||
s.writeError(c, http.StatusBadRequest, "请求参数无效", err)
|
||||
return
|
||||
}
|
||||
existing, err := s.repository.GetByID(c.Request.Context(), c.Param("id"))
|
||||
existing, err := s.projectResources.GetVisibleDomain(c.Request.Context(), c.GetString(authenticatedActorKey), c.Param("id"))
|
||||
if err != nil {
|
||||
if repository.IsNotFound(err) {
|
||||
s.writeError(c, http.StatusNotFound, "域名不存在", err)
|
||||
return
|
||||
}
|
||||
s.writeError(c, http.StatusInternalServerError, "读取域名失败", err)
|
||||
s.writeProjectResourceError(c, err)
|
||||
return
|
||||
}
|
||||
if !s.authorizeDomainMutation(c, existing.ProjectID) {
|
||||
|
||||
@ -88,26 +88,52 @@ func (r *SQLTaskRepository) CreateTask(ctx context.Context, item Task) error {
|
||||
item.ID, item.DomainID, item.TaskType, item.Status, item.Stage, item.Message, item.OperatorID,
|
||||
item.CertificateVersionID, item.IdempotencyKey, nullableTime(item.StartedAt), nullableTime(item.FinishedAt), now, updated,
|
||||
}
|
||||
query := `INSERT INTO tasks
|
||||
insertQuery := `INSERT INTO tasks
|
||||
(id, domain_id, task_type, status, stage, message, operator_id, certificate_version_id, idempotency_key, started_at, finished_at, created_at, updated_at, scope_type, project_id) `
|
||||
if item.TaskType == "management_renewal" {
|
||||
if item.DomainID != ManagementResourceID || !item.HistoryScope.matches(HistoryScope{ScopeType: ScopeSystem}) {
|
||||
return errors.New("管理证书任务范围无效")
|
||||
}
|
||||
query += `VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'system', '')`
|
||||
} else {
|
||||
if item.DomainID == ManagementResourceID || (item.ScopeType != "" && item.ScopeType != ScopeProject) {
|
||||
return errors.New("业务任务必须归属于实际域名项目")
|
||||
result, err := r.db.ExecContext(ctx, r.query(insertQuery+`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'system', '')`), args...)
|
||||
if err != nil {
|
||||
return fmt.Errorf("创建任务失败:%w", err)
|
||||
}
|
||||
// 一条语句读取资源归属并写入快照,不能先检查再插入悬空任务。
|
||||
query += `SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'project', d.project_id
|
||||
FROM domains d JOIN projects p ON p.id = d.project_id WHERE d.id = ? AND (? = '' OR d.project_id = ?)`
|
||||
args = append(args, item.DomainID, item.ProjectID, item.ProjectID)
|
||||
return validateTaskInsertResult(result)
|
||||
}
|
||||
result, err := r.db.ExecContext(ctx, r.query(query), args...)
|
||||
if item.DomainID == ManagementResourceID || (item.ScopeType != "" && item.ScopeType != ScopeProject) {
|
||||
return errors.New("业务任务必须归属于实际域名项目")
|
||||
}
|
||||
|
||||
// 创建任务与删除域名使用相同的域名行锁,避免删除检查完成后又插入在途任务。
|
||||
tx, err := r.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("开始任务创建事务失败:%w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
var projectID string
|
||||
if err := tx.QueryRowContext(ctx, r.query(`SELECT d.project_id FROM domains d
|
||||
JOIN projects p ON p.id = d.project_id WHERE d.id = ? AND (? = '' OR d.project_id = ?)`+taskResourceWriteLock(r.driver)),
|
||||
item.DomainID, item.ProjectID, item.ProjectID).Scan(&projectID); err != nil {
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return errors.New("任务资源不存在或项目不匹配")
|
||||
}
|
||||
return fmt.Errorf("读取任务所属域名失败:%w", err)
|
||||
}
|
||||
args = append(args, projectID)
|
||||
result, err := tx.ExecContext(ctx, r.query(insertQuery+`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'project', ?)`), args...)
|
||||
if err != nil {
|
||||
return fmt.Errorf("创建任务失败:%w", err)
|
||||
}
|
||||
if err := validateTaskInsertResult(result); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("提交任务创建事务失败:%w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateTaskInsertResult(result sql.Result) error {
|
||||
if count, err := result.RowsAffected(); err != nil {
|
||||
return fmt.Errorf("读取任务创建结果失败:%w", err)
|
||||
} else if count != 1 {
|
||||
@ -116,6 +142,15 @@ func (r *SQLTaskRepository) CreateTask(ctx context.Context, item Task) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func taskResourceWriteLock(driver string) string {
|
||||
switch driver {
|
||||
case "mysql", "postgres":
|
||||
return " FOR UPDATE"
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
func (r *SQLTaskRepository) GetTask(ctx context.Context, id string) (Task, error) {
|
||||
if r == nil || r.db == nil {
|
||||
return Task{}, errors.New("数据库连接不能为空")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user