mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 06:17:41 +00:00
1 parent
501800743f
commit
f9440baca2
15 files changed
+584
-72
No files matched your search
@@ -84,6 +84,9 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
|
||||
cronCtx, cancelCron := context.WithTimeout(context.Background(), jobTimeout)
|
||||
defer cancelCron()
|
||||
cronCtx = store.WithTenantID(cronCtx, job.TenantID)
|
||||
if job.Payload.CredentialUserID != "" {
|
||||
cronCtx = store.WithCredentialUserID(cronCtx, job.Payload.CredentialUserID)
|
||||
}
|
||||
|
||||
// Reset session before each cron run to prevent tool errors from previous
|
||||
// runs from polluting the context and blocking future executions (#294).
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/agent"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
func TestCronJobHandlerInjectsPayloadCredentialUserID(t *testing.T) {
|
||||
wantCredentialUserID := "tenant-user-123"
|
||||
var gotCredentialUserID string
|
||||
|
||||
sched := scheduler.NewScheduler(
|
||||
scheduler.DefaultLanes(),
|
||||
scheduler.QueueConfig{
|
||||
Mode: scheduler.QueueModeQueue,
|
||||
Cap: 1,
|
||||
Drop: scheduler.DropOld,
|
||||
DebounceMs: 0,
|
||||
MaxConcurrent: 1,
|
||||
},
|
||||
func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) {
|
||||
gotCredentialUserID = store.CredentialUserIDFromContext(ctx)
|
||||
return &agent.RunResult{Content: "ok"}, nil
|
||||
},
|
||||
)
|
||||
defer sched.Stop()
|
||||
|
||||
handler := makeCronJobHandler(
|
||||
sched,
|
||||
nil,
|
||||
&config.Config{},
|
||||
nil,
|
||||
nil,
|
||||
nil,
|
||||
)
|
||||
|
||||
result, err := handler(&store.CronJob{
|
||||
ID: uuid.NewString(),
|
||||
TenantID: uuid.New(),
|
||||
Name: "credentialed-report",
|
||||
AgentID: "reporter",
|
||||
UserID: "group:telegram:-100123",
|
||||
Stateless: true,
|
||||
Payload: store.CronPayload{
|
||||
Kind: "agent_turn",
|
||||
Message: "run gh issue list",
|
||||
CredentialUserID: wantCredentialUserID,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("cron handler returned error: %v", err)
|
||||
}
|
||||
if result == nil || result.Content != "ok" {
|
||||
t.Fatalf("cron result = %#v, want content ok", result)
|
||||
}
|
||||
if gotCredentialUserID != wantCredentialUserID {
|
||||
t.Fatalf("credential user ID in scheduled context = %q, want %q", gotCredentialUserID, wantCredentialUserID)
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,21 @@ Significant changes, features, and fixes in reverse chronological order.
|
||||
|
||||
---
|
||||
|
||||
## 2026-05-24
|
||||
|
||||
### Cron SecureCLI credential context
|
||||
|
||||
**Fixes**
|
||||
|
||||
- Fixed cron-triggered agent turns so credentialed CLI lookups preserve the explicit tenant user credential identity captured when the cron job is created.
|
||||
- Kept cron `user_id` ownership unchanged for group-scoped list/remove behavior while storing credential lookup identity separately in cron payload metadata.
|
||||
|
||||
**Tests**
|
||||
|
||||
- Added regression coverage for cron payload credential metadata, legacy payload compatibility, cron scheduler context injection, and SQLite cron persistence parity.
|
||||
|
||||
---
|
||||
|
||||
## 2026-05-22
|
||||
|
||||
### Messaging debounce hardening
|
||||
|
||||
@@ -45,7 +45,7 @@ func (l *Loop) injectContext(ctx context.Context, req *RunRequest) (contextSetup
|
||||
// Resolve merged tenant user identity for credential lookups.
|
||||
// Keeps UserID unchanged (session/workspace scoping) but sets a separate
|
||||
// CredentialUserID for SecureCLI, MCP, and other per-user features.
|
||||
if l.userResolver != nil && req.UserID != "" {
|
||||
if l.userResolver != nil && req.UserID != "" && store.ExplicitCredentialUserIDFromContext(ctx) == "" {
|
||||
credUserID := l.resolveCredentialUserID(ctx, *req)
|
||||
if credUserID != "" && credUserID != req.UserID {
|
||||
ctx = store.WithCredentialUserID(ctx, credUserID)
|
||||
@@ -354,7 +354,7 @@ func (l *Loop) injectContext(ctx context.Context, req *RunRequest) (contextSetup
|
||||
providerName = l.provider.Name()
|
||||
}
|
||||
// Extract resolved credential user ID (set earlier via WithCredentialUserID, empty if not resolved).
|
||||
credUserID, _ := ctx.Value(store.CredentialUserIDKey).(string)
|
||||
credUserID := store.ExplicitCredentialUserIDFromContext(ctx)
|
||||
rc := &store.RunContext{
|
||||
AgentID: l.agentUUID,
|
||||
AgentKey: l.id,
|
||||
|
||||
@@ -52,6 +52,7 @@ func (m *CronMethods) handleList(ctx context.Context, client *gateway.Client, re
|
||||
userID = client.UserID()
|
||||
}
|
||||
jobs := m.service.ListJobs(ctx, params.IncludeDisabled, "", userID)
|
||||
jobs = store.RedactCronJobsCredentialContext(jobs)
|
||||
|
||||
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
|
||||
"jobs": jobs,
|
||||
@@ -112,7 +113,7 @@ func (m *CronMethods) handleCreate(ctx context.Context, client *gateway.Client,
|
||||
}
|
||||
|
||||
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
|
||||
"job": job,
|
||||
"job": store.RedactCronJobCredentialContext(*job),
|
||||
}))
|
||||
emitAudit(m.eventBus, client, "cron.created", "cron", job.ID)
|
||||
}
|
||||
@@ -131,14 +132,17 @@ func (m *CronMethods) handleDelete(ctx context.Context, client *gateway.Client,
|
||||
return
|
||||
}
|
||||
|
||||
job, ok := m.service.GetJob(ctx, params.JobID)
|
||||
if !ok {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, i18n.T(locale, i18n.MsgJobNotFound)))
|
||||
return
|
||||
}
|
||||
if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) {
|
||||
job, ok := m.service.GetJob(ctx, params.JobID)
|
||||
if !ok || job.UserID != client.UserID() {
|
||||
if job.UserID != client.UserID() {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job")))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if err := m.service.RemoveJob(ctx, params.JobID); err != nil {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, err.Error()))
|
||||
return
|
||||
@@ -165,13 +169,23 @@ func (m *CronMethods) handleToggle(ctx context.Context, client *gateway.Client,
|
||||
return
|
||||
}
|
||||
|
||||
job, ok := m.service.GetJob(ctx, params.JobID)
|
||||
if !ok {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, i18n.T(locale, i18n.MsgJobNotFound)))
|
||||
return
|
||||
}
|
||||
if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) {
|
||||
job, ok := m.service.GetJob(ctx, params.JobID)
|
||||
if !ok || job.UserID != client.UserID() {
|
||||
if job.UserID != client.UserID() {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job")))
|
||||
return
|
||||
}
|
||||
}
|
||||
if params.Enabled {
|
||||
if err := store.CheckCronCredentialOwner(ctx, job); err != nil {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job credential context")))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if err := m.service.EnableJob(ctx, params.JobID, params.Enabled); err != nil {
|
||||
code := protocol.ErrInvalidRequest
|
||||
@@ -213,13 +227,21 @@ func (m *CronMethods) handleUpdate(ctx context.Context, client *gateway.Client,
|
||||
return
|
||||
}
|
||||
|
||||
existing, ok := m.service.GetJob(ctx, jobID)
|
||||
if !ok {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, i18n.T(locale, i18n.MsgJobNotFound)))
|
||||
return
|
||||
}
|
||||
if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) {
|
||||
existing, ok := m.service.GetJob(ctx, jobID)
|
||||
if !ok || existing.UserID != client.UserID() {
|
||||
if existing.UserID != client.UserID() {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job")))
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := store.CheckCronCredentialOwner(ctx, existing); err != nil {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job credential context")))
|
||||
return
|
||||
}
|
||||
|
||||
job, err := m.service.UpdateJob(ctx, jobID, params.Patch)
|
||||
if err != nil {
|
||||
@@ -232,7 +254,7 @@ func (m *CronMethods) handleUpdate(ctx context.Context, client *gateway.Client,
|
||||
}
|
||||
|
||||
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
|
||||
"job": job,
|
||||
"job": store.RedactCronJobCredentialContext(*job),
|
||||
}))
|
||||
emitAudit(m.eventBus, client, "cron.updated", "cron", jobID)
|
||||
}
|
||||
@@ -272,6 +294,10 @@ func (m *CronMethods) handleRun(ctx context.Context, client *gateway.Client, req
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := store.CheckCronCredentialOwner(ctx, job); err != nil {
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job credential context")))
|
||||
return
|
||||
}
|
||||
|
||||
// Respond immediately — job execution happens in background
|
||||
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
|
||||
|
||||
@@ -14,12 +14,15 @@ import (
|
||||
// ---- stub CronStore ----
|
||||
|
||||
type stubCronStore struct {
|
||||
jobs map[string]*store.CronJob
|
||||
addErr error
|
||||
removeErr error
|
||||
updateErr error
|
||||
enableErr error
|
||||
addedJob *store.CronJob
|
||||
jobs map[string]*store.CronJob
|
||||
addErr error
|
||||
removeErr error
|
||||
updateErr error
|
||||
enableErr error
|
||||
addedJob *store.CronJob
|
||||
updateCnt int
|
||||
enableCnt int
|
||||
runCnt int
|
||||
}
|
||||
|
||||
func newStubCronStore() *stubCronStore {
|
||||
@@ -36,10 +39,10 @@ func (s *stubCronStore) AddJob(_ context.Context, name string, schedule store.Cr
|
||||
return nil, s.addErr
|
||||
}
|
||||
job := &store.CronJob{
|
||||
ID: "new-job-id",
|
||||
Name: name,
|
||||
UserID: userID,
|
||||
Enabled: true,
|
||||
ID: "new-job-id",
|
||||
Name: name,
|
||||
UserID: userID,
|
||||
Enabled: true,
|
||||
Schedule: schedule,
|
||||
}
|
||||
s.addedJob = job
|
||||
@@ -71,6 +74,7 @@ func (s *stubCronStore) RemoveJob(_ context.Context, jobID string) error {
|
||||
}
|
||||
|
||||
func (s *stubCronStore) UpdateJob(_ context.Context, jobID string, patch store.CronJobPatch) (*store.CronJob, error) {
|
||||
s.updateCnt++
|
||||
if s.updateErr != nil {
|
||||
return nil, s.updateErr
|
||||
}
|
||||
@@ -82,6 +86,7 @@ func (s *stubCronStore) UpdateJob(_ context.Context, jobID string, patch store.C
|
||||
}
|
||||
|
||||
func (s *stubCronStore) EnableJob(_ context.Context, jobID string, enabled bool) error {
|
||||
s.enableCnt++
|
||||
if s.enableErr != nil {
|
||||
return s.enableErr
|
||||
}
|
||||
@@ -98,15 +103,16 @@ func (s *stubCronStore) GetRunLog(_ context.Context, _ string, _, _ int) ([]stor
|
||||
func (s *stubCronStore) Status() map[string]any { return map[string]any{"running": true} }
|
||||
|
||||
// Lifecycle stubs (not called in unit tests)
|
||||
func (s *stubCronStore) Start() error { return nil }
|
||||
func (s *stubCronStore) Stop() {}
|
||||
func (s *stubCronStore) Start() error { return nil }
|
||||
func (s *stubCronStore) Stop() {}
|
||||
func (s *stubCronStore) SetOnJob(_ func(*store.CronJob) (*store.CronJobResult, error)) {}
|
||||
func (s *stubCronStore) SetOnEvent(_ func(store.CronEvent)) {}
|
||||
func (s *stubCronStore) SetOnEvent(_ func(store.CronEvent)) {}
|
||||
func (s *stubCronStore) RunJob(_ context.Context, _ string, _ bool) (bool, string, error) {
|
||||
s.runCnt++
|
||||
return true, "", nil
|
||||
}
|
||||
func (s *stubCronStore) SetDefaultTimezone(_ string) {}
|
||||
func (s *stubCronStore) GetDueJobs(_ time.Time) []store.CronJob { return nil }
|
||||
func (s *stubCronStore) SetDefaultTimezone(_ string) {}
|
||||
func (s *stubCronStore) GetDueJobs(_ time.Time) []store.CronJob { return nil }
|
||||
|
||||
// ---- helpers ----
|
||||
|
||||
@@ -160,7 +166,7 @@ func TestCronCreate_MissingName_ReturnsInvalidRequest(t *testing.T) {
|
||||
m := buildCronMethods(t, svc)
|
||||
client := nullClient()
|
||||
req := cronReqFrame(t, protocol.MethodCronCreate, map[string]any{
|
||||
"message": "hello",
|
||||
"message": "hello",
|
||||
"schedule": map[string]any{"kind": "every", "everyMs": 60000},
|
||||
})
|
||||
m.handleCreate(context.Background(), client, req)
|
||||
@@ -199,8 +205,8 @@ func TestCronCreate_ValidParams_CreatesJob(t *testing.T) {
|
||||
m := buildCronMethods(t, svc)
|
||||
client := nullClient()
|
||||
req := cronReqFrame(t, protocol.MethodCronCreate, map[string]any{
|
||||
"name": "my-daily-job",
|
||||
"message": "do the thing",
|
||||
"name": "my-daily-job",
|
||||
"message": "do the thing",
|
||||
"schedule": map[string]any{"kind": "every", "everyMs": 3600000},
|
||||
})
|
||||
m.handleCreate(context.Background(), client, req)
|
||||
@@ -259,3 +265,90 @@ func TestCronToggle_MissingJobID_ReturnsInvalidRequest(t *testing.T) {
|
||||
m.handleToggle(context.Background(), client, req)
|
||||
// No panic = jobId-required error path hit
|
||||
}
|
||||
|
||||
func TestCronToggle_BlocksCredentialBoundEnableByDifferentUser(t *testing.T) {
|
||||
svc := newStubCronStore()
|
||||
svc.jobs["job-1"] = &store.CronJob{
|
||||
ID: "job-1",
|
||||
UserID: "",
|
||||
Payload: store.CronPayload{
|
||||
CredentialUserID: "tenant-user-a",
|
||||
},
|
||||
}
|
||||
m := buildCronMethods(t, svc)
|
||||
client := nullClient()
|
||||
ctx := store.WithCredentialUserID(context.Background(), "tenant-user-b")
|
||||
|
||||
req := cronReqFrame(t, protocol.MethodCronToggle, map[string]any{"jobId": "job-1", "enabled": true})
|
||||
m.handleToggle(ctx, client, req)
|
||||
|
||||
if svc.enableCnt != 0 {
|
||||
t.Fatalf("EnableJob called %d times, want 0", svc.enableCnt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCronToggle_AllowsCredentialBoundDisableByDifferentUser(t *testing.T) {
|
||||
svc := newStubCronStore()
|
||||
svc.jobs["job-1"] = &store.CronJob{
|
||||
ID: "job-1",
|
||||
UserID: "",
|
||||
Payload: store.CronPayload{
|
||||
CredentialUserID: "tenant-user-a",
|
||||
},
|
||||
}
|
||||
m := buildCronMethods(t, svc)
|
||||
client := nullClient()
|
||||
ctx := store.WithCredentialUserID(context.Background(), "tenant-user-b")
|
||||
|
||||
req := cronReqFrame(t, protocol.MethodCronToggle, map[string]any{"jobId": "job-1", "enabled": false})
|
||||
m.handleToggle(ctx, client, req)
|
||||
|
||||
if svc.enableCnt != 1 {
|
||||
t.Fatalf("EnableJob called %d times, want 1", svc.enableCnt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCronUpdate_BlocksCredentialBoundJobByDifferentUser(t *testing.T) {
|
||||
svc := newStubCronStore()
|
||||
svc.jobs["job-1"] = &store.CronJob{
|
||||
ID: "job-1",
|
||||
UserID: "",
|
||||
Payload: store.CronPayload{
|
||||
CredentialUserID: "tenant-user-a",
|
||||
},
|
||||
}
|
||||
m := buildCronMethods(t, svc)
|
||||
client := nullClient()
|
||||
ctx := store.WithCredentialUserID(context.Background(), "tenant-user-b")
|
||||
|
||||
req := cronReqFrame(t, protocol.MethodCronUpdate, map[string]any{
|
||||
"jobId": "job-1",
|
||||
"patch": map[string]any{"message": "run gh issue list"},
|
||||
})
|
||||
m.handleUpdate(ctx, client, req)
|
||||
|
||||
if svc.updateCnt != 0 {
|
||||
t.Fatalf("UpdateJob called %d times, want 0", svc.updateCnt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCronRun_BlocksCredentialBoundJobByDifferentUser(t *testing.T) {
|
||||
svc := newStubCronStore()
|
||||
svc.jobs["job-1"] = &store.CronJob{
|
||||
ID: "job-1",
|
||||
UserID: "",
|
||||
Payload: store.CronPayload{
|
||||
CredentialUserID: "tenant-user-a",
|
||||
},
|
||||
}
|
||||
m := buildCronMethods(t, svc)
|
||||
client := nullClient()
|
||||
ctx := store.WithCredentialUserID(context.Background(), "tenant-user-b")
|
||||
|
||||
req := cronReqFrame(t, protocol.MethodCronRun, map[string]any{"jobId": "job-1", "mode": "force"})
|
||||
m.handleRun(ctx, client, req)
|
||||
|
||||
if svc.runCnt != 0 {
|
||||
t.Fatalf("RunJob called %d times, want 0", svc.runCnt)
|
||||
}
|
||||
}
|
||||
@@ -116,10 +116,19 @@ func WithCredentialUserID(ctx context.Context, id string) context.Context {
|
||||
return context.WithValue(ctx, CredentialUserIDKey, id)
|
||||
}
|
||||
|
||||
// ExplicitCredentialUserIDFromContext returns only the explicitly injected credential identity.
|
||||
// Unlike CredentialUserIDFromContext, it does not fall back to UserID.
|
||||
func ExplicitCredentialUserIDFromContext(ctx context.Context) string {
|
||||
if v, ok := ctx.Value(CredentialUserIDKey).(string); ok {
|
||||
return v
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// CredentialUserIDFromContext returns the resolved identity for credential lookups.
|
||||
// Falls back to RunContext.CredentialUserID, then UserIDFromContext.
|
||||
func CredentialUserIDFromContext(ctx context.Context) string {
|
||||
if v, ok := ctx.Value(CredentialUserIDKey).(string); ok && v != "" {
|
||||
if v := ExplicitCredentialUserIDFromContext(ctx); v != "" {
|
||||
return v
|
||||
}
|
||||
if rc := RunContextFromCtx(ctx); rc != nil && rc.CredentialUserID != "" {
|
||||
|
||||
@@ -12,8 +12,9 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
ErrCronJobNotFound = errors.New("cron job not found")
|
||||
ErrCronJobNoFutureRun = errors.New("cron job has no future run")
|
||||
ErrCronJobNotFound = errors.New("cron job not found")
|
||||
ErrCronJobNoFutureRun = errors.New("cron job has no future run")
|
||||
ErrCronCredentialOwnerMismatch = errors.New("cron credential context belongs to a different user")
|
||||
)
|
||||
|
||||
// CronJob represents a scheduled job.
|
||||
@@ -48,9 +49,40 @@ type CronSchedule struct {
|
||||
|
||||
// CronPayload describes what a job does when triggered.
|
||||
type CronPayload struct {
|
||||
Kind string `json:"kind" db:"-"`
|
||||
Message string `json:"message" db:"-"`
|
||||
Command string `json:"command,omitempty" db:"-"`
|
||||
Kind string `json:"kind" db:"-"`
|
||||
Message string `json:"message" db:"-"`
|
||||
Command string `json:"command,omitempty" db:"-"`
|
||||
CredentialUserID string `json:"credentialUserId,omitempty" db:"-"`
|
||||
}
|
||||
|
||||
// CheckCronCredentialOwner blocks user-triggered mutation/execution of a
|
||||
// credential-bound cron job unless the caller owns that credential context.
|
||||
func CheckCronCredentialOwner(ctx context.Context, job *CronJob) error {
|
||||
if job == nil || job.Payload.CredentialUserID == "" {
|
||||
return nil
|
||||
}
|
||||
if ExplicitCredentialUserIDFromContext(ctx) == job.Payload.CredentialUserID {
|
||||
return nil
|
||||
}
|
||||
return ErrCronCredentialOwnerMismatch
|
||||
}
|
||||
|
||||
// RedactCronJobCredentialContext returns a response-safe copy of a cron job.
|
||||
func RedactCronJobCredentialContext(job CronJob) CronJob {
|
||||
job.Payload.CredentialUserID = ""
|
||||
return job
|
||||
}
|
||||
|
||||
// RedactCronJobsCredentialContext returns response-safe copies of cron jobs.
|
||||
func RedactCronJobsCredentialContext(jobs []CronJob) []CronJob {
|
||||
if len(jobs) == 0 {
|
||||
return jobs
|
||||
}
|
||||
redacted := make([]CronJob, len(jobs))
|
||||
for i := range jobs {
|
||||
redacted[i] = RedactCronJobCredentialContext(jobs[i])
|
||||
}
|
||||
return redacted
|
||||
}
|
||||
|
||||
// CronJobState tracks runtime state for a job.
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -104,6 +106,80 @@ func TestNextRunForToggle_ExpiredAtReturnsError(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCronPayloadCredentialUserIDJSONRoundTrip(t *testing.T) {
|
||||
payload := CronPayload{
|
||||
Kind: "agent_turn",
|
||||
Message: "run credentialed report",
|
||||
CredentialUserID: "tenant-user-123",
|
||||
}
|
||||
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal payload: %v", err)
|
||||
}
|
||||
if !json.Valid(data) {
|
||||
t.Fatalf("payload JSON is invalid: %s", data)
|
||||
}
|
||||
|
||||
var got CronPayload
|
||||
if err := json.Unmarshal(data, &got); err != nil {
|
||||
t.Fatalf("unmarshal payload: %v", err)
|
||||
}
|
||||
if got.CredentialUserID != payload.CredentialUserID {
|
||||
t.Fatalf("credential user ID = %q, want %q", got.CredentialUserID, payload.CredentialUserID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCronPayloadLegacyJSONKeepsEmptyCredentialUserID(t *testing.T) {
|
||||
var got CronPayload
|
||||
if err := json.Unmarshal([]byte(`{"kind":"agent_turn","message":"legacy"}`), &got); err != nil {
|
||||
t.Fatalf("unmarshal legacy payload: %v", err)
|
||||
}
|
||||
if got.CredentialUserID != "" {
|
||||
t.Fatalf("legacy payload credential user ID = %q, want empty", got.CredentialUserID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckCronCredentialOwnerBlocksMismatch(t *testing.T) {
|
||||
ctx := WithCredentialUserID(context.Background(), "tenant-user-b")
|
||||
job := &CronJob{Payload: CronPayload{CredentialUserID: "tenant-user-a"}}
|
||||
|
||||
if err := CheckCronCredentialOwner(ctx, job); !errors.Is(err, ErrCronCredentialOwnerMismatch) {
|
||||
t.Fatalf("got %v, want ErrCronCredentialOwnerMismatch", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckCronCredentialOwnerIgnoresUserIDFallback(t *testing.T) {
|
||||
ctx := WithUserID(context.Background(), "tenant-user-a")
|
||||
job := &CronJob{Payload: CronPayload{CredentialUserID: "tenant-user-a"}}
|
||||
|
||||
if err := CheckCronCredentialOwner(ctx, job); !errors.Is(err, ErrCronCredentialOwnerMismatch) {
|
||||
t.Fatalf("got %v, want ErrCronCredentialOwnerMismatch", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckCronCredentialOwnerAllowsMatchAndLegacy(t *testing.T) {
|
||||
ctx := WithCredentialUserID(context.Background(), "tenant-user-a")
|
||||
if err := CheckCronCredentialOwner(ctx, &CronJob{Payload: CronPayload{CredentialUserID: "tenant-user-a"}}); err != nil {
|
||||
t.Fatalf("matching credential owner should pass: %v", err)
|
||||
}
|
||||
if err := CheckCronCredentialOwner(context.Background(), &CronJob{}); err != nil {
|
||||
t.Fatalf("legacy job should pass: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRedactCronJobCredentialContext(t *testing.T) {
|
||||
job := CronJob{Payload: CronPayload{CredentialUserID: "tenant-user-a", Message: "run"}}
|
||||
redacted := RedactCronJobCredentialContext(job)
|
||||
|
||||
if redacted.Payload.CredentialUserID != "" {
|
||||
t.Fatalf("redacted credential user ID = %q, want empty", redacted.Payload.CredentialUserID)
|
||||
}
|
||||
if job.Payload.CredentialUserID == "" {
|
||||
t.Fatal("redaction mutated source job")
|
||||
}
|
||||
}
|
||||
|
||||
//go:fix inline
|
||||
func int64Ptr(v int64) *int64 {
|
||||
return new(v)
|
||||
|
||||
@@ -24,7 +24,9 @@ func (s *PGCronStore) AddJob(ctx context.Context, name string, schedule store.Cr
|
||||
}
|
||||
|
||||
payload := store.CronPayload{
|
||||
Kind: "agent_turn", Message: message,
|
||||
Kind: "agent_turn",
|
||||
Message: message,
|
||||
CredentialUserID: store.ExplicitCredentialUserIDFromContext(ctx),
|
||||
}
|
||||
payloadJSON, _ := json.Marshal(payload)
|
||||
|
||||
|
||||
@@ -78,30 +78,30 @@ func validCronRow(payloadJSON []byte) []any {
|
||||
expr := "*/5 * * * *"
|
||||
tz := "UTC"
|
||||
return []any{
|
||||
id, // id
|
||||
tenantID, // tenant_id
|
||||
(*uuid.UUID)(nil), // agent_id
|
||||
(*string)(nil), // user_id
|
||||
"test-job", // name
|
||||
true, // enabled
|
||||
"cron", // schedule_kind
|
||||
&expr, // cron_expression
|
||||
(*time.Time)(nil), // run_at
|
||||
&tz, // timezone
|
||||
(*int64)(nil), // interval_ms
|
||||
payloadJSON, // payload
|
||||
false, // delete_after_run
|
||||
false, // stateless
|
||||
false, // deliver
|
||||
"", // deliver_channel
|
||||
"", // deliver_to
|
||||
false, // wake_heartbeat
|
||||
(*time.Time)(nil), // next_run_at
|
||||
(*time.Time)(nil), // last_run_at
|
||||
(*string)(nil), // last_status
|
||||
(*string)(nil), // last_error
|
||||
now, // created_at
|
||||
now, // updated_at
|
||||
id, // id
|
||||
tenantID, // tenant_id
|
||||
(*uuid.UUID)(nil), // agent_id
|
||||
(*string)(nil), // user_id
|
||||
"test-job", // name
|
||||
true, // enabled
|
||||
"cron", // schedule_kind
|
||||
&expr, // cron_expression
|
||||
(*time.Time)(nil), // run_at
|
||||
&tz, // timezone
|
||||
(*int64)(nil), // interval_ms
|
||||
payloadJSON, // payload
|
||||
false, // delete_after_run
|
||||
false, // stateless
|
||||
false, // deliver
|
||||
"", // deliver_channel
|
||||
"", // deliver_to
|
||||
false, // wake_heartbeat
|
||||
(*time.Time)(nil), // next_run_at
|
||||
(*time.Time)(nil), // last_run_at
|
||||
(*string)(nil), // last_status
|
||||
(*string)(nil), // last_error
|
||||
now, // created_at
|
||||
now, // updated_at
|
||||
}
|
||||
}
|
||||
|
||||
@@ -123,17 +123,18 @@ func TestScanCronRow_MalformedPayloadJSON(t *testing.T) {
|
||||
// TestScanCronRow_ValidPayload verifies correct behavior with valid JSON.
|
||||
func TestScanCronRow_ValidPayload(t *testing.T) {
|
||||
payload := store.CronPayload{
|
||||
Kind: "message",
|
||||
Message: "hello world",
|
||||
Kind: "message",
|
||||
Message: "hello world",
|
||||
CredentialUserID: "tenant-user-123",
|
||||
}
|
||||
payloadJSON, _ := json.Marshal(payload)
|
||||
|
||||
// Build row with deliver/deliver_channel/deliver_to set as dedicated columns.
|
||||
rowVals := validCronRow(payloadJSON)
|
||||
// Indices: stateless=13, deliver=14, deliver_channel=15, deliver_to=16, wake_heartbeat=17
|
||||
rowVals[14] = true // deliver
|
||||
rowVals[14] = true // deliver
|
||||
rowVals[15] = "telegram" // deliver_channel
|
||||
rowVals[16] = "user123" // deliver_to
|
||||
rowVals[16] = "user123" // deliver_to
|
||||
row := &mockRowScanner{values: rowVals}
|
||||
|
||||
job, err := scanCronRow(row)
|
||||
@@ -144,6 +145,9 @@ func TestScanCronRow_ValidPayload(t *testing.T) {
|
||||
if job.Payload.Message != "hello world" {
|
||||
t.Errorf("expected Message 'hello world', got %q", job.Payload.Message)
|
||||
}
|
||||
if job.Payload.CredentialUserID != "tenant-user-123" {
|
||||
t.Errorf("expected CredentialUserID 'tenant-user-123', got %q", job.Payload.CredentialUserID)
|
||||
}
|
||||
if job.DeliverChannel != "telegram" {
|
||||
t.Errorf("expected DeliverChannel 'telegram', got %q", job.DeliverChannel)
|
||||
}
|
||||
|
||||
@@ -28,7 +28,9 @@ func (s *SQLiteCronStore) AddJob(ctx context.Context, name string, schedule stor
|
||||
}
|
||||
|
||||
payload := store.CronPayload{
|
||||
Kind: "agent_turn", Message: message,
|
||||
Kind: "agent_turn",
|
||||
Message: message,
|
||||
CredentialUserID: store.ExplicitCredentialUserIDFromContext(ctx),
|
||||
}
|
||||
payloadJSON, _ := json.Marshal(payload)
|
||||
|
||||
@@ -395,7 +397,6 @@ func (s *SQLiteCronStore) lockCronJobForMutation(ctx context.Context, tx *sql.Tx
|
||||
return &state, nil
|
||||
}
|
||||
|
||||
|
||||
func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates map[string]any) error {
|
||||
if len(updates) == 0 {
|
||||
return nil
|
||||
@@ -433,4 +434,3 @@ func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -65,6 +65,29 @@ func TestSQLiteCronStore_ReenableRestoresNextRun(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSQLiteCronStore_AddJobPersistsCredentialUserID(t *testing.T) {
|
||||
cronStore, ctx, _ := newTestSQLiteCronStore(t)
|
||||
ctx = store.WithCredentialUserID(ctx, "tenant-user-123")
|
||||
everyMS := int64(time.Minute / time.Millisecond)
|
||||
|
||||
job, err := cronStore.AddJob(ctx, "job-credential-context", store.CronSchedule{
|
||||
Kind: "every",
|
||||
EveryMS: &everyMS,
|
||||
}, "run credentialed report", false, "", "", "", "group:telegram:-100123")
|
||||
if err != nil {
|
||||
t.Fatalf("AddJob error: %v", err)
|
||||
}
|
||||
if job == nil {
|
||||
job = mustOnlyJob(t, cronStore, ctx)
|
||||
}
|
||||
if job.UserID != "group:telegram:-100123" {
|
||||
t.Fatalf("cron user ID = %q, want group-scoped owner", job.UserID)
|
||||
}
|
||||
if job.Payload.CredentialUserID != "tenant-user-123" {
|
||||
t.Fatalf("credential user ID = %q, want tenant-user-123", job.Payload.CredentialUserID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSQLiteCronStore_EnableAlreadyEnabledPreservesNextRun(t *testing.T) {
|
||||
cronStore, ctx, _ := newTestSQLiteCronStore(t)
|
||||
everyMS := int64(time.Hour / time.Millisecond)
|
||||
|
||||
+15
-5
@@ -185,6 +185,7 @@ func (t *CronTool) handleStatus() *Result {
|
||||
func (t *CronTool) handleList(ctx context.Context, args map[string]any, agentID, userID string) *Result {
|
||||
includeDisabled, _ := args["includeDisabled"].(bool)
|
||||
jobs := t.cronStore.ListJobs(ctx, includeDisabled, agentID, userID)
|
||||
jobs = store.RedactCronJobsCredentialContext(jobs)
|
||||
|
||||
result := map[string]any{
|
||||
"jobs": jobs,
|
||||
@@ -305,7 +306,7 @@ func (t *CronTool) handleAdd(ctx context.Context, args map[string]any, agentID,
|
||||
}
|
||||
}
|
||||
|
||||
data, _ := json.MarshalIndent(map[string]any{"job": job}, "", " ")
|
||||
data, _ := json.MarshalIndent(map[string]any{"job": store.RedactCronJobCredentialContext(*job)}, "", " ")
|
||||
return NewResult(string(data))
|
||||
}
|
||||
|
||||
@@ -334,9 +335,13 @@ func (t *CronTool) handleUpdate(ctx context.Context, args map[string]any, agentI
|
||||
return ErrorResult("jobId is required for update action")
|
||||
}
|
||||
|
||||
if _, errResult := t.checkJobOwnership(ctx, jobID, agentID, userID); errResult != nil {
|
||||
existing, errResult := t.checkJobOwnership(ctx, jobID, agentID, userID)
|
||||
if errResult != nil {
|
||||
return errResult
|
||||
}
|
||||
if err := store.CheckCronCredentialOwner(ctx, existing); err != nil {
|
||||
return ErrorResult("permission denied: cron job uses a credential context owned by another user")
|
||||
}
|
||||
|
||||
patchObj, ok := args["patch"].(map[string]any)
|
||||
if !ok {
|
||||
@@ -360,7 +365,7 @@ func (t *CronTool) handleUpdate(ctx context.Context, args map[string]any, agentI
|
||||
return ErrorResult(fmt.Sprintf("failed to update cron job: %v", err))
|
||||
}
|
||||
|
||||
data, _ := json.MarshalIndent(map[string]any{"job": job}, "", " ")
|
||||
data, _ := json.MarshalIndent(map[string]any{"job": store.RedactCronJobCredentialContext(*job)}, "", " ")
|
||||
return NewResult(string(data))
|
||||
}
|
||||
|
||||
@@ -370,7 +375,8 @@ func (t *CronTool) handleRemove(ctx context.Context, args map[string]any, agentI
|
||||
return ErrorResult("jobId is required for remove action")
|
||||
}
|
||||
|
||||
if _, errResult := t.checkJobOwnership(ctx, jobID, agentID, userID); errResult != nil {
|
||||
_, errResult := t.checkJobOwnership(ctx, jobID, agentID, userID)
|
||||
if errResult != nil {
|
||||
return errResult
|
||||
}
|
||||
|
||||
@@ -388,9 +394,13 @@ func (t *CronTool) handleRun(ctx context.Context, args map[string]any, agentID,
|
||||
return ErrorResult("jobId is required for run action")
|
||||
}
|
||||
|
||||
if _, errResult := t.checkJobOwnership(ctx, jobID, agentID, userID); errResult != nil {
|
||||
existing, errResult := t.checkJobOwnership(ctx, jobID, agentID, userID)
|
||||
if errResult != nil {
|
||||
return errResult
|
||||
}
|
||||
if err := store.CheckCronCredentialOwner(ctx, existing); err != nil {
|
||||
return ErrorResult("permission denied: cron job uses a credential context owned by another user")
|
||||
}
|
||||
|
||||
runMode, _ := args["runMode"].(string)
|
||||
force := runMode == "force"
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
package tools
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
type testCronStore struct {
|
||||
jobs map[string]*store.CronJob
|
||||
updateCnt int
|
||||
runCnt int
|
||||
lastForce bool
|
||||
updateErr error
|
||||
runErr error
|
||||
}
|
||||
|
||||
func newTestCronStore(job *store.CronJob) *testCronStore {
|
||||
jobs := map[string]*store.CronJob{}
|
||||
if job != nil {
|
||||
jobs[job.ID] = job
|
||||
}
|
||||
return &testCronStore{jobs: jobs}
|
||||
}
|
||||
|
||||
func (s *testCronStore) AddJob(context.Context, string, store.CronSchedule, string, bool, string, string, string, string) (*store.CronJob, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (s *testCronStore) GetJob(_ context.Context, jobID string) (*store.CronJob, bool) {
|
||||
job, ok := s.jobs[jobID]
|
||||
return job, ok
|
||||
}
|
||||
|
||||
func (s *testCronStore) ListJobs(context.Context, bool, string, string) []store.CronJob {
|
||||
var jobs []store.CronJob
|
||||
for _, job := range s.jobs {
|
||||
jobs = append(jobs, *job)
|
||||
}
|
||||
return jobs
|
||||
}
|
||||
|
||||
func (s *testCronStore) RemoveJob(context.Context, string) error { return nil }
|
||||
|
||||
func (s *testCronStore) UpdateJob(_ context.Context, jobID string, patch store.CronJobPatch) (*store.CronJob, error) {
|
||||
s.updateCnt++
|
||||
if s.updateErr != nil {
|
||||
return nil, s.updateErr
|
||||
}
|
||||
job := s.jobs[jobID]
|
||||
if patch.Message != "" {
|
||||
job.Payload.Message = patch.Message
|
||||
}
|
||||
return job, nil
|
||||
}
|
||||
|
||||
func (s *testCronStore) EnableJob(context.Context, string, bool) error { return nil }
|
||||
func (s *testCronStore) GetRunLog(context.Context, string, int, int) ([]store.CronRunLogEntry, int) {
|
||||
return nil, 0
|
||||
}
|
||||
func (s *testCronStore) Status() map[string]any { return map[string]any{} }
|
||||
func (s *testCronStore) Start() error { return nil }
|
||||
func (s *testCronStore) Stop() {}
|
||||
func (s *testCronStore) SetOnJob(func(*store.CronJob) (*store.CronJobResult, error)) {}
|
||||
func (s *testCronStore) SetOnEvent(func(store.CronEvent)) {}
|
||||
|
||||
func (s *testCronStore) RunJob(_ context.Context, _ string, force bool) (bool, string, error) {
|
||||
s.runCnt++
|
||||
s.lastForce = force
|
||||
return true, "", s.runErr
|
||||
}
|
||||
|
||||
func (s *testCronStore) GetDueJobs(time.Time) []store.CronJob { return nil }
|
||||
func (s *testCronStore) SetDefaultTimezone(string) {}
|
||||
|
||||
func TestCronToolBlocksCredentialBoundUpdateByDifferentUser(t *testing.T) {
|
||||
cronStore := newTestCronStore(&store.CronJob{
|
||||
ID: "job-1",
|
||||
UserID: "group:telegram:-100123",
|
||||
Payload: store.CronPayload{
|
||||
Message: "old",
|
||||
CredentialUserID: "tenant-user-a",
|
||||
},
|
||||
})
|
||||
tool := NewCronTool(cronStore)
|
||||
ctx := store.WithUserID(context.Background(), "group:telegram:-100123")
|
||||
ctx = store.WithCredentialUserID(ctx, "tenant-user-b")
|
||||
|
||||
result := tool.Execute(ctx, map[string]any{
|
||||
"action": "update",
|
||||
"jobId": "job-1",
|
||||
"patch": map[string]any{"message": "run gh issue list"},
|
||||
})
|
||||
|
||||
if !result.IsError || !strings.Contains(result.ForLLM, "credential context") {
|
||||
t.Fatalf("expected credential context error, got %#v", result)
|
||||
}
|
||||
if cronStore.updateCnt != 0 {
|
||||
t.Fatalf("UpdateJob called %d times, want 0", cronStore.updateCnt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCronToolListRedactsCredentialUserID(t *testing.T) {
|
||||
cronStore := newTestCronStore(&store.CronJob{
|
||||
ID: "job-1",
|
||||
UserID: "group:telegram:-100123",
|
||||
Payload: store.CronPayload{
|
||||
Message: "run gh issue list",
|
||||
CredentialUserID: "tenant-user-a",
|
||||
},
|
||||
})
|
||||
tool := NewCronTool(cronStore)
|
||||
ctx := store.WithUserID(context.Background(), "group:telegram:-100123")
|
||||
|
||||
result := tool.Execute(ctx, map[string]any{"action": "list", "includeDisabled": true})
|
||||
|
||||
if result.IsError {
|
||||
t.Fatalf("list returned error: %#v", result)
|
||||
}
|
||||
if strings.Contains(result.ForLLM, "tenant-user-a") || strings.Contains(result.ForLLM, "credentialUserId") {
|
||||
t.Fatalf("credential identity leaked in list response: %s", result.ForLLM)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCronToolBlocksCredentialBoundRunByDifferentUser(t *testing.T) {
|
||||
cronStore := newTestCronStore(&store.CronJob{
|
||||
ID: "job-1",
|
||||
UserID: "group:telegram:-100123",
|
||||
Payload: store.CronPayload{
|
||||
Message: "run gh issue list",
|
||||
CredentialUserID: "tenant-user-a",
|
||||
},
|
||||
})
|
||||
tool := NewCronTool(cronStore)
|
||||
ctx := store.WithUserID(context.Background(), "group:telegram:-100123")
|
||||
ctx = store.WithCredentialUserID(ctx, "tenant-user-b")
|
||||
|
||||
result := tool.Execute(ctx, map[string]any{
|
||||
"action": "run",
|
||||
"jobId": "job-1",
|
||||
"runMode": "force",
|
||||
})
|
||||
|
||||
if !result.IsError || !strings.Contains(result.ForLLM, "credential context") {
|
||||
t.Fatalf("expected credential context error, got %#v", result)
|
||||
}
|
||||
if cronStore.runCnt != 0 {
|
||||
t.Fatalf("RunJob called %d times, want 0", cronStore.runCnt)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user