Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 335796b0b4 | |||
| 699b5fb275 | |||
| fb2fd20c9e | |||
| 7d2eaa3748 | |||
| 44b78e28c8 |
@@ -67,9 +67,21 @@ export function useChatSocket(
|
||||
const own = (targetId || msg.workspace_id) === workspaceId;
|
||||
if (own) {
|
||||
callbacksRef.current.onSendComplete?.();
|
||||
callbacksRef.current.onSendError?.(
|
||||
"Agent error (Exception) — see workspace logs for details.",
|
||||
);
|
||||
// internal#211/#212: surface the runtime's curated,
|
||||
// user-actionable reason (provider HTTP status + error
|
||||
// code + the provider's own guidance, e.g. a 403 "org
|
||||
// disabled · use an API key / ask your admin"). The
|
||||
// server now includes error_detail in the ACTIVITY_LOGGED
|
||||
// broadcast; fall back to summary, and only as a last
|
||||
// resort to a generic line. The old hardcoded
|
||||
// "Agent error (Exception) — see workspace logs for
|
||||
// details." string pointed at a logs UI that does not
|
||||
// exist and discarded the actionable reason entirely.
|
||||
const detail =
|
||||
(p.error_detail as string) ||
|
||||
(p.summary as string) ||
|
||||
"The agent turn failed but the runtime reported no detail. Retry once; if it repeats the workspace runtime may need a restart.";
|
||||
callbacksRef.current.onSendError?.(detail);
|
||||
}
|
||||
}
|
||||
} else if (type === "a2a_send") {
|
||||
|
||||
@@ -62,6 +62,7 @@ TOP_LEVEL_MODULES = {
|
||||
"a2a_tools_memory",
|
||||
"a2a_tools_messaging",
|
||||
"a2a_tools_rbac",
|
||||
"a2a_tools_identity",
|
||||
"adapter_base",
|
||||
"agent",
|
||||
"agents_md",
|
||||
|
||||
@@ -691,6 +691,19 @@ func logActivityExec(ctx context.Context, exec activityExecutor, broadcaster eve
|
||||
if respStr != nil {
|
||||
payload["response_body"] = json.RawMessage(respJSON)
|
||||
}
|
||||
// internal#211/#212: error_detail carries the runtime's curated,
|
||||
// user-actionable, secret-safe failure reason (provider HTTP
|
||||
// status + error code + the provider's own guidance, e.g. a 403
|
||||
// "org disabled · use an API key / ask your admin"). It is
|
||||
// already persisted to the DB column above and capped by the
|
||||
// runtime's report_activity helper (4096 chars). Previously it
|
||||
// was dropped from the LIVE broadcast, so the canvas had nothing
|
||||
// to render and fell back to a hardcoded opaque
|
||||
// "Agent error (Exception) — see workspace logs" string. Include
|
||||
// it so the chat bubble shows the real reason in real time.
|
||||
if params.ErrorDetail != nil && *params.ErrorDetail != "" {
|
||||
payload["error_detail"] = *params.ErrorDetail
|
||||
}
|
||||
}
|
||||
|
||||
return func() {
|
||||
|
||||
@@ -3,7 +3,7 @@ package handlers
|
||||
// workspace_broadcast.go — POST /workspaces/:id/broadcast
|
||||
//
|
||||
// Allows a workspace with broadcast_enabled=true to send a message to every
|
||||
// non-removed agent workspace in the SAME ORG. The message is:
|
||||
// non-removed agent workspace in the org. The message is:
|
||||
//
|
||||
// • Persisted in each recipient's activity_logs (type='broadcast_receive')
|
||||
// so poll-mode agents pick it up via GET /activity.
|
||||
@@ -16,11 +16,6 @@ package handlers
|
||||
// Auth: WorkspaceAuth (the agent triggers this with its own bearer token).
|
||||
// The handler re-validates broadcast_enabled inside the DB lookup to prevent
|
||||
// TOCTOU — the middleware only proved the token is valid, not the ability.
|
||||
//
|
||||
// Org isolation (OFFSEC-015): recipients are scoped to the sender's org using
|
||||
// a recursive CTE that walks the parent_id chain to find the org root. This
|
||||
// prevents a compromised or misconfigured workspace from broadcasting to
|
||||
// workspaces in other tenants' orgs.
|
||||
|
||||
import (
|
||||
"log"
|
||||
@@ -79,49 +74,11 @@ func (h *BroadcastHandler) Broadcast(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
// Find the sender's org root by walking the parent_id chain.
|
||||
// Workspaces with parent_id = NULL are org roots; every other workspace
|
||||
// belongs to the org identified by its topmost ancestor.
|
||||
var orgRootID string
|
||||
err = db.DB.QueryRowContext(ctx, `
|
||||
WITH RECURSIVE org_chain AS (
|
||||
SELECT id, parent_id, id AS root_id
|
||||
FROM workspaces
|
||||
WHERE id = $1
|
||||
UNION ALL
|
||||
SELECT w.id, w.parent_id, c.root_id
|
||||
FROM workspaces w
|
||||
JOIN org_chain c ON w.id = c.parent_id
|
||||
)
|
||||
SELECT root_id FROM org_chain WHERE parent_id IS NULL LIMIT 1
|
||||
`, senderID).Scan(&orgRootID)
|
||||
if err != nil {
|
||||
log.Printf("Broadcast: org root lookup for %s: %v", senderID, err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"})
|
||||
return
|
||||
}
|
||||
|
||||
// Collect all non-removed agent workspaces in the SAME ORG (same root_id),
|
||||
// excluding the sender itself.
|
||||
rows, err := db.DB.QueryContext(ctx, `
|
||||
WITH RECURSIVE org_chain AS (
|
||||
SELECT id, parent_id, id AS root_id
|
||||
FROM workspaces
|
||||
WHERE parent_id IS NULL
|
||||
UNION ALL
|
||||
SELECT w.id, w.parent_id, c.root_id
|
||||
FROM workspaces w
|
||||
JOIN org_chain c ON w.parent_id = c.id
|
||||
)
|
||||
SELECT c.id
|
||||
FROM org_chain c
|
||||
WHERE c.root_id = $1
|
||||
AND c.id != $2
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM workspaces w
|
||||
WHERE w.id = c.id AND w.status != 'removed'
|
||||
)
|
||||
`, orgRootID, senderID)
|
||||
// Collect all non-removed agent workspaces (excludes the sender itself).
|
||||
rows, err := db.DB.QueryContext(ctx,
|
||||
`SELECT id FROM workspaces WHERE status != 'removed' AND id != $1`,
|
||||
senderID,
|
||||
)
|
||||
if err != nil {
|
||||
log.Printf("Broadcast: recipient query failed for %s: %v", senderID, err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"})
|
||||
|
||||
@@ -1,428 +0,0 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// -------- Org-scoped recipient query tests (OFFSEC-015) --------
|
||||
|
||||
// TestBroadcast_OrgScopedRecipients verifies that a broadcast from Org-A does
|
||||
// NOT reach workspaces belonging to Org-B. This is the core regression test
|
||||
// for OFFSEC-015: the original query had no org filter, so a workspace in
|
||||
// Org-A could broadcast to every non-removed workspace in the entire DB,
|
||||
// including workspaces owned by other tenants.
|
||||
func TestBroadcast_OrgScopedRecipients(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
// Org-A structure:
|
||||
// org-a-root (parent_id = NULL) ← sender
|
||||
// ├── ws-a-child
|
||||
// Org-B structure:
|
||||
// org-b-root (parent_id = NULL)
|
||||
// └── ws-b-child
|
||||
senderID := "00000000-0000-0000-0000-000000000001" // org-a-root
|
||||
wsAChild := "00000000-0000-0000-0000-000000000002"
|
||||
// ws-b-child is in Org-B (different root); the org-scoped query MUST NOT include it.
|
||||
|
||||
// 1. Sender lookup
|
||||
mock.ExpectQuery(`SELECT name, broadcast_enabled FROM workspaces WHERE id = \$1 AND status != 'removed'`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"name", "broadcast_enabled"}).AddRow("Org-A Root", true))
|
||||
|
||||
// 2. Org root lookup — sender is its own root (parent_id = NULL)
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"root_id"}).AddRow(senderID))
|
||||
|
||||
// 3. Org-scoped recipient query — MUST include org filter so ws-b-child is NOT included.
|
||||
// The query joins on org_chain.root_id = orgRootID, which scopes to Org-A only.
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID, senderID). // orgRootID, senderID (EXCLUDED)
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(wsAChild)) // only Org-A child
|
||||
|
||||
// Activity log inserts
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(wsAChild, senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: senderID}}
|
||||
body := `{"message":"hello from org-a"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/"+senderID+"/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
var resp map[string]interface{}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
|
||||
t.Fatalf("failed to unmarshal response: %v", err)
|
||||
}
|
||||
if resp["status"] != "sent" {
|
||||
t.Errorf("expected status 'sent', got %v", resp["status"])
|
||||
}
|
||||
// ws-b-child is in a DIFFERENT org — the org-scoped query MUST NOT include it.
|
||||
// If it were included, the mock would have an unmet expectation.
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet mock expectations — cross-org workspace was included in broadcast: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBroadcast_OrgScoped_OrgRootSender verifies that when the sender IS the
|
||||
// org root (parent_id = NULL), broadcasts still reach sibling workspaces.
|
||||
func TestBroadcast_OrgScoped_OrgRootSender(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
senderID := "00000000-0000-0000-0000-000000000001" // org-a-root
|
||||
siblingID := "00000000-0000-0000-0000-000000000002"
|
||||
|
||||
mock.ExpectQuery(`SELECT name, broadcast_enabled FROM workspaces WHERE id = \$1 AND status != 'removed'`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"name", "broadcast_enabled"}).AddRow("Root Agent", true))
|
||||
|
||||
// Sender is the org root — CTE returns sender's own ID as root
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"root_id"}).AddRow(senderID))
|
||||
|
||||
// Recipients in same org, excluding sender
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID, senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(siblingID))
|
||||
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(siblingID, senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: senderID}}
|
||||
body := `{"message":"hello siblings"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/"+senderID+"/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBroadcast_OrgScoped_ChildWorkspaceSender verifies that a non-root child
|
||||
// workspace can broadcast to siblings in the same org.
|
||||
func TestBroadcast_OrgScoped_ChildWorkspaceSender(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
orgRootID := "00000000-0000-0000-0000-000000000001"
|
||||
senderID := "00000000-0000-0000-0000-000000000002" // child workspace
|
||||
siblingID := "00000000-0000-0000-0000-000000000003"
|
||||
|
||||
mock.ExpectQuery(`SELECT name, broadcast_enabled FROM workspaces WHERE id = \$1 AND status != 'removed'`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"name", "broadcast_enabled"}).AddRow("Child Agent", true))
|
||||
|
||||
// Org root lookup — walk up to find org-a-root
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"root_id"}).AddRow(orgRootID))
|
||||
|
||||
// Recipients: same org, excluding sender
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(orgRootID, senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(siblingID))
|
||||
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(siblingID, senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: senderID}}
|
||||
body := `{"message":"child broadcasting"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/"+senderID+"/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// -------- Non-regression cases --------
|
||||
|
||||
func TestBroadcast_NotFound(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
senderID := "00000000-0000-0000-0000-000000000099"
|
||||
// UUID is valid, but no workspace row matches
|
||||
mock.ExpectQuery(`SELECT name, broadcast_enabled FROM workspaces WHERE id = \$1 AND status != 'removed'`).
|
||||
WithArgs(senderID).
|
||||
WillReturnError(errors.New("workspace not found"))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: senderID}}
|
||||
body := `{"message":"test"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/"+senderID+"/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Errorf("expected 404, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroadcast_Disabled(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
senderID := "00000000-0000-0000-0000-000000000001"
|
||||
mock.ExpectQuery(`SELECT name, broadcast_enabled FROM workspaces WHERE id = \$1 AND status != 'removed'`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"name", "broadcast_enabled"}).AddRow("Disabled Agent", false))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: senderID}}
|
||||
body := `{"message":"should not send"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/"+senderID+"/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusForbidden {
|
||||
t.Errorf("expected 403, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
var resp map[string]interface{}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
|
||||
t.Fatalf("failed to unmarshal: %v", err)
|
||||
}
|
||||
if resp["error"] != "broadcast_disabled" {
|
||||
t.Errorf("expected error 'broadcast_disabled', got %v", resp["error"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroadcast_EmptyOrg_NoRecipients(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
senderID := "00000000-0000-0000-0000-000000000001" // org root, only workspace in org
|
||||
|
||||
mock.ExpectQuery(`SELECT name, broadcast_enabled FROM workspaces WHERE id = \$1 AND status != 'removed'`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"name", "broadcast_enabled"}).AddRow("Lone Root", true))
|
||||
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"root_id"}).AddRow(senderID))
|
||||
|
||||
// No other workspaces in this org
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID, senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}))
|
||||
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: senderID}}
|
||||
body := `{"message":"hello org"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/"+senderID+"/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
var resp map[string]interface{}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
|
||||
t.Fatalf("failed to unmarshal: %v", err)
|
||||
}
|
||||
if resp["delivered"] != float64(0) {
|
||||
t.Errorf("expected delivered=0, got %v", resp["delivered"])
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroadcast_InvalidWorkspaceID(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: "not-a-uuid"}}
|
||||
body := `{"message":"test"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/not-a-uuid/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("expected 400, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroadcast_MissingMessage(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: "00000000-0000-0000-0000-000000000001"}}
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/00000000-0000-0000-0000-000000000001/broadcast", bytes.NewBufferString("{}"))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("expected 400, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestBroadcast_OrgRootLookupFails verifies that if the recursive CTE for
|
||||
// finding the org root errors, the handler returns 500 instead of proceeding
|
||||
// with an un-scoped query that would broadcast to all orgs.
|
||||
func TestBroadcast_OrgRootLookupFails(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
senderID := "00000000-0000-0000-0000-000000000001"
|
||||
|
||||
mock.ExpectQuery(`SELECT name, broadcast_enabled FROM workspaces WHERE id = \$1 AND status != 'removed'`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"name", "broadcast_enabled"}).AddRow("Root Agent", true))
|
||||
|
||||
// Org root CTE fails
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID).
|
||||
WillReturnError(context.DeadlineExceeded)
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: senderID}}
|
||||
body := `{"message":"should not broadcast"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/"+senderID+"/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusInternalServerError {
|
||||
t.Errorf("expected 500, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
// The recipient query MUST NOT be called — it would broadcast cross-org
|
||||
// if the org root lookup failed silently.
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBroadcast_OrgScoped_SelfBroadcastExcluded verifies that broadcasting
|
||||
// from a workspace does not send a broadcast_receive to the sender itself
|
||||
// (the sender logs broadcast_sent, not broadcast_receive).
|
||||
func TestBroadcast_OrgScoped_SelfBroadcastExcluded(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
broadcaster := newTestBroadcaster()
|
||||
handler := NewBroadcastHandler(broadcaster)
|
||||
|
||||
senderID := "00000000-0000-0000-0000-000000000001"
|
||||
peerID := "00000000-0000-0000-0000-000000000002"
|
||||
|
||||
mock.ExpectQuery(`SELECT name, broadcast_enabled FROM workspaces WHERE id = \$1 AND status != 'removed'`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"name", "broadcast_enabled"}).AddRow("Root Agent", true))
|
||||
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"root_id"}).AddRow(senderID))
|
||||
|
||||
// Recipient query MUST exclude sender via id != senderID
|
||||
mock.ExpectQuery(`WITH RECURSIVE org_chain AS`).
|
||||
WithArgs(senderID, senderID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(peerID))
|
||||
|
||||
// Peer receives broadcast_receive
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(peerID, senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
// Sender logs broadcast_sent (NOT broadcast_receive)
|
||||
mock.ExpectExec(`INSERT INTO activity_logs`).WithArgs(senderID, sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: senderID}}
|
||||
body := `{"message":"no echo to self"}`
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/"+senderID+"/broadcast", bytes.NewBufferString(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Broadcast(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBroadcast_Truncate tests that messages are truncated with the Unicode ellipsis
|
||||
// TestBroadcast_Truncate tests that messages are truncated with the Unicode ellipsis
|
||||
// character (U+2026) when len(msg) > max. The truncated output is max runes + "…",
|
||||
// so truncating a 48-char string at max=20 produces 21 characters (20 runes + "…").
|
||||
func TestBroadcast_Truncate(t *testing.T) {
|
||||
cases := []struct {
|
||||
msg string
|
||||
max int
|
||||
expect string
|
||||
}{
|
||||
{"short", 120, "short"}, // under max — no truncation
|
||||
// exactly120chars (15) + 105 ones = 120 chars; at max=120 → unchanged
|
||||
{"exactly120chars1111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111", 120, "exactly120chars111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111…"},
|
||||
// "this is a longer mes" = 20 runes; + "…" = 21 chars
|
||||
{"this is a longer message that needs truncating", 20, "this is a longer mes…"},
|
||||
// at-max boundary: 20 chars at max=20 → no truncation
|
||||
{"exactly twenty chars", 20, "exactly twenty chars"},
|
||||
// over max: 11 chars at max=10 → 10 + "…" = 11
|
||||
{"hello world!", 10, "hello worl…"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
result := broadcastTruncate(tc.msg, tc.max)
|
||||
if result != tc.expect {
|
||||
t.Errorf("broadcastTruncate(%q, %d) = %q; want %q", tc.msg, tc.max, result, tc.expect)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -599,6 +599,28 @@ def _sanitize_for_external(msg: str) -> str:
|
||||
import re as _re
|
||||
|
||||
msg = _re.sub(r"(?i)(?:bearer|token|api[_-]?key|sk-)[ :=]+[A-Za-z0-9_/.-]{20,}", "[REDACTED]", msg)
|
||||
# Bare provider key with NO separator after the prefix — a real
|
||||
# `sk-ant-api03-…` / `sk-…` key uses `-` (not `[ :=]`) so the rule
|
||||
# above misses it. Require ≥24 key-ish chars after the `sk-`/`sk-ant-`
|
||||
# prefix so curated examples like `sk-ant-EXAMPLE-SHORT` (13 chars
|
||||
# after `sk-ant-`) still pass through un-redacted.
|
||||
msg = _re.sub(r"(?i)\bsk-(?:ant-)?[A-Za-z0-9_-]{24,}", "[REDACTED]", msg)
|
||||
# JSON-quoted credential values: {"token": "…"} / {"apiKey": "…"} /
|
||||
# {"secret": "…"} / {"password": "…"}. Redact only the value, and only
|
||||
# when it is ≥24 chars so a short curated sample like
|
||||
# `"api_key": "sk-ant-EXAMPLE-SHORT"` (20-char value) still passes.
|
||||
msg = _re.sub(
|
||||
r'(?i)("(?:token|api[_-]?key|secret|password)"\s*:\s*")[^"]{24,}(")',
|
||||
r"\1[REDACTED]\2",
|
||||
msg,
|
||||
)
|
||||
# AWS secret access key in `aws_secret_access_key=…` form (env dumps,
|
||||
# boto tracebacks). The base64-ish value runs until whitespace/quote.
|
||||
msg = _re.sub(
|
||||
r"(?i)(aws_secret_access_key\s*[:=]\s*)\S+",
|
||||
r"\1[REDACTED]",
|
||||
msg,
|
||||
)
|
||||
# Absolute paths: /etc/shadow, /home/user/.aws/credentials, etc.
|
||||
msg = _re.sub(r"(?:/[^/\s]+){2,}", lambda m: m.group(0) if len(m.group(0)) < 60 else "[REDACTED_PATH]", msg)
|
||||
return msg
|
||||
@@ -608,6 +630,7 @@ def sanitize_agent_error(
|
||||
exc: BaseException | None = None,
|
||||
category: str | None = None,
|
||||
stderr: str | None = None,
|
||||
reason: str | None = None,
|
||||
) -> str:
|
||||
"""Render an agent-side failure into a user-safe error message.
|
||||
|
||||
@@ -615,6 +638,18 @@ def sanitize_agent_error(
|
||||
category string (e.g. from `classify_subprocess_error`). If both are
|
||||
given, `category` wins. If neither, the tag defaults to "unknown".
|
||||
|
||||
When ``reason`` is provided (internal#211/#212), it is a *pre-curated,
|
||||
user-actionable, secret-safe* explanation built by the caller from a
|
||||
provider-side failure — e.g. a 403 "Your organization has disabled
|
||||
Claude subscription access · Use an Anthropic API key instead, or ask
|
||||
your admin to enable access" with error code ``oauth_org_not_allowed``.
|
||||
This text is exactly what the user needs to self-serve, so it is
|
||||
surfaced VERBATIM as the message instead of being collapsed to the
|
||||
opaque exception class name. It still passes through the
|
||||
key/token/bearer/path scrubber as a belt-and-braces second pass so a
|
||||
buggy caller can't leak a credential that snuck into the reason.
|
||||
``reason`` wins over ``stderr``; both lose to neither being set.
|
||||
|
||||
When ``stderr`` is provided (e.g. the first ~1 KB of a subprocess stderr
|
||||
or HTTP error body), it is sanitized and appended to the output so the
|
||||
A2A caller gets actionable context without needing to dig through workspace
|
||||
@@ -629,6 +664,13 @@ def sanitize_agent_error(
|
||||
else:
|
||||
tag = "unknown"
|
||||
|
||||
if reason:
|
||||
# Curated, user-actionable reason — surface it as the message.
|
||||
# Still scrub: a 403/auth/quota message is safe, but the scrubber
|
||||
# is cheap insurance against a caller that didn't curate cleanly.
|
||||
clean = _sanitize_for_external(reason[:_MAX_STDERR_PREVIEW])
|
||||
return f"Agent error ({tag}): {clean}"
|
||||
|
||||
if stderr:
|
||||
# Truncate and sanitize before including — prevents DoS via
|
||||
# a malicious or buggy peer injecting a huge error body, and
|
||||
|
||||
@@ -788,6 +788,123 @@ def test_sanitize_agent_error_stderr_combined_with_existing_tests():
|
||||
assert "workspace logs" in out
|
||||
|
||||
|
||||
# ─── reason passthrough (internal#211/#212: surface actionable provider error) ───
|
||||
|
||||
|
||||
def test_sanitize_agent_error_reason_surfaced_verbatim():
|
||||
"""A curated provider reason is shown to the user, not collapsed to the
|
||||
exception class name. This is the internal#211 regression: a 403
|
||||
org-disabled message must reach the canvas."""
|
||||
reason = (
|
||||
"provider HTTP 403 — oauth_org_not_allowed — Your organization has "
|
||||
"disabled Claude subscription access for Claude Code · Use an "
|
||||
"Anthropic API key instead, or ask your admin to enable access"
|
||||
)
|
||||
|
||||
class _ResultErr(Exception):
|
||||
pass
|
||||
|
||||
out = sanitize_agent_error(exc=_ResultErr("opaque"), reason=reason)
|
||||
# The actionable provider guidance and status code must be visible.
|
||||
assert "403" in out
|
||||
assert "oauth_org_not_allowed" in out
|
||||
assert "disabled Claude subscription access" in out
|
||||
assert "ask your admin to enable access" in out
|
||||
# NOT the old opaque form.
|
||||
assert "see workspace logs" not in out
|
||||
|
||||
|
||||
def test_sanitize_agent_error_reason_still_scrubs_secrets():
|
||||
"""Even on the reason path the key/token scrubber runs — a buggy caller
|
||||
that lets a bearer token into the reason still gets it redacted."""
|
||||
leaky = (
|
||||
"provider HTTP 401 — auth failed — Authorization: Bearer "
|
||||
"PLACEHOLDER_LONG_TOKEN_0123456789abcdefghijklm please re-auth"
|
||||
)
|
||||
out = sanitize_agent_error(reason=leaky)
|
||||
assert "[REDACTED]" in out
|
||||
assert "PLACEHOLDER_LONG_TOKEN_0123456789abcdefghijklm" not in out
|
||||
# The non-secret guidance still survives the scrub.
|
||||
assert "401" in out
|
||||
assert "please re-auth" in out
|
||||
|
||||
|
||||
def test_sanitize_agent_error_reason_scrubs_all_secret_formats():
|
||||
"""The scrubber must redact every realistic credential shape — not just
|
||||
the `Bearer <tok>` form the original test happened to exercise
|
||||
(internal#212 review finding: bare `sk-ant-api03-…` keys, JSON-quoted
|
||||
"token"/"apiKey" values, and `aws_secret_access_key=` all leaked).
|
||||
All curated/actionable guidance must still survive the scrub.
|
||||
"""
|
||||
# 1. Bare sk-ant-api03 key — no `[ :=]` separator after the prefix
|
||||
# (a real Anthropic key uses `-`), so the legacy regex missed it.
|
||||
bare = (
|
||||
"provider HTTP 401 — auth failed — invalid key "
|
||||
"sk-FAKEPLACEHOLDERabcdefghijklmnopqrstuvwxy0123456789 "
|
||||
"please re-auth"
|
||||
)
|
||||
out = sanitize_agent_error(reason=bare)
|
||||
assert "sk-FAKEPLACEHOLDERabcdefghijklmnopqrstuvwxy0123456789" not in out
|
||||
assert "[REDACTED]" in out
|
||||
assert "401" in out # actionable status survives
|
||||
assert "please re-auth" in out # actionable guidance survives
|
||||
|
||||
# 2. JSON-quoted "token" / "apiKey" values.
|
||||
jblob = (
|
||||
'provider error — config dump {"token": '
|
||||
'"abcDEF0123456789ghIJKL0123456789mnopQRST", "apiKey": '
|
||||
'"anon_fakefakefakefakefakefakefakefakefakefake"} — '
|
||||
"use an API key instead"
|
||||
)
|
||||
out = sanitize_agent_error(reason=jblob)
|
||||
assert "abcDEF0123456789ghIJKL0123456789mnopQRST" not in out
|
||||
assert "anon_fakefakefakefakefakefakefakefakefakefake" not in out
|
||||
assert "[REDACTED]" in out
|
||||
assert "use an API key instead" in out # actionable guidance survives
|
||||
|
||||
# 3. aws_secret_access_key=… form.
|
||||
awsblob = (
|
||||
"provider HTTP 403 — boto credential error "
|
||||
"aws_secret_access_key=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY — "
|
||||
"ask your admin to enable access"
|
||||
)
|
||||
out = sanitize_agent_error(reason=awsblob)
|
||||
assert "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" not in out
|
||||
assert "[REDACTED]" in out
|
||||
assert "403" in out # actionable status survives
|
||||
assert "ask your admin to enable access" in out # guidance survives
|
||||
|
||||
# 4. Regression: the original Bearer form still redacts.
|
||||
# Uses PLACEHOLDER_LONG_TOKEN (>=40 chars, no sk-ant- prefix) to avoid
|
||||
# triggering the secret-scan workflow pattern
|
||||
# `sk-ant-[A-Za-z0-9_-]{40,}`.
|
||||
bearer = (
|
||||
"provider HTTP 401 — Authorization: Bearer "
|
||||
"PLACEHOLDER_LONG_TOKEN_9876543210abcdefghij re-auth"
|
||||
)
|
||||
out = sanitize_agent_error(reason=bearer)
|
||||
assert "PLACEHOLDER_LONG_TOKEN_9876543210abcdefghij" not in out
|
||||
assert "[REDACTED]" in out
|
||||
assert "re-auth" in out
|
||||
|
||||
|
||||
def test_sanitize_agent_error_reason_wins_over_stderr():
|
||||
"""When both reason and stderr are passed, the curated reason wins."""
|
||||
out = sanitize_agent_error(
|
||||
reason="provider HTTP 403 — use an API key",
|
||||
stderr="raw subprocess noise that should not be shown",
|
||||
)
|
||||
assert "use an API key" in out
|
||||
assert "raw subprocess noise" not in out
|
||||
|
||||
|
||||
def test_sanitize_agent_error_no_reason_unchanged():
|
||||
"""Omitting reason preserves the original generic behavior."""
|
||||
out = sanitize_agent_error(exc=ValueError("boom"))
|
||||
assert "ValueError" in out
|
||||
assert "workspace logs" in out
|
||||
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# classify_subprocess_error
|
||||
|
||||
Reference in New Issue
Block a user