Merge pull request #1266 from trheyi/main

Enhance job and execution access control with authorization checks

job/data.go

@@ -23,6 +23,7 @@ var JobFields = []interface{}{
 	"max_retry_count", "default_timeout", "priority", "created_by",
 	"next_run_at", "last_run_at", "current_execution_id", "config",
 	"sort", "enabled", "system", "readonly", "created_at", "updated_at",
+	"__yao_created_by", "__yao_updated_by", "__yao_team_id", "__yao_tenant_id",
 }
 
 // CategoryFields defines the fields to select for category queries

openapi/job/executions.go

@@ -9,16 +9,38 @@ import (
 	"github.com/yaoapp/gou/process"
 	"github.com/yaoapp/kun/log"
 	"github.com/yaoapp/yao/job"
+	"github.com/yaoapp/yao/openapi/oauth/authorized"
 )
 
 // ListExecutions lists executions for a specific job
 func ListExecutions(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	jobID := c.Param("jobID")
 	if jobID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
 		return
 	}
 
+	// Get the job first to check access
+	jobInstance, err := job.GetJob(jobID)
+	if err != nil {
+		log.Error("Failed to get job %s: %v", jobID, err)
+		if err.Error() == "job not found: "+jobID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	// Check if user has access to this job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		return
+	}
+
 	// Get executions for the job
 	executions, err := job.GetExecutions(jobID)
 	if err != nil {
@@ -78,6 +100,9 @@ func ListExecutions(c *gin.Context) {
 
 // GetExecution gets a specific execution by ID
 func GetExecution(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	executionID := c.Param("executionID")
 	if executionID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"})
@@ -96,11 +121,28 @@ func GetExecution(c *gin.Context) {
 		return
 	}
 
+	// Get the job to check access
+	jobInstance, err := job.GetJob(execution.JobID)
+	if err != nil {
+		log.Error("Failed to get job %s for execution %s: %v", execution.JobID, executionID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Check if user has access to the job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Execution not found"})
+		return
+	}
+
 	c.JSON(http.StatusOK, execution)
 }
 
 // StopExecution stops a running execution
 func StopExecution(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	executionID := c.Param("executionID")
 	if executionID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"})
@@ -127,6 +169,12 @@ func StopExecution(c *gin.Context) {
 		return
 	}
 
+	// Check if user has access to the job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Execution not found"})
+		return
+	}
+
 	// For now, we stop the entire job since individual execution stopping
 	// would require more complex implementation in the job package
 	err = jobInstance.Stop()
@@ -146,6 +194,9 @@ func StopExecution(c *gin.Context) {
 
 // GetExecutionProgress gets execution progress information
 func GetExecutionProgress(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	executionID := c.Param("executionID")
 	if executionID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"})
@@ -164,6 +215,20 @@ func GetExecutionProgress(c *gin.Context) {
 		return
 	}
 
+	// Get the job to check access
+	jobInstance, err := job.GetJob(execution.JobID)
+	if err != nil {
+		log.Error("Failed to get job %s for execution %s: %v", execution.JobID, executionID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Check if user has access to the job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Execution not found"})
+		return
+	}
+
 	response := gin.H{
 		"execution_id":  executionID,
 		"job_id":        execution.JobID,

openapi/job/filter.go

@@ -0,0 +1,90 @@
+package job
+
+import (
+	"github.com/gin-gonic/gin"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/yao/job"
+	"github.com/yaoapp/yao/openapi/oauth/authorized"
+	"github.com/yaoapp/yao/openapi/oauth/types"
+)
+
+// AuthFilter applies permission-based filtering to query wheres
+// This function builds where clauses based on the user's authorization constraints
+// It supports TeamOnly and OwnerOnly constraints for data access control
+//
+// Note: Unlike the kb module, job doesn't have 'public' and 'share' fields,
+// so the filtering is simpler and based only on __yao_team_id and __yao_created_by
+//
+// Parameters:
+//   - c: gin.Context containing authorization information
+//   - authInfo: authorized information extracted from the context
+//
+// Returns:
+//   - []model.QueryWhere: array of where clauses to apply to the query
+func AuthFilter(c *gin.Context, authInfo *types.AuthorizedInfo) []model.QueryWhere {
+	if authInfo == nil {
+		return []model.QueryWhere{}
+	}
+
+	var wheres []model.QueryWhere
+	scope := authInfo.AccessScope()
+
+	// Team only - User can access:
+	// 1. Records in their team where __yao_team_id matches
+	if authInfo.Constraints.TeamOnly && authorized.IsTeamMember(c) {
+		wheres = append(wheres, model.QueryWhere{
+			Column: "__yao_team_id",
+			Value:  scope.TeamID,
+		})
+		return wheres
+	}
+
+	// Owner only - User can access:
+	// 1. Records they created where:
+	//    - __yao_team_id is null (not team records)
+	//    - __yao_created_by matches their user ID
+	if authInfo.Constraints.OwnerOnly && authInfo.UserID != "" {
+		wheres = append(wheres, model.QueryWhere{
+			Wheres: []model.QueryWhere{
+				{Column: "__yao_team_id", OP: "null"},
+				{Column: "__yao_created_by", Value: scope.CreatedBy},
+			},
+		})
+		return wheres
+	}
+
+	return wheres
+}
+
+// HasJobAccess checks if the current user has access to a specific job
+// This is useful for checking access to job-related resources like executions and logs
+//
+// Parameters:
+//   - c: gin.Context containing authorization information
+//   - authInfo: authorized information extracted from the context
+//   - jobInstance: the job instance to check access for
+//
+// Returns:
+//   - bool: true if the user has access to the job, false otherwise
+func HasJobAccess(c *gin.Context, authInfo *types.AuthorizedInfo, jobInstance *job.Job) bool {
+	if authInfo == nil {
+		// No auth info means public access (or no auth required)
+		return true
+	}
+
+	scope := authInfo.AccessScope()
+
+	// Team only - Check if job belongs to user's team
+	if authInfo.Constraints.TeamOnly && authorized.IsTeamMember(c) {
+		return jobInstance.YaoTeamID == scope.TeamID
+	}
+
+	// Owner only - Check if job was created by user and not in a team
+	if authInfo.Constraints.OwnerOnly && authInfo.UserID != "" {
+		return jobInstance.YaoTeamID == "" && jobInstance.YaoCreatedBy == scope.CreatedBy
+	}
+
+	// No constraints means access is allowed
+	return true
+}
+

openapi/job/jobs.go

@@ -9,10 +9,14 @@ import (
 	"github.com/yaoapp/gou/process"
 	"github.com/yaoapp/kun/log"
 	"github.com/yaoapp/yao/job"
+	"github.com/yaoapp/yao/openapi/oauth/authorized"
 )
 
 // ListJobs lists jobs with pagination
 func ListJobs(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	// Parse pagination parameters
 	page := 1
 	pagesize := 20
@@ -36,9 +40,15 @@ func ListJobs(c *gin.Context) {
 		},
 	}
 
+	// Add filters
+	var wheres []model.QueryWhere
+
+	// Apply permission-based filtering
+	wheres = append(wheres, AuthFilter(c, authInfo)...)
+
 	// Add status filter if provided
 	if status := c.Query("status"); status != "" {
-		param.Wheres = append(param.Wheres, model.QueryWhere{
+		wheres = append(wheres, model.QueryWhere{
 			Column: "status",
 			Value:  status,
 		})
@@ -46,7 +56,7 @@ func ListJobs(c *gin.Context) {
 
 	// Add category filter if provided
 	if categoryID := c.Query("category_id"); categoryID != "" {
-		param.Wheres = append(param.Wheres, model.QueryWhere{
+		wheres = append(wheres, model.QueryWhere{
 			Column: "category_id",
 			Value:  categoryID,
 		})
@@ -54,7 +64,7 @@ func ListJobs(c *gin.Context) {
 
 	// Add keywords filter if provided (search in name and description)
 	if keywords := c.Query("keywords"); keywords != "" {
-		param.Wheres = append(param.Wheres, model.QueryWhere{
+		wheres = append(wheres, model.QueryWhere{
 			Wheres: []model.QueryWhere{
 				{
 					Column: "name",
@@ -74,12 +84,12 @@ func ListJobs(c *gin.Context) {
 	// Add enabled filter (default to show all for debugging)
 	switch enabled := c.Query("enabled"); enabled {
 	case "true", "1", "yes", "on":
-		param.Wheres = append(param.Wheres, model.QueryWhere{
+		wheres = append(wheres, model.QueryWhere{
 			Column: "enabled",
 			Value:  true,
 		})
 	case "false", "0", "no", "off":
-		param.Wheres = append(param.Wheres, model.QueryWhere{
+		wheres = append(wheres, model.QueryWhere{
 			Column: "enabled",
 			Value:  false,
 		})
@@ -87,6 +97,9 @@ func ListJobs(c *gin.Context) {
 		// Default: show all records regardless of enabled status
 	}
 
+	// Apply all filters to param
+	param.Wheres = wheres
+
 	// Call job.ListJobs function
 	result, err := job.ListJobs(param, page, pagesize)
 	if err != nil {
@@ -100,6 +113,9 @@ func ListJobs(c *gin.Context) {
 
 // GetJob gets a specific job by ID
 func GetJob(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	jobID := c.Param("jobID")
 	if jobID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
@@ -118,11 +134,20 @@ func GetJob(c *gin.Context) {
 		return
 	}
 
+	// Check if user has access to this job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		return
+	}
+
 	c.JSON(http.StatusOK, jobInstance)
 }
 
 // StopJob stops a running job
 func StopJob(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	jobID := c.Param("jobID")
 	if jobID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
@@ -141,6 +166,12 @@ func StopJob(c *gin.Context) {
 		return
 	}
 
+	// Check if user has access to this job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		return
+	}
+
 	// Stop the job
 	err = jobInstance.Stop()
 	if err != nil {
@@ -158,6 +189,9 @@ func StopJob(c *gin.Context) {
 
 // GetJobProgress gets job progress information
 func GetJobProgress(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	jobID := c.Param("jobID")
 	if jobID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
@@ -176,6 +210,12 @@ func GetJobProgress(c *gin.Context) {
 		return
 	}
 
+	// Check if user has access to this job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		return
+	}
+
 	// Get executions for progress calculation
 	executions, err := job.GetExecutions(jobID)
 	if err != nil {
@@ -225,8 +265,16 @@ func GetJobProgress(c *gin.Context) {
 
 // GetStats gets overall job statistics
 func GetStats(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
+	// Build base auth filter
+	baseAuthFilter := AuthFilter(c, authInfo)
+
 	// Count total jobs
-	totalJobs, err := job.CountJobs(model.QueryParam{})
+	totalJobs, err := job.CountJobs(model.QueryParam{
+		Wheres: baseAuthFilter,
+	})
 	if err != nil {
 		log.Error("Failed to count total jobs: %v", err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
@@ -235,9 +283,9 @@ func GetStats(c *gin.Context) {
 
 	// Count running jobs
 	runningJobs, err := job.CountJobs(model.QueryParam{
-		Wheres: []model.QueryWhere{
+		Wheres: append(baseAuthFilter, model.QueryWhere{
-			{Column: "status", Value: "running"},
+			Column: "status", Value: "running",
-		},
+		}),
 	})
 	if err != nil {
 		log.Error("Failed to count running jobs: %v", err)
@@ -246,9 +294,9 @@ func GetStats(c *gin.Context) {
 
 	// Count completed jobs
 	completedJobs, err := job.CountJobs(model.QueryParam{
-		Wheres: []model.QueryWhere{
+		Wheres: append(baseAuthFilter, model.QueryWhere{
-			{Column: "status", Value: "completed"},
+			Column: "status", Value: "completed",
-		},
+		}),
 	})
 	if err != nil {
 		log.Error("Failed to count completed jobs: %v", err)
@@ -257,9 +305,9 @@ func GetStats(c *gin.Context) {
 
 	// Count failed jobs
 	failedJobs, err := job.CountJobs(model.QueryParam{
-		Wheres: []model.QueryWhere{
+		Wheres: append(baseAuthFilter, model.QueryWhere{
-			{Column: "status", Value: "failed"},
+			Column: "status", Value: "failed",
-		},
+		}),
 	})
 	if err != nil {
 		log.Error("Failed to count failed jobs: %v", err)
@@ -276,9 +324,9 @@ func GetStats(c *gin.Context) {
 	categoryStats := make(map[string]int)
 	for _, category := range categories {
 		count, err := job.CountJobs(model.QueryParam{
-			Wheres: []model.QueryWhere{
+			Wheres: append(baseAuthFilter, model.QueryWhere{
-				{Column: "category_id", Value: category.CategoryID},
+				Column: "category_id", Value: category.CategoryID,
-			},
+			}),
 		})
 		if err != nil {
 			count = 0

openapi/job/logs.go

@@ -9,16 +9,38 @@ import (
 	"github.com/yaoapp/gou/process"
 	"github.com/yaoapp/kun/log"
 	"github.com/yaoapp/yao/job"
+	"github.com/yaoapp/yao/openapi/oauth/authorized"
 )
 
 // ListLogs lists logs for a specific job
 func ListLogs(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	jobID := c.Param("jobID")
 	if jobID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
 		return
 	}
 
+	// Get the job first to check access
+	jobInstance, err := job.GetJob(jobID)
+	if err != nil {
+		log.Error("Failed to get job %s: %v", jobID, err)
+		if err.Error() == "job not found: "+jobID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	// Check if user has access to this job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		return
+	}
+
 	// Parse pagination parameters
 	page := 1
 	pagesize := 50
@@ -71,6 +93,9 @@ func ListLogs(c *gin.Context) {
 
 // ListExecutionLogs lists logs for a specific execution
 func ListExecutionLogs(c *gin.Context) {
+	// Get authorized information
+	authInfo := authorized.GetInfo(c)
+
 	executionID := c.Param("executionID")
 	if executionID == "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"})
@@ -89,6 +114,20 @@ func ListExecutionLogs(c *gin.Context) {
 		return
 	}
 
+	// Get the job to check access
+	jobInstance, err := job.GetJob(execution.JobID)
+	if err != nil {
+		log.Error("Failed to get job %s for execution %s: %v", execution.JobID, executionID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Check if user has access to the job
+	if !HasJobAccess(c, authInfo, jobInstance) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Execution not found"})
+		return
+	}
+
 	// Parse pagination parameters
 	page := 1
 	pagesize := 50