|
| 1 | +package lifecycle |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "encoding/json" |
| 6 | + "errors" |
| 7 | + "testing" |
| 8 | + |
| 9 | + "github.com/kandev/kandev/internal/auth/authn" |
| 10 | + "github.com/kandev/kandev/internal/common/logger" |
| 11 | + ws "github.com/kandev/kandev/pkg/websocket" |
| 12 | +) |
| 13 | + |
| 14 | +// recordingMCPHandler captures the context its dispatch ran under so tests can |
| 15 | +// assert which identity (if any) reached the tool handlers. |
| 16 | +type recordingMCPHandler struct { |
| 17 | + gotCtx context.Context |
| 18 | + calls int |
| 19 | +} |
| 20 | + |
| 21 | +func (h *recordingMCPHandler) Dispatch(ctx context.Context, msg *ws.Message) (*ws.Message, error) { |
| 22 | + h.gotCtx = ctx |
| 23 | + h.calls++ |
| 24 | + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{"ok": true}) |
| 25 | +} |
| 26 | + |
| 27 | +func newMCPStreamManager(t *testing.T, inner *recordingMCPHandler, scoper MCPIdentityScoper) *StreamManager { |
| 28 | + t.Helper() |
| 29 | + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) |
| 30 | + if err != nil { |
| 31 | + t.Fatalf("logger: %v", err) |
| 32 | + } |
| 33 | + sm := NewStreamManager(log, StreamCallbacks{}, inner, nil) |
| 34 | + sm.mcpIdentityScoper = scoper |
| 35 | + return sm |
| 36 | +} |
| 37 | + |
| 38 | +func mcpRequest(t *testing.T, payload map[string]interface{}) *ws.Message { |
| 39 | + t.Helper() |
| 40 | + data, err := json.Marshal(payload) |
| 41 | + if err != nil { |
| 42 | + t.Fatalf("marshal payload: %v", err) |
| 43 | + } |
| 44 | + return &ws.Message{ID: "req-1", Type: ws.MessageTypeRequest, Action: "mcp.list_tasks", Payload: data} |
| 45 | +} |
| 46 | + |
| 47 | +// TestMCPHandlerForScopesToExecutionTask is the core wiring assertion: the |
| 48 | +// identity handed to the tool handlers comes from the execution that owns the |
| 49 | +// stream. |
| 50 | +func TestMCPHandlerForScopesToExecutionTask(t *testing.T) { |
| 51 | + inner := &recordingMCPHandler{} |
| 52 | + var scopedTaskIDs []string |
| 53 | + sm := newMCPStreamManager(t, inner, func(ctx context.Context, taskID string) (context.Context, error) { |
| 54 | + scopedTaskIDs = append(scopedTaskIDs, taskID) |
| 55 | + return authn.WithIdentity(ctx, authn.Identity{UserID: "owner-of-" + taskID, Role: authn.RoleMember}), nil |
| 56 | + }) |
| 57 | + |
| 58 | + handler := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1", TaskID: "task-a"}) |
| 59 | + resp, err := handler.Dispatch(context.Background(), mcpRequest(t, map[string]interface{}{})) |
| 60 | + if err != nil { |
| 61 | + t.Fatalf("Dispatch: %v", err) |
| 62 | + } |
| 63 | + if resp.Type != ws.MessageTypeResponse { |
| 64 | + t.Fatalf("response type = %q, want response", resp.Type) |
| 65 | + } |
| 66 | + |
| 67 | + if len(scopedTaskIDs) != 1 || scopedTaskIDs[0] != "task-a" { |
| 68 | + t.Fatalf("scoped task IDs = %v, want [task-a]", scopedTaskIDs) |
| 69 | + } |
| 70 | + identity, ok := authn.IdentityFromContext(inner.gotCtx) |
| 71 | + if !ok { |
| 72 | + t.Fatal("tool handlers received no identity") |
| 73 | + } |
| 74 | + if identity.UserID != "owner-of-task-a" { |
| 75 | + t.Errorf("UserID = %q, want owner-of-task-a", identity.UserID) |
| 76 | + } |
| 77 | +} |
| 78 | + |
| 79 | +// TestMCPHandlerForIgnoresPayloadSessionID is the privilege-escalation pin. The |
| 80 | +// payload is agent-controlled, so scoping must key off the execution's task |
| 81 | +// even when the request names a different session or task. |
| 82 | +func TestMCPHandlerForIgnoresPayloadSessionID(t *testing.T) { |
| 83 | + inner := &recordingMCPHandler{} |
| 84 | + var scopedTaskIDs []string |
| 85 | + sm := newMCPStreamManager(t, inner, func(ctx context.Context, taskID string) (context.Context, error) { |
| 86 | + scopedTaskIDs = append(scopedTaskIDs, taskID) |
| 87 | + return ctx, nil |
| 88 | + }) |
| 89 | + |
| 90 | + handler := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1", TaskID: "task-a"}) |
| 91 | + _, err := handler.Dispatch(context.Background(), mcpRequest(t, map[string]interface{}{ |
| 92 | + "session_id": "session-of-victim", |
| 93 | + "task_id": "task-victim", |
| 94 | + })) |
| 95 | + if err != nil { |
| 96 | + t.Fatalf("Dispatch: %v", err) |
| 97 | + } |
| 98 | + |
| 99 | + if len(scopedTaskIDs) != 1 || scopedTaskIDs[0] != "task-a" { |
| 100 | + t.Errorf("scoped task IDs = %v, want [task-a] — payload IDs must not steer scoping", scopedTaskIDs) |
| 101 | + } |
| 102 | +} |
| 103 | + |
| 104 | +// TestMCPHandlerForDeniesWhenScopingFails pins fail-closed behavior: an |
| 105 | +// unresolvable owner must not fall through to the unscoped handlers. |
| 106 | +func TestMCPHandlerForDeniesWhenScopingFails(t *testing.T) { |
| 107 | + inner := &recordingMCPHandler{} |
| 108 | + sm := newMCPStreamManager(t, inner, func(context.Context, string) (context.Context, error) { |
| 109 | + return nil, errors.New("db unavailable") |
| 110 | + }) |
| 111 | + |
| 112 | + handler := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1", TaskID: "task-a"}) |
| 113 | + resp, err := handler.Dispatch(context.Background(), mcpRequest(t, map[string]interface{}{})) |
| 114 | + if err != nil { |
| 115 | + t.Fatalf("Dispatch: %v", err) |
| 116 | + } |
| 117 | + |
| 118 | + if resp.Type != ws.MessageTypeError { |
| 119 | + t.Errorf("response type = %q, want error", resp.Type) |
| 120 | + } |
| 121 | + if inner.calls != 0 { |
| 122 | + t.Errorf("inner handler ran %d times, want 0 — the request must be denied", inner.calls) |
| 123 | + } |
| 124 | +} |
| 125 | + |
| 126 | +// TestMCPHandlerForPassesThroughWithoutScoper keeps single-user instances and |
| 127 | +// isolated tests on the original unwrapped handler. |
| 128 | +func TestMCPHandlerForPassesThroughWithoutScoper(t *testing.T) { |
| 129 | + inner := &recordingMCPHandler{} |
| 130 | + sm := newMCPStreamManager(t, inner, nil) |
| 131 | + |
| 132 | + if got := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1", TaskID: "task-a"}); got != inner { |
| 133 | + t.Errorf("handler = %T, want the unwrapped inner handler", got) |
| 134 | + } |
| 135 | +} |
| 136 | + |
| 137 | +// TestMCPHandlerForPassesThroughWithoutTaskID covers executions with no task |
| 138 | +// (there is no owner to resolve, so wrapping would deny every call). |
| 139 | +func TestMCPHandlerForPassesThroughWithoutTaskID(t *testing.T) { |
| 140 | + inner := &recordingMCPHandler{} |
| 141 | + sm := newMCPStreamManager(t, inner, func(ctx context.Context, _ string) (context.Context, error) { |
| 142 | + return ctx, nil |
| 143 | + }) |
| 144 | + |
| 145 | + if got := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1"}); got != inner { |
| 146 | + t.Errorf("handler = %T, want the unwrapped inner handler", got) |
| 147 | + } |
| 148 | +} |
| 149 | + |
| 150 | +func TestSetMCPIdentityScoperReachesStreamManager(t *testing.T) { |
| 151 | + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) |
| 152 | + if err != nil { |
| 153 | + t.Fatalf("logger: %v", err) |
| 154 | + } |
| 155 | + m := &Manager{streamManager: NewStreamManager(log, StreamCallbacks{}, &recordingMCPHandler{}, nil)} |
| 156 | + |
| 157 | + m.SetMCPIdentityScoper(func(ctx context.Context, _ string) (context.Context, error) { return ctx, nil }) |
| 158 | + |
| 159 | + if m.streamManager.mcpIdentityScoper == nil { |
| 160 | + t.Error("SetMCPIdentityScoper did not reach the stream manager") |
| 161 | + } |
| 162 | +} |
0 commit comments