From 4a43a1d620fc6508635f08671d294cef0fd42aea Mon Sep 17 00:00:00 2001 From: Hz_ Date: Fri, 31 Jul 2026 16:24:25 +0800 Subject: [PATCH] fix(go-agent): align logs auth with Python and harden event data (#17617) ## Summary - Align Go agent logs access with the Python behavior by checking tenant access to the target agent instead of requiring the token to be bound to that agent. - Marshal canvas lifecycle event payloads safely so non-serializable inputs do not break Thinking expansion in the UI. ## Testing - `CGO_ENABLED=0 go test -count=1 -v -run "Handler|Attachment|Logs|Chatbot|Agentbot|BuildWorkflow|NodeLifecycle|VarRef|Resolve|Extract"./internal/handler/ ./internal/agent/canvas/...` PASS --- internal/agent/canvas/scheduler.go | 23 +++- internal/agent/canvas/scheduler_test.go | 87 +++++++++++++++ internal/handler/bot.go | 52 ++++----- internal/handler/bot_test.go | 137 ++++++++++++++++++------ internal/service/bot.go | 23 ++++ 5 files changed, 252 insertions(+), 70 deletions(-) diff --git a/internal/agent/canvas/scheduler.go b/internal/agent/canvas/scheduler.go index 452d1dc9b5..ad07c0857a 100644 --- a/internal/agent/canvas/scheduler.go +++ b/internal/agent/canvas/scheduler.go @@ -20,7 +20,6 @@ package canvas import ( "context" - "encoding/json" "fmt" "strings" "time" @@ -285,7 +284,7 @@ func nodeStartedAt(ctx context.Context, state *CanvasState, cpnID, componentName state.Sys["_node_start_"+cpnID] = now state.Sys["_node_inputs_"+cpnID] = sanitizeNodeInputs(inputs) } - nsData, _ := json.Marshal(NodeStartedData{ + nsData, err := runtime.SafeJSONMarshal(NodeStartedData{ Inputs: sanitizeNodeInputs(inputs), CreatedAt: now, ComponentID: cpnID, @@ -293,6 +292,15 @@ func nodeStartedAt(ctx context.Context, state *CanvasState, cpnID, componentName ComponentType: componentType, Thoughts: "", }) + if err != nil { + common.Warn("node_started marshal failed", + zap.String("cpnID", cpnID), + zap.String("componentName", componentName), + zap.Error(err), + ) + nsData = []byte(fmt.Sprintf(`{"component_id":%q,"component_name":%q,"component_type":%q}`, + cpnID, componentName, componentType)) + } meta := GetRunMeta(ctx) msgID, sessionID := "", "" if meta != nil { @@ -346,7 +354,7 @@ func nodeFinishedNow(ctx context.Context, state *CanvasState, cpnID, componentNa nfErr = nodeErr.Error() } - nfData, _ := json.Marshal(NodeFinishedData{ + nfData, err := runtime.SafeJSONMarshal(NodeFinishedData{ Inputs: inputs, Outputs: outputs, ComponentID: cpnID, @@ -356,6 +364,15 @@ func nodeFinishedNow(ctx context.Context, state *CanvasState, cpnID, componentNa ElapsedTime: elapsed, CreatedAt: now, }) + if err != nil { + common.Warn("node_finished marshal failed", + zap.String("cpnID", cpnID), + zap.String("componentName", componentName), + zap.Error(err), + ) + nfData = []byte(fmt.Sprintf(`{"component_id":%q,"component_name":%q,"component_type":%q}`, + cpnID, componentName, componentType)) + } meta := GetRunMeta(ctx) msgID, sessionID := "", "" if meta != nil { diff --git a/internal/agent/canvas/scheduler_test.go b/internal/agent/canvas/scheduler_test.go index 135a32ef07..fe4d86f625 100644 --- a/internal/agent/canvas/scheduler_test.go +++ b/internal/agent/canvas/scheduler_test.go @@ -3,6 +3,7 @@ package canvas import ( "context" + "encoding/json" "strings" "testing" ) @@ -245,3 +246,89 @@ func TestBuildWorkflow_ErrorsOnSelfEdge(t *testing.T) { t.Fatalf("expected 'self-edge' in error, got: %v", err) } } + +func TestNodeLifecycleEventsSafeMarshalNonSerializableInputs(t *testing.T) { + events := make(chan RunEvent, 2) + ctx := WithRunMeta(context.Background(), &RunMeta{ + Events: events, + MessageID: "msg-1", + SessionID: "session-1", + }) + state := NewCanvasState("run-1", "session-1") + inputs := map[string]any{ + "query": "hi", + "callback": func() {}, + } + + nodeStartedAt(ctx, state, "agent_0", "Agent", "Agent", inputs) + nodeFinishedNow(ctx, state, "agent_0", "Agent", "Agent", nil) + + for i, wantType := range []string{"node_started", "node_finished"} { + select { + case ev := <-events: + if ev.Type != wantType { + t.Fatalf("event[%d].Type = %q, want %q", i, ev.Type, wantType) + } + var payload map[string]any + if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil { + t.Fatalf("%s Data should be valid JSON, got %q: %v", wantType, ev.Data, err) + } + if payload["component_type"] != "Agent" { + t.Fatalf("%s component_type = %v, want Agent; payload=%v", wantType, payload["component_type"], payload) + } + in, ok := payload["inputs"].(map[string]any) + if !ok { + t.Fatalf("%s inputs = %T %v, want object", wantType, payload["inputs"], payload["inputs"]) + } + if _, ok := in["callback"]; !ok { + t.Fatalf("%s inputs missing sanitized callback key: %v", wantType, in) + } + if in["callback"] != nil { + t.Fatalf("%s callback = %T %v, want nil after safe JSON cleanup", wantType, in["callback"], in["callback"]) + } + default: + t.Fatalf("missing %s event", wantType) + } + } +} + +func TestNodeLifecycleEventsFallbackMarshalErrorPreservesIdentity(t *testing.T) { + events := make(chan RunEvent, 2) + ctx := WithRunMeta(context.Background(), &RunMeta{ + Events: events, + MessageID: "msg-2", + SessionID: "session-2", + }) + state := NewCanvasState("run-2", "session-2") + inputs := map[string]any{ + "query": "hi", + "broken": complex(1, 2), + } + + nodeStartedAt(ctx, state, "agent_1", "Agent", "Agent", inputs) + nodeFinishedNow(ctx, state, "agent_1", "Agent", "Agent", nil) + + for i, wantType := range []string{"node_started", "node_finished"} { + select { + case ev := <-events: + if ev.Type != wantType { + t.Fatalf("event[%d].Type = %q, want %q", i, ev.Type, wantType) + } + var payload map[string]any + if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil { + t.Fatalf("%s Data should be valid JSON, got %q: %v", wantType, ev.Data, err) + } + if payload["component_id"] != "agent_1" { + t.Fatalf("%s component_id = %v, want agent_1; payload=%v", wantType, payload["component_id"], payload) + } + if payload["component_name"] != "Agent" { + t.Fatalf("%s component_name = %v, want Agent; payload=%v", wantType, payload["component_name"], payload) + } + if payload["component_type"] != "Agent" { + t.Fatalf("%s component_type = %v, want Agent; payload=%v", wantType, payload["component_type"], payload) + } + default: + t.Fatalf("missing %s event", wantType) + } + } +} diff --git a/internal/handler/bot.go b/internal/handler/bot.go index cd61f07605..51c4236cdf 100644 --- a/internal/handler/bot.go +++ b/internal/handler/bot.go @@ -18,14 +18,11 @@ package handler import ( "context" - "encoding/json" - "fmt" "github.com/gin-gonic/gin" "ragflow/internal/agent/canvas" "ragflow/internal/common" - "ragflow/internal/engine/redis" "ragflow/internal/service" ) @@ -48,6 +45,7 @@ type botService interface { ec common.ErrorCode, err error) AgentbotCompletion(ctx context.Context, tenantID, agentID string, req service.AgentbotCompletionRequest) ( <-chan canvas.RunEvent, common.ErrorCode, error) + AgentbotLogs(ctx context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) ChatbotCompletion(ctx context.Context, tenantID, dialogID string, req service.ChatbotCompletionRequest) ( <-chan service.ChatbotSSEFrame, common.ErrorCode, error) } @@ -230,13 +228,12 @@ func (h *BotHandler) ChatbotCompletion(c *gin.Context) { // GetAgentbotLogs GET /api/v1/agentbots//logs/ // -// Beta-token sibling of GetAgentLogs. The shared/embedded chat -// page's "Thinking" button hits this endpoint because the share -// flow authenticates with a beta APIToken (no session JWT) and -// the regular /api/v1/agents//logs/ requires @login_required. -// Mirrors python bot_api.py:agent_bot_logs. +// Beta-token sibling of GetAgentLogs. The shared/embedded chat page's +// "Thinking" button hits this endpoint because the share flow authenticates +// with a beta APIToken (no session JWT). Mirrors python bot_api.py:agent_bot_logs. func (h *BotHandler) GetAgentbotLogs(c *gin.Context) { - if _, code, msg := GetUser(c); code != common.CodeSuccess { + user, code, msg := GetUser(c) + if code != common.CodeSuccess { common.ResponseWithCodeData(c, code, nil, msg) return } @@ -245,38 +242,25 @@ func (h *BotHandler) GetAgentbotLogs(c *gin.Context) { common.ResponseWithCodeData(c, common.CodeArgumentError, nil, "agent_id is required") return } - boundAgentID, _ := c.Get("agent_id") - boundAgentIDStr, _ := boundAgentID.(string) - if boundAgentIDStr == "" { - common.ResponseWithCodeData(c, common.CodeDataError, nil, "API token is not bound to an agent.") - return - } - if boundAgentIDStr != agentIDStr { - common.ResponseWithCodeData(c, common.CodeUnauthorized, nil, "API token is not authorized for this agent.") - return - } messageID := c.Param("message_id") if messageID == "" { common.ResponseWithCodeData(c, common.CodeArgumentError, nil, "message_id is required") return } - key := fmt.Sprintf("%s-%s-logs", agentIDStr, messageID) - payload, rerr := redis.Get().Get(key) - // Surface Redis / decode failures instead of silently returning - // `{code: 0, data: {}}` — the previous form made the endpoint - // indistinguishable from "logs not yet written", which masked - // real outages and corrupted payloads from operators (PR review - // round 5, Major #6). - if rerr != nil { - common.ResponseWithCodeData(c, common.CodeServerError, nil, "failed to read agent logs") - return - } - data := map[string]interface{}{} - if payload != "" { - if uerr := json.Unmarshal([]byte(payload), &data); uerr != nil { - common.ResponseWithCodeData(c, common.CodeServerError, nil, "failed to decode agent logs") + // A beta token with DialogID is restricted to that agent. Tokens without + // DialogID remain tenant-scoped and are checked by AgentbotLogs through + // the existing Canvas access guard. + if boundAgentID, ok := c.Get("agent_id"); ok { + boundAgentIDStr, ok := boundAgentID.(string) + if ok && boundAgentIDStr != "" && boundAgentIDStr != agentIDStr { + common.ResponseWithCodeData(c, common.CodeUnauthorized, nil, "API token is not authorized for this agent.") return } } + data, ec, err := h.botService.AgentbotLogs(c.Request.Context(), user.ID, agentIDStr, messageID) + if err != nil { + common.ResponseWithCodeData(c, ec, nil, err.Error()) + return + } common.SuccessWithData(c, data, "success") } diff --git a/internal/handler/bot_test.go b/internal/handler/bot_test.go index b702d27c74..2e98453141 100644 --- a/internal/handler/bot_test.go +++ b/internal/handler/bot_test.go @@ -42,6 +42,7 @@ type stubBotService struct { chatbotInfoFn func(ctx context.Context, tenantID, dialogID string) (string, string, string, string, bool, common.ErrorCode, error) agentbotInputsFn func(ctx context.Context, tenantID, agentID string) (string, string, string, string, map[string]any, common.ErrorCode, error) agentbotCompleteFn func(ctx context.Context, tenantID, agentID string, req service.AgentbotCompletionRequest) (<-chan canvas.RunEvent, common.ErrorCode, error) + agentbotLogsFn func(ctx context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) chatbotCompleteFn func(ctx context.Context, tenantID, dialogID string, req service.ChatbotCompletionRequest) (<-chan service.ChatbotSSEFrame, common.ErrorCode, error) } @@ -66,6 +67,13 @@ func (s *stubBotService) AgentbotCompletion(ctx context.Context, tenantID, agent return nil, common.CodeDataError, errors.New("not stubbed") } +func (s *stubBotService) AgentbotLogs(ctx context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) { + if s.agentbotLogsFn != nil { + return s.agentbotLogsFn(ctx, tenantID, agentID, messageID) + } + return nil, common.CodeDataError, errors.New("not stubbed") +} + func (s *stubBotService) ChatbotCompletion(ctx context.Context, tenantID, dialogID string, req service.ChatbotCompletionRequest) (<-chan service.ChatbotSSEFrame, common.ErrorCode, error) { if s.chatbotCompleteFn != nil { return s.chatbotCompleteFn(ctx, tenantID, dialogID, req) @@ -1209,7 +1217,7 @@ func TestGetAgentbotLogs_MissingRouteAgentID(t *testing.T) { } } -func TestGetAgentbotLogs_RequiresBoundAgentID(t *testing.T) { +func TestGetAgentbotLogs_AllowsUnboundTokenForAccessibleAgent(t *testing.T) { gin.SetMode(gin.TestMode) w := httptest.NewRecorder() c, _ := gin.CreateTestContext(w) @@ -1222,6 +1230,99 @@ func TestGetAgentbotLogs_RequiresBoundAgentID(t *testing.T) { } h := NewBotHandler(nil) + h.botService = &stubBotService{agentbotLogsFn: func(_ context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) { + if tenantID != "u1" || agentID != "agent-a" || messageID != "msg-1" { + t.Fatalf("logs arguments = (%q, %q, %q)", tenantID, agentID, messageID) + } + return map[string]any{"events": []any{}}, common.CodeSuccess, nil + }} + h.GetAgentbotLogs(c) + + var resp struct { + Code int `json:"code"` + Message string `json:"message"` + } + _ = json.Unmarshal(w.Body.Bytes(), &resp) + if resp.Code != int(common.CodeSuccess) { + t.Errorf("code = %d, want %d; message=%q", resp.Code, common.CodeSuccess, resp.Message) + } +} + +func TestGetAgentbotLogs_AllowsMatchingBoundAgent(t *testing.T) { + gin.SetMode(gin.TestMode) + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest("GET", "/api/v1/agentbots/agent-a/logs/msg-1", nil) + c.Set("user", &entity.User{ID: "u1"}) + c.Set("agent_id", "agent-a") + c.Params = gin.Params{ + {Key: "agent_id", Value: "agent-a"}, + {Key: "message_id", Value: "msg-1"}, + } + + h := NewBotHandler(nil) + h.botService = &stubBotService{agentbotLogsFn: func(_ context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) { + return map[string]any{"events": []any{}}, common.CodeSuccess, nil + }} + h.GetAgentbotLogs(c) + + var resp struct { + Code int `json:"code"` + } + _ = json.Unmarshal(w.Body.Bytes(), &resp) + if resp.Code != int(common.CodeSuccess) { + t.Fatalf("code = %d, want %d; body = %s", resp.Code, common.CodeSuccess, w.Body.String()) + } +} + +func TestGetAgentbotLogs_DeniesMismatchedBoundAgent(t *testing.T) { + gin.SetMode(gin.TestMode) + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest("GET", "/api/v1/agentbots/agent-b/logs/msg-1", nil) + c.Set("user", &entity.User{ID: "u1"}) + c.Set("agent_id", "agent-a") + c.Params = gin.Params{ + {Key: "agent_id", Value: "agent-b"}, + {Key: "message_id", Value: "msg-1"}, + } + + h := NewBotHandler(nil) + h.botService = &stubBotService{agentbotLogsFn: func(context.Context, string, string, string) (map[string]any, common.ErrorCode, error) { + t.Fatal("AgentbotLogs should not be called for a mismatched bound agent") + return nil, common.CodeServerError, errors.New("unexpected call") + }} + h.GetAgentbotLogs(c) + + var resp struct { + Code int `json:"code"` + Message string `json:"message"` + } + _ = json.Unmarshal(w.Body.Bytes(), &resp) + if resp.Code != int(common.CodeUnauthorized) { + t.Fatalf("code = %d, want %d; body = %s", resp.Code, common.CodeUnauthorized, w.Body.String()) + } + if !strings.Contains(resp.Message, "not authorized") { + t.Fatalf("message = %q, want bound-agent authorization error", resp.Message) + } +} + +func TestGetAgentbotLogs_DeniesInaccessibleAgent(t *testing.T) { + gin.SetMode(gin.TestMode) + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest("GET", + "/api/v1/agentbots/agent-b/logs/msg-1", nil) + c.Set("user", &entity.User{ID: "u1"}) + c.Params = gin.Params{ + {Key: "agent_id", Value: "agent-b"}, + {Key: "message_id", Value: "msg-1"}, + } + + h := NewBotHandler(nil) + h.botService = &stubBotService{agentbotLogsFn: func(context.Context, string, string, string) (map[string]any, common.ErrorCode, error) { + return nil, common.CodeDataError, errors.New("Can't find agent by ID: agent-b") + }} h.GetAgentbotLogs(c) var resp struct { @@ -1232,37 +1333,8 @@ func TestGetAgentbotLogs_RequiresBoundAgentID(t *testing.T) { if resp.Code != int(common.CodeDataError) { t.Errorf("code = %d, want %d", resp.Code, common.CodeDataError) } - if !strings.Contains(resp.Message, "not bound") { - t.Errorf("message = %q, want it to mention 'not bound'", resp.Message) - } -} - -func TestGetAgentbotLogs_CrossAgentDenied(t *testing.T) { - gin.SetMode(gin.TestMode) - w := httptest.NewRecorder() - c, _ := gin.CreateTestContext(w) - c.Request = httptest.NewRequest("GET", - "/api/v1/agentbots/agent-b/logs/msg-1", nil) - c.Set("user", &entity.User{ID: "u1"}) - c.Set("agent_id", "agent-a") - c.Params = gin.Params{ - {Key: "agent_id", Value: "agent-b"}, - {Key: "message_id", Value: "msg-1"}, - } - - h := NewBotHandler(nil) - h.GetAgentbotLogs(c) - - var resp struct { - Code int `json:"code"` - Message string `json:"message"` - } - _ = json.Unmarshal(w.Body.Bytes(), &resp) - if resp.Code != int(common.CodeUnauthorized) { - t.Errorf("code = %d, want %d", resp.Code, common.CodeUnauthorized) - } - if !strings.Contains(resp.Message, "not authorized") { - t.Errorf("message = %q, want it to mention 'not authorized'", resp.Message) + if !strings.Contains(resp.Message, "Can't find agent") { + t.Errorf("message = %q, want an access denial", resp.Message) } } @@ -1275,7 +1347,6 @@ func TestGetAgentbotLogs_MissingMessageID(t *testing.T) { c.Request = httptest.NewRequest("GET", "/api/v1/agentbots/shared-x/logs/", nil) c.Set("user", &entity.User{ID: "u1"}) - c.Set("agent_id", "agent-real") c.Params = gin.Params{{Key: "agent_id", Value: "agent-real"}} // Gin's path param extraction returns "" for a missing // segment so the handler must reject with CodeArgumentError. diff --git a/internal/service/bot.go b/internal/service/bot.go index 13b96a6952..8213353263 100644 --- a/internal/service/bot.go +++ b/internal/service/bot.go @@ -24,6 +24,7 @@ package service import ( "context" + "encoding/json" "errors" "fmt" "hash/fnv" @@ -34,6 +35,7 @@ import ( "ragflow/internal/agent/dsl" "ragflow/internal/common" "ragflow/internal/dao" + "ragflow/internal/engine/redis" "ragflow/internal/entity" ) @@ -178,6 +180,27 @@ func (s *BotService) AgentbotCompletion( return ch, common.CodeSuccess, nil } +// AgentbotLogs returns the stored execution timeline for an agentbot run. +// Access is scoped to the caller's accessible tenants, matching the Python +// agent_bot_logs endpoint. +func (s *BotService) AgentbotLogs(ctx context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) { + if _, err := s.loadCanvas(ctx, tenantID, agentID); err != nil { + return nil, common.CodeDataError, err + } + payload, err := redis.Get().Get(fmt.Sprintf("%s-%s-logs", agentID, messageID)) + if err != nil { + return nil, common.CodeServerError, errors.New("failed to read agent logs") + } + data := map[string]any{} + if payload == "" { + return data, common.CodeSuccess, nil + } + if err := json.Unmarshal([]byte(payload), &data); err != nil { + return nil, common.CodeServerError, errors.New("failed to decode agent logs") + } + return data, common.CodeSuccess, nil +} + // AgentbotCompletionRequest is the request body for // /api/v1/agentbots//completions. We intentionally accept // the same fields the production /agents/chat/completions handler