diff --git a/backend/internal/httpapi/server.go b/backend/internal/httpapi/server.go index 54f7082..7958b13 100644 --- a/backend/internal/httpapi/server.go +++ b/backend/internal/httpapi/server.go @@ -212,11 +212,7 @@ func (s *Server) registerRoutes() { domainAPI.POST("/domains", s.dnsOperation, s.createDomain) domainAPI.PUT("/domains/:id", s.dnsOperation, s.updateDomain) domainAPI.DELETE("/domains/:id", s.deleteDomain) - adminAPI.POST("/domains/:id/issue", s.longOperation(certificateTaskTimeout), s.issueDomain) - adminAPI.GET("/tasks", s.listTasks) - adminAPI.GET("/tasks/:id", s.getTask) - adminAPI.DELETE("/tasks/:id", s.cancelTask) - adminAPI.GET("/tasks/:id/attempts", s.listTaskAttempts) + domainAPI.POST("/domains/:id/issue", s.longOperation(certificateTaskTimeout), s.issueDomain) adminAPI.GET("/settings", s.getSettings) adminAPI.PUT("/settings", s.updateSettings) diff --git a/backend/internal/httpapi/task_api.go b/backend/internal/httpapi/task_api.go index 94966e9..036135e 100644 --- a/backend/internal/httpapi/task_api.go +++ b/backend/internal/httpapi/task_api.go @@ -20,20 +20,33 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { s.writeError(c, http.StatusNotImplemented, "证书申请接口尚未配置", errors.New("证书申请工作流未配置")) return } - if s.repository == nil || s.tasks == nil { - s.writeError(c, http.StatusNotImplemented, "证书任务接口尚未配置", errors.New("域名仓储或任务仓储未配置")) + if s.tasks == nil || s.projectResources == nil || s.projectMembers == nil { + s.writeError(c, http.StatusServiceUnavailable, "证书任务权限服务尚未就绪", errors.New("任务、项目资源或项目权限服务未配置")) return } domainID := strings.TrimSpace(c.Param("id")) - resource, err := s.repository.GetByID(c.Request.Context(), domainID) + resource, err := s.projectResources.GetVisibleDomain(c.Request.Context(), c.GetString(authenticatedActorKey), domainID) if err != nil { - if repository.IsNotFound(err) { - s.writeError(c, http.StatusNotFound, "域名不存在", err) + if errors.Is(err, repository.ErrProjectAccessDenied) || errors.Is(err, sql.ErrNoRows) { + s.writeError(c, http.StatusForbidden, "无权管理此项目的证书任务", nil) return } s.writeError(c, http.StatusInternalServerError, "读取域名失败", err) return } + authorizer, ok := s.projectMembers.(repository.ProjectAuthorizer) + if !ok { + s.writeError(c, http.StatusServiceUnavailable, "证书任务权限服务尚未就绪", errors.New("项目授权服务未配置")) + return + } + if err := authorizer.AuthorizeProject(c.Request.Context(), c.GetString(authenticatedActorKey), resource.ResourceScope.Normalize().ProjectID, domain.PermissionCertificateManage); err != nil { + if errors.Is(err, repository.ErrProjectAccessDenied) { + s.writeError(c, http.StatusForbidden, "无权管理此项目的证书任务", nil) + return + } + s.writeError(c, http.StatusInternalServerError, "检查证书任务权限失败", err) + return + } s.issueTaskMu.Lock() idempotencyKey := strings.TrimSpace(c.GetHeader("Idempotency-Key")) if idempotencyKey != "" { @@ -187,16 +200,29 @@ func (s *Server) unregisterTaskCancel(taskID string) { delete(s.taskCancels, taskID) } -func (s *Server) cancelTask(c *gin.Context) { +func (s *Server) projectTaskReader(c *gin.Context) (repository.ProjectTaskReader, bool) { if s.tasks == nil { s.writeError(c, http.StatusNotImplemented, "任务管理接口尚未配置", errors.New("任务仓储未配置")) + return nil, false + } + reader, ok := s.tasks.(repository.ProjectTaskReader) + if !ok { + s.writeError(c, http.StatusNotImplemented, "任务权限接口尚未配置", errors.New("项目任务读取服务未配置")) + return nil, false + } + return reader, true +} + +func (s *Server) cancelTask(c *gin.Context) { + reader, ok := s.projectTaskReader(c) + if !ok { return } taskID := strings.TrimSpace(c.Param("id")) - item, err := s.tasks.GetTask(c.Request.Context(), taskID) + item, err := reader.GetVisibleTask(c.Request.Context(), c.GetString(authenticatedActorKey), taskID) if err != nil { - if errors.Is(err, sql.ErrNoRows) { - s.writeError(c, http.StatusNotFound, "任务不存在", err) + if errors.Is(err, sql.ErrNoRows) || errors.Is(err, repository.ErrProjectAccessDenied) { + s.writeError(c, http.StatusForbidden, "无权访问此任务", nil) return } s.writeError(c, http.StatusInternalServerError, "读取任务失败", err) @@ -215,7 +241,11 @@ func (s *Server) cancelTask(c *gin.Context) { return } finished := time.Now().UTC() - if err := s.tasks.CancelTask(c.Request.Context(), taskID, "任务已由用户取消", &finished); err != nil { + if err := reader.CancelVisibleTask(c.Request.Context(), c.GetString(authenticatedActorKey), taskID, "任务已由用户取消", &finished); err != nil { + if errors.Is(err, repository.ErrProjectAccessDenied) || errors.Is(err, sql.ErrNoRows) { + s.writeError(c, http.StatusForbidden, "无权访问此任务", nil) + return + } s.writeError(c, http.StatusConflict, "取消任务失败", err) return } @@ -237,7 +267,6 @@ func (s *Server) cancelTask(c *gin.Context) { } c.JSON(http.StatusOK, gin.H{"task": item}) } - func validTaskIdempotencyKey(value string) bool { if len(value) < 8 || len(value) > 128 { return false @@ -250,12 +279,12 @@ func validTaskIdempotencyKey(value string) bool { return true } func (s *Server) listTasks(c *gin.Context) { - if s.tasks == nil { - s.writeError(c, http.StatusNotImplemented, "任务管理接口尚未配置", errors.New("任务仓储未配置")) + reader, ok := s.projectTaskReader(c) + if !ok { return } limit, _ := strconv.Atoi(c.DefaultQuery("limit", "50")) - items, err := s.tasks.ListTasks(c.Request.Context(), c.Query("domain_id"), limit) + items, err := reader.ListVisibleTasks(c.Request.Context(), c.GetString(authenticatedActorKey), c.Query("domain_id"), limit) if err != nil { s.writeError(c, http.StatusInternalServerError, "查询任务失败", err) return @@ -264,14 +293,14 @@ func (s *Server) listTasks(c *gin.Context) { } func (s *Server) getTask(c *gin.Context) { - if s.tasks == nil { - s.writeError(c, http.StatusNotImplemented, "任务管理接口尚未配置", errors.New("任务仓储未配置")) + reader, ok := s.projectTaskReader(c) + if !ok { return } - item, err := s.tasks.GetTask(c.Request.Context(), c.Param("id")) + item, err := reader.GetVisibleTask(c.Request.Context(), c.GetString(authenticatedActorKey), c.Param("id")) if err != nil { - if errors.Is(err, sql.ErrNoRows) { - s.writeError(c, http.StatusNotFound, "任务不存在", err) + if errors.Is(err, sql.ErrNoRows) || errors.Is(err, repository.ErrProjectAccessDenied) { + s.writeError(c, http.StatusForbidden, "无权访问此任务", nil) return } s.writeError(c, http.StatusInternalServerError, "读取任务失败", err) @@ -281,26 +310,26 @@ func (s *Server) getTask(c *gin.Context) { } func (s *Server) listTaskAttempts(c *gin.Context) { - if s.tasks == nil { - s.writeError(c, http.StatusNotImplemented, "任务管理接口尚未配置", errors.New("任务仓储未配置")) + reader, ok := s.projectTaskReader(c) + if !ok { return } - if _, err := s.tasks.GetTask(c.Request.Context(), c.Param("id")); err != nil { - if errors.Is(err, sql.ErrNoRows) { - s.writeError(c, http.StatusNotFound, "任务不存在", err) + actorID := c.GetString(authenticatedActorKey) + if _, err := reader.GetVisibleTask(c.Request.Context(), actorID, c.Param("id")); err != nil { + if errors.Is(err, sql.ErrNoRows) || errors.Is(err, repository.ErrProjectAccessDenied) { + s.writeError(c, http.StatusForbidden, "无权访问此任务", nil) return } s.writeError(c, http.StatusInternalServerError, "读取任务失败", err) return } - items, err := s.tasks.ListAttempts(c.Request.Context(), c.Param("id")) + items, err := reader.ListVisibleTaskAttempts(c.Request.Context(), actorID, c.Param("id")) if err != nil { s.writeError(c, http.StatusInternalServerError, "查询任务尝试失败", err) return } c.JSON(http.StatusOK, gin.H{"items": items}) } - func (s *Server) listAuditLogs(c *gin.Context) { if !s.requireProjectResourceReader(c) { return diff --git a/backend/internal/repository/project_task_reader.go b/backend/internal/repository/project_task_reader.go new file mode 100644 index 0000000..ca2a756 --- /dev/null +++ b/backend/internal/repository/project_task_reader.go @@ -0,0 +1,177 @@ +package repository + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/domain" +) + +// ProjectTaskReader 只返回当前账户有权查看的任务和任务尝试。 +type ProjectTaskReader interface { + ListVisibleTasks(context.Context, string, string, int) ([]Task, error) + GetVisibleTask(context.Context, string, string) (Task, error) + ListVisibleTaskAttempts(context.Context, string, string) ([]TaskAttempt, error) + CancelVisibleTask(context.Context, string, string, string, *time.Time) error +} + +var _ ProjectTaskReader = (*SQLTaskRepository)(nil) + +func taskVisibilityPredicate(alias string) (string, []any) { + roles := []domain.ProjectRole{ + domain.ProjectRoleAdmin, + domain.ProjectRoleAuditViewer, + domain.ProjectRoleReadOnly, + } + placeholders := make([]string, 0, len(roles)) + args := []any{""} + for _, role := range roles { + placeholders = append(placeholders, "?") + args = append(args, string(role)) + } + return `EXISTS ( + SELECT 1 FROM users task_actor + WHERE task_actor.id = ? AND task_actor.status = 'active' + AND (task_actor.role IN ('super_admin', 'admin') OR ( + task_actor.role = 'user' AND ` + alias + `.scope_type = 'project' + AND EXISTS (SELECT 1 FROM project_members task_member + WHERE task_member.project_id = ` + alias + `.project_id + AND task_member.user_id = task_actor.id + AND task_member.role IN (` + strings.Join(placeholders, ", ") + `)))) + )`, args +} + +func (r *SQLTaskRepository) ListVisibleTasks(ctx context.Context, actorID, domainID string, limit int) ([]Task, error) { + if r == nil || r.db == nil { + return nil, errors.New("数据库连接不能为空") + } + if strings.TrimSpace(actorID) == "" { + return nil, ErrProjectAccessDenied + } + if limit <= 0 || limit > 200 { + limit = 50 + } + predicate, args := taskVisibilityPredicate("tasks") + args[0] = strings.TrimSpace(actorID) + query := taskSelect + ` WHERE ` + predicate + if domainID = strings.TrimSpace(domainID); domainID != "" { + query += " AND tasks.domain_id = ?" + args = append(args, domainID) + } + query += fmt.Sprintf(" ORDER BY tasks.created_at DESC, tasks.id DESC LIMIT %d", limit) + rows, err := r.db.QueryContext(ctx, r.query(query), args...) + if err != nil { + return nil, fmt.Errorf("查询可见任务失败:%w", err) + } + defer rows.Close() + items := make([]Task, 0) + for rows.Next() { + item, err := scanTask(rows) + if err != nil { + return nil, err + } + items = append(items, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("读取可见任务失败:%w", err) + } + return items, nil +} + +func (r *SQLTaskRepository) GetVisibleTask(ctx context.Context, actorID, taskID string) (Task, error) { + if r == nil || r.db == nil { + return Task{}, errors.New("数据库连接不能为空") + } + predicate, args := taskVisibilityPredicate("tasks") + args[0] = strings.TrimSpace(actorID) + args = append(args, strings.TrimSpace(taskID)) + return scanTask(r.db.QueryRowContext(ctx, r.query(taskSelect+` WHERE `+predicate+` AND tasks.id = ?`), args...)) +} + +func (r *SQLTaskRepository) ListVisibleTaskAttempts(ctx context.Context, actorID, taskID string) ([]TaskAttempt, error) { + if r == nil || r.db == nil { + return nil, errors.New("数据库连接不能为空") + } + predicate, args := taskVisibilityPredicate("tasks") + args[0] = strings.TrimSpace(actorID) + args = append(args, strings.TrimSpace(taskID)) + rows, err := r.db.QueryContext(ctx, r.query(`SELECT attempts.id, attempts.task_id, attempts.attempt_number, + attempts.stage, attempts.status, attempts.message, attempts.started_at, attempts.finished_at, + attempts.created_at, COALESCE(tasks.scope_type, 'unknown'), COALESCE(tasks.project_id, '') + FROM task_attempts attempts JOIN tasks ON tasks.id = attempts.task_id + WHERE `+predicate+` AND attempts.task_id = ? ORDER BY attempts.attempt_number`), args...) + if err != nil { + return nil, fmt.Errorf("查询可见任务尝试失败:%w", err) + } + defer rows.Close() + items := make([]TaskAttempt, 0) + for rows.Next() { + item, err := scanAttempt(rows) + if err != nil { + return nil, err + } + items = append(items, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("读取可见任务尝试失败:%w", err) + } + return items, nil +} + +func (r *SQLTaskRepository) CancelVisibleTask(ctx context.Context, actorID, taskID, message string, finishedAt *time.Time) error { + if r == nil || r.db == nil { + return errors.New("数据库连接不能为空") + } + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("开始取消任务事务失败:%w", err) + } + rollback := func() { _ = tx.Rollback() } + predicate, predicateArgs := taskVisibilityPredicate("tasks") + predicateArgs[0] = strings.TrimSpace(actorID) + args := []any{message, nullableTime(finishedAt), time.Now().UTC(), strings.TrimSpace(taskID)} + args = append(args, predicateArgs...) + result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ? + WHERE id = ? AND status IN ('queued', 'running') AND `+predicate), args...) + if err != nil { + rollback() + return fmt.Errorf("取消任务失败:%w", err) + } + count, err := result.RowsAffected() + if err != nil { + rollback() + return fmt.Errorf("读取任务取消结果失败:%w", err) + } + if count == 0 { + item, getErr := scanTask(tx.QueryRowContext(ctx, r.query(taskSelect+` WHERE id = ?`), strings.TrimSpace(taskID))) + if errors.Is(getErr, sql.ErrNoRows) { + rollback() + return sql.ErrNoRows + } + if getErr != nil { + rollback() + return getErr + } + if item.Status == "cancelled" { + if err := tx.Commit(); err != nil { + return fmt.Errorf("提交重复取消任务事务失败:%w", err) + } + return nil + } + rollback() + return ErrProjectAccessDenied + } + if _, err := tx.ExecContext(ctx, r.query(`UPDATE task_attempts SET status = 'cancelled', message = ?, finished_at = ? + WHERE task_id = ? AND status IN ('queued', 'running')`), message, nullableTime(finishedAt), strings.TrimSpace(taskID)); err != nil { + rollback() + return fmt.Errorf("取消任务尝试失败:%w", err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("提交取消任务事务失败:%w", err) + } + return nil +}