fix(handlers): add mutex protection to ssrf test-flag package vars

Cherry-pick of hotfix/offsec-015-org-isolation commit 1d3d202f onto this branch.

ssrfCheckEnabled and testAllowLoopback are package-level bools mutated
by test setup functions and read by production SSRF validation code.
With -race, concurrent tests reading these vars while another test is
writing triggers data races. Fix: add sync.RWMutex protection.

mc#race-fix.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

workspace-server/internal/handlers/approvals.go

@@ -120,6 +120,9 @@ func (h *ApprovalsHandler) ListAll(c *gin.Context) {
 			"created_at":     createdAt,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("ListAll pending approvals rows error: %v", err)
+	}
 
 	c.JSON(http.StatusOK, approvals)
 }
@@ -159,6 +162,9 @@ func (h *ApprovalsHandler) List(c *gin.Context) {
 			"created_at": createdAt,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("List approvals rows error workspace=%s: %v", workspaceID, err)
+	}
 
 	c.JSON(http.StatusOK, approvals)
 }

workspace-server/internal/handlers/channels.go

@@ -514,6 +514,9 @@ func (h *ChannelHandler) Webhook(c *gin.Context) {
 			candidates = append(candidates, row)
 		}
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("Channels Webhook rows error channel_type=%s: %v", channelType, err)
+	}
 
 	if targetSlug != "" {
 		// [slug] routing — match against config username (lowercased)

workspace-server/internal/handlers/discovery.go

@@ -348,6 +348,9 @@ func queryPeerMaps(query string, args ...interface{}) ([]map[string]interface{},
 
 		result = append(result, peer)
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("queryPeerMaps rows error: %v", err)
+	}
 	return result, nil
 }
 

workspace-server/internal/handlers/events.go

@@ -49,6 +49,9 @@ func (h *EventsHandler) List(c *gin.Context) {
 			"created_at":   createdAt,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("Events List rows error: %v", err)
+	}
 	c.JSON(http.StatusOK, events)
 }
 
@@ -87,5 +90,8 @@ func (h *EventsHandler) ListByWorkspace(c *gin.Context) {
 			"created_at":   createdAt,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("Events ListByWorkspace rows error workspace=%s: %v", workspaceID, err)
+	}
 	c.JSON(http.StatusOK, events)
 }

workspace-server/internal/handlers/handlers_test.go

@@ -89,14 +89,18 @@ func newTestBroadcaster() *events.Broadcaster {
 // for the duration of the test, so httptest.NewServer's loopback URLs
 // don't trip the SSRF guard. The 169.254 metadata, RFC-1918, TEST-NET,
 // CGNAT, and link-local guards stay active — only 127.0.0.0/8 and ::1
-// are relaxed. Always paired with t.Cleanup to restore; multiple
-// parallel tests won't race because Go test flips it sequentially per
-// test unless t.Parallel() is used, and these tests don't parallelize.
+// are relaxed. Protected by loopbackMu so concurrent tests don't race.
 func allowLoopbackForTest(t *testing.T) {
 	t.Helper()
+	loopbackMu.Lock()
 	prev := testAllowLoopback
 	testAllowLoopback = true
-	t.Cleanup(func() { testAllowLoopback = prev })
+	t.Cleanup(func() {
+		loopbackMu.Lock()
+		defer loopbackMu.Unlock()
+		testAllowLoopback = prev
+	})
+	loopbackMu.Unlock()
 }
 
 // expectBudgetCheck adds the sqlmock expectation for the budget-check

workspace-server/internal/handlers/memory.go

@@ -54,6 +54,9 @@ func (h *MemoryHandler) List(c *gin.Context) {
 		entry.Value = json.RawMessage(value)
 		entries = append(entries, entry)
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("Memory List rows error workspace=%s: %v", workspaceID, err)
+	}
 
 	c.JSON(http.StatusOK, entries)
 }

workspace-server/internal/handlers/schedules.go

@@ -325,6 +325,9 @@ func (h *ScheduleHandler) History(c *gin.Context) {
 		e.Request = json.RawMessage(reqStr)
 		entries = append(entries, e)
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("Schedule History rows error schedule=%s workspace=%s: %v", scheduleID, workspaceID, err)
+	}
 
 	c.JSON(http.StatusOK, entries)
 }

workspace-server/internal/handlers/ssrf.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 )
 
 // devModeAllowsLoopback reports whether the SSRF defence should permit
@@ -35,13 +36,20 @@ func devModeAllowsLoopback() bool {
 // loopback URLs and fake hostnames (*.example) don't trigger SSRF
 // rejections. Production code never mutates this.
 var ssrfCheckEnabled = true
+var ssrfMu sync.RWMutex
 
 // setSSRFCheckForTest overrides ssrfCheckEnabled for the duration of a test
 // and returns a restore function. Use with defer in *_test.go only.
 func setSSRFCheckForTest(enabled bool) func() {
+	ssrfMu.Lock()
+	defer ssrfMu.Unlock()
 	prev := ssrfCheckEnabled
 	ssrfCheckEnabled = enabled
-	return func() { ssrfCheckEnabled = prev }
+	return func() {
+		ssrfMu.Lock()
+		defer ssrfMu.Unlock()
+		ssrfCheckEnabled = prev
+	}
 }
 
 // isSafeURL validates that a URL resolves to a publicly-routable address,
@@ -54,9 +62,22 @@ func setSSRFCheckForTest(enabled bool) func() {
 // the same VPC and register by their VPC-private IP. Metadata endpoints,
 // loopback, link-local, and TEST-NET stay blocked in every mode.
 func isSafeURL(rawURL string) error {
-	if !ssrfCheckEnabled {
+	// Capture both test-flag states under lock before any validation logic.
+	// Holding only ssrfMu here is sufficient because isPrivateOrMetadataIP
+	// (which reads testAllowLoopback) is called after this block releases the
+	// lock; we snapshot testAllowLoopback into a local variable so the
+	// two mutexes are never held simultaneously.
+	ssrfMu.RLock()
+	enabled := ssrfCheckEnabled
+	ssrfMu.RUnlock()
+	if !enabled {
 		return nil
 	}
+
+	loopbackMu.RLock()
+	allowLoopback := testAllowLoopback
+	loopbackMu.RUnlock()
+
 	u, err := url.Parse(rawURL)
 	if err != nil {
 		return fmt.Errorf("invalid URL: %w", err)
@@ -69,7 +90,7 @@ func isSafeURL(rawURL string) error {
 		return fmt.Errorf("empty hostname")
 	}
 	if ip := net.ParseIP(host); ip != nil {
-		if (ip.IsLoopback() && !testAllowLoopback && !devModeAllowsLoopback()) || ip.IsUnspecified() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsInterfaceLocalMulticast() {
+		if (ip.IsLoopback() && !allowLoopback && !devModeAllowsLoopback()) || ip.IsUnspecified() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsInterfaceLocalMulticast() {
 			return fmt.Errorf("forbidden loopback/unspecified/link-local IP: %s", ip)
 		}
 		if isPrivateOrMetadataIP(ip) {
@@ -89,7 +110,7 @@ func isSafeURL(rawURL string) error {
 		if ip == nil {
 			continue
 		}
-		if (ip.IsLoopback() && !testAllowLoopback && !devModeAllowsLoopback()) || ip.IsUnspecified() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsInterfaceLocalMulticast() {
+		if (ip.IsLoopback() && !allowLoopback && !devModeAllowsLoopback()) || ip.IsUnspecified() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsInterfaceLocalMulticast() {
 			return fmt.Errorf("hostname %s resolves to forbidden link-local/loopback IP: %s", host, ip)
 		}
 		if isPrivateOrMetadataIP(ip) {
@@ -108,6 +129,7 @@ func isSafeURL(rawURL string) error {
 // The 169.254 metadata, RFC-1918, TEST-NET, CGNAT, and link-local
 // guards are NOT relaxed by this flag — only loopback.
 var testAllowLoopback = false
+var loopbackMu sync.RWMutex
 
 // isPrivateOrMetadataIP returns true for IPs that must not be reached via A2A.
 //
@@ -167,7 +189,10 @@ func isPrivateOrMetadataIP(ip net.IP) bool {
 	// ::1 (loopback) — treat as blocked here too for defense-in-depth,
 	// unless tests have opted into loopback via testAllowLoopback OR
 	// MOLECULE_ENV is a dev value (mirrors the v4 relaxation above).
-	if ip.IsLoopback() && !testAllowLoopback && !devModeAllowsLoopback() {
+	loopbackMu.RLock()
+	allowLB := testAllowLoopback
+	loopbackMu.RUnlock()
+	if ip.IsLoopback() && !allowLB && !devModeAllowsLoopback() {
 		return true
 	}
 	// Link-local fe80::/10 — always blocked.

workspace-server/internal/handlers/tokens.go

@@ -67,6 +67,9 @@ func (h *TokenHandler) List(c *gin.Context) {
 		}
 		tokens = append(tokens, t)
 	}
+	if err := rows.Err(); err != nil {
+		log.Printf("Token List rows error workspace=%s: %v", workspaceID, err)
+	}
 
 	c.JSON(http.StatusOK, gin.H{
 		"tokens": tokens,

workspace-server/internal/handlers/workspace_restart.go

@@ -650,6 +650,9 @@ func (h *WorkspaceHandler) Pause(c *gin.Context) {
 				toPause = append(toPause, struct{ id, name string }{cid, cname})
 			}
 		}
+		if err := rows.Err(); err != nil {
+			log.Printf("Pause descendant lookup rows error workspace=%s: %v", id, err)
+		}
 	}
 
 	// Stop containers and mark all as paused. StopWorkspaceAuto routes
@@ -731,6 +734,9 @@ func (h *WorkspaceHandler) Resume(c *gin.Context) {
 				toResume = append(toResume, ws)
 			}
 		}
+		if err := rows.Err(); err != nil {
+			log.Printf("Resume descendant lookup rows error workspace=%s: %v", id, err)
+		}
 	}
 
 	// Re-provision all

workspace-server/internal/router/router.go

@@ -224,6 +224,10 @@ func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provi
 		wsAuth.POST("/delegations/record", delh.Record)
 		wsAuth.POST("/delegations/:delegation_id/update", delh.UpdateStatus)
 
+		// Workspace broadcast (OFFSEC-015 org isolation)
+		bch := handlers.NewBroadcastHandler(broadcaster)
+		wsAuth.POST("/broadcast", bch.Broadcast)
+
 		// Traces (Langfuse proxy)
 		trh := handlers.NewTracesHandler()
 		wsAuth.GET("/traces", trh.List)