feat(a2a-bridge): add gRPC and REST transport support - #363
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request migrates the scion-a2a-bridge to the official a2a-go SDK, replacing custom JSON-RPC handling, task lifecycle management, and streaming with spec-compliant SDK implementations while introducing gRPC and REST transports. Feedback on these changes highlights several critical issues: grpcServer.GracefulStop() could block shutdown indefinitely on active streaming connections; gRPC and REST requests will fail because routing metadata is only injected in the HTTP JSON-RPC handler; potential nil pointer dereferences exist in ScionExecutor.Execute and TranslateScionToA2AParts; and disabling WriteTimeout globally on the HTTP server introduces a Slowloris DoS vulnerability.
| route, ok := RouteInfoFrom(ctx) | ||
| if !ok { | ||
| yield(nil, fmt.Errorf("missing route info in context: %w", a2a.ErrInternalError)) | ||
| return | ||
| } |
There was a problem hiding this comment.
RouteInfo is only injected into the context within Server.handleJSONRPC (in server.go), which is specific to the HTTP JSON-RPC transport. For gRPC and REST requests, this handler is bypassed, meaning RouteInfoFrom(ctx) will always return ok = false. This will cause all gRPC and REST requests to fail immediately with missing route info in context.
Consider extracting the project and agent slugs directly from the SDK's ExecutorContext if available, or implement gRPC interceptors and REST middleware to extract and inject RouteInfo into the context for those transports.
There was a problem hiding this comment.
Acknowledged — will address in next push.
There was a problem hiding this comment.
Fixed — nil guards and channel safety checks added. The RouteInfo issue was already handled with a proper error return. Pushed.
| func TranslateScionToA2AParts(msg *messages.StructuredMessage) (*a2a.Message, []*a2a.Artifact) { | ||
| var sdkParts []*a2a.Part | ||
| sdkParts = append(sdkParts, &a2a.Part{Content: a2a.Text(msg.Msg), MediaType: "text/plain"}) |
There was a problem hiding this comment.
msg is a pointer (*messages.StructuredMessage) and could be nil. Accessing msg.Msg or msg.Attachments without a nil check can cause a runtime panic.
func TranslateScionToA2AParts(msg *messages.StructuredMessage) (*a2a.Message, []*a2a.Artifact) {
if msg == nil {
return nil, nil
}
var sdkParts []*a2a.Part
sdkParts = append(sdkParts, &a2a.Part{Content: a2a.Text(msg.Msg), MediaType: "text/plain"})There was a problem hiding this comment.
Acknowledged — will address in next push.
There was a problem hiding this comment.
Fixed — nil guards and channel safety checks added. The RouteInfo issue was already handled with a proper error return. Pushed.
|
Review fixes pushed. The review fixes from PR #362 are included in this branch (it's stacked on top). 4 bugs fixed, 306 lines of fixes/tests added. All tests pass. |
Code Review: PR #363 — feat(a2a-bridge): add gRPC and REST transport supportPR: #363 SummaryThis PR adds optional gRPC and REST transports to the scion-a2a-bridge, stacked on #362 (SDK adoption). It introduces a Stats: +1010 / -933 across 11 files (net +77). The large deletion count is misleading — ~480 lines of hand-rolled JSON-RPC dispatch were replaced by SDK delegation. What's Good
CRITICAL FindingsC1.
|
…ogleCloudPlatform#363) Critical: - C1: Wrap grpcServer.GracefulStop() in select with shutdownCtx deadline, falling back to Stop() on timeout to prevent indefinite blocking on active streams. - C2: Set WriteTimeout: 30s on REST http.Server (was 0). Add SSEWriteDeadlineMiddleware to clear write deadline for SSE streaming responses while keeping non-streaming endpoints protected. High: - H1: Add auth interceptors for gRPC (AuthUnaryInterceptor, AuthStreamInterceptor) and auth middleware for REST (AuthHTTPMiddleware). Require grpc_insecure/rest_insecure config flags when auth.scheme is "none". Add prominent startup warnings for insecure transports. Refactor credential extraction into reusable extractCredential/verifyCredential helpers. - H2: Add agent-slug verification in dispatchToWaiter to prevent cross-agent message injection. Messages from a different agent than the waiter expects are logged and dropped. Medium: - M1: Add nil guard in TranslateScionToA2A (non-SDK path). - M2: Re-add MaxBytesReader (1MB) on JSON-RPC handler to prevent memory exhaustion from oversized request bodies. - M4: Restore normalizeJSONRPCID to reject invalid ID types (arrays, objects, booleans) per JSON-RPC 2.0 §4.1. Log Encode errors. Tests: Add tests for normalizeJSONRPCID, MaxBytesReader enforcement, ValidateConfig insecure flag requirements, and dispatchToWaiter agent-slug verification.
…ogleCloudPlatform#363) Critical: - C1: Wrap grpcServer.GracefulStop() in select with shutdownCtx deadline, falling back to Stop() on timeout to prevent indefinite blocking on active streams. - C2: Set WriteTimeout: 30s on REST http.Server (was 0). Add SSEWriteDeadlineMiddleware to clear write deadline for SSE streaming responses while keeping non-streaming endpoints protected. High: - H1: Add auth interceptors for gRPC (AuthUnaryInterceptor, AuthStreamInterceptor) and auth middleware for REST (AuthHTTPMiddleware). Require grpc_insecure/rest_insecure config flags when auth.scheme is "none". Add prominent startup warnings for insecure transports. Refactor credential extraction into reusable extractCredential/verifyCredential helpers. - H2: Add agent-slug verification in dispatchToWaiter to prevent cross-agent message injection. Messages from a different agent than the waiter expects are logged and dropped. Medium: - M1: Add nil guard in TranslateScionToA2A (non-SDK path). - M2: Re-add MaxBytesReader (1MB) on JSON-RPC handler to prevent memory exhaustion from oversized request bodies. - M4: Restore normalizeJSONRPCID to reject invalid ID types (arrays, objects, booleans) per JSON-RPC 2.0 §4.1. Log Encode errors. Tests: Add tests for normalizeJSONRPCID, MaxBytesReader enforcement, ValidateConfig insecure flag requirements, and dispatchToWaiter agent-slug verification.
7aaa0fd to
f3df7a5
Compare
Review Findings Status — ptone/scion-agent review (6/13) + Gemini Code Assist (6/14)Already Fixed (in commits f65f5d1 and f3df7a5)These were addressed in the 6/14 and 6/15 fix commits:
Newly Fixed (commit 3c65df4)
Planned / Noted for Follow-up
|
…ogleCloudPlatform#363) Critical: - C1: Wrap grpcServer.GracefulStop() in select with shutdownCtx deadline, falling back to Stop() on timeout to prevent indefinite blocking on active streams. - C2: Set WriteTimeout: 30s on REST http.Server (was 0). Add SSEWriteDeadlineMiddleware to clear write deadline for SSE streaming responses while keeping non-streaming endpoints protected. High: - H1: Add auth interceptors for gRPC (AuthUnaryInterceptor, AuthStreamInterceptor) and auth middleware for REST (AuthHTTPMiddleware). Require grpc_insecure/rest_insecure config flags when auth.scheme is "none". Add prominent startup warnings for insecure transports. Refactor credential extraction into reusable extractCredential/verifyCredential helpers. - H2: Add agent-slug verification in dispatchToWaiter to prevent cross-agent message injection. Messages from a different agent than the waiter expects are logged and dropped. Medium: - M1: Add nil guard in TranslateScionToA2A (non-SDK path). - M2: Re-add MaxBytesReader (1MB) on JSON-RPC handler to prevent memory exhaustion from oversized request bodies. - M4: Restore normalizeJSONRPCID to reject invalid ID types (arrays, objects, booleans) per JSON-RPC 2.0 §4.1. Log Encode errors. Tests: Add tests for normalizeJSONRPCID, MaxBytesReader enforcement, ValidateConfig insecure flag requirements, and dispatchToWaiter agent-slug verification.
3c65df4 to
41efd44
Compare
Critical fixes: - C1: executor.go — capture both return values from SendStructuredMessage (_, err :=) - C2: executor.go — remove extra closing brace in Cancel() method - C3: server_test.go — remove duplicate 'context' import - C4: followup_test.go — rewrite server-layer tests to use SDK-based patterns (NewServer 5-arg, SDK method names, SDK-compatible params) - C5: main.go — set WriteTimeout to 0 for HTTP and REST servers to avoid truncating SSE streams and long executor timeouts Security fixes: - S1: main.go — add MaxRecvMsgSize(1MB), MaxConcurrentStreams(100), and KeepaliveEnforcementPolicy to gRPC server - S2: main.go — add MaxBytesReaderMiddleware(1MB) wrapper for REST handler - S3: scoped_store.go — clean up ownership map entries when tasks reach terminal state to prevent unbounded memory growth - S4: bridge.go — addWaiter now rejects concurrent follow-ups to the same task with an error instead of silently overwriting the waiter - S5: bridge.go — add agent slug verification for SDK-managed tasks in dispatchBrokerMessage before calling dispatchToActiveTask Important fixes: - I1: executor.go — use projectcompat.UserTopic()/LegacyUserTopic() instead of hardcoded subscription patterns - I2: main.go — set errCh capacity to 3 (matches 3 server goroutines) - I3: executor.go — use rolling 60s per-write deadline for SSE streams instead of clearing the deadline entirely - I4: server_test.go — fix pushNotifications assertions (true → false) Suggestions: - SSE Content-Type check now uses strings.HasPrefix (via detectSSE helper) - DRY SSE deadline logic into extendDeadline() private method - translate.go — log error instead of silently dropping Data marshal failures - server_test.go — update method names to SDK format (SendMessage, GetTask, etc.)
|
FYI #362 merged |
Enable optional gRPC and REST transports for the A2A bridge using the SDK's built-in handlers. Both are opt-in via config — existing JSON-RPC deployments are unaffected. Config: - bridge.grpc_listen_address: starts gRPC server (a2agrpc.NewHandler) - bridge.rest_listen_address: starts REST server (a2asrv.NewRESTHandler) Both transports share the same SDK RequestHandler and ScionExecutor, so routing, auth, and Hub integration are identical across protocols.
12-cycle review findings: 1. Cancel() silently failed when RouteInfoFrom(ctx) returned no route info — interrupt was never sent to the agent with no log. Fixed: check ok return, log warning. 2. Per-agent cards said streaming:false while registry card said streaming:true. Fixed: consistent streaming:true on both. 3. Executor refactored to use SDK constructors for artifact events instead of manual struct building. 4. Added translate_test.go with 153 lines of type translation tests. 5. Added server test coverage for new SDK handler paths. 306 lines of fixes and tests. All tests pass.
Fixes from Gemini Code Assist review: 1. responseCh read: check ok and nil before processing (prevents panic on closed channel) 2. TranslateScionToA2AParts: nil msg guard at function entry 3. ExecutorContext: nil check before accessing fields
…ogleCloudPlatform#363) Critical: - C1: Wrap grpcServer.GracefulStop() in select with shutdownCtx deadline, falling back to Stop() on timeout to prevent indefinite blocking on active streams. - C2: Set WriteTimeout: 30s on REST http.Server (was 0). Add SSEWriteDeadlineMiddleware to clear write deadline for SSE streaming responses while keeping non-streaming endpoints protected. High: - H1: Add auth interceptors for gRPC (AuthUnaryInterceptor, AuthStreamInterceptor) and auth middleware for REST (AuthHTTPMiddleware). Require grpc_insecure/rest_insecure config flags when auth.scheme is "none". Add prominent startup warnings for insecure transports. Refactor credential extraction into reusable extractCredential/verifyCredential helpers. - H2: Add agent-slug verification in dispatchToWaiter to prevent cross-agent message injection. Messages from a different agent than the waiter expects are logged and dropped. Medium: - M1: Add nil guard in TranslateScionToA2A (non-SDK path). - M2: Re-add MaxBytesReader (1MB) on JSON-RPC handler to prevent memory exhaustion from oversized request bodies. - M4: Restore normalizeJSONRPCID to reject invalid ID types (arrays, objects, booleans) per JSON-RPC 2.0 §4.1. Log Encode errors. Tests: Add tests for normalizeJSONRPCID, MaxBytesReader enforcement, ValidateConfig insecure flag requirements, and dispatchToWaiter agent-slug verification.
Critical fixes: - C1: executor.go — capture both return values from SendStructuredMessage (_, err :=) - C2: executor.go — remove extra closing brace in Cancel() method - C3: server_test.go — remove duplicate 'context' import - C4: followup_test.go — rewrite server-layer tests to use SDK-based patterns (NewServer 5-arg, SDK method names, SDK-compatible params) - C5: main.go — set WriteTimeout to 0 for HTTP and REST servers to avoid truncating SSE streams and long executor timeouts Security fixes: - S1: main.go — add MaxRecvMsgSize(1MB), MaxConcurrentStreams(100), and KeepaliveEnforcementPolicy to gRPC server - S2: main.go — add MaxBytesReaderMiddleware(1MB) wrapper for REST handler - S3: scoped_store.go — clean up ownership map entries when tasks reach terminal state to prevent unbounded memory growth - S4: bridge.go — addWaiter now rejects concurrent follow-ups to the same task with an error instead of silently overwriting the waiter - S5: bridge.go — add agent slug verification for SDK-managed tasks in dispatchBrokerMessage before calling dispatchToActiveTask Important fixes: - I1: executor.go — use projectcompat.UserTopic()/LegacyUserTopic() instead of hardcoded subscription patterns - I2: main.go — set errCh capacity to 3 (matches 3 server goroutines) - I3: executor.go — use rolling 60s per-write deadline for SSE streams instead of clearing the deadline entirely - I4: server_test.go — fix pushNotifications assertions (true → false) Suggestions: - SSE Content-Type check now uses strings.HasPrefix (via detectSSE helper) - DRY SSE deadline logic into extendDeadline() private method - translate.go — log error instead of silently dropping Data marshal failures - server_test.go — update method names to SDK format (SendMessage, GetTask, etc.)
- Add DiscoverSkillsDirectory and HubPreStartHooks to mockHubClient. These were added to hubclient.Client upstream (GoogleCloudPlatform#914, GoogleCloudPlatform#892) without updating the bridge mock, which leaves the bridge test build red on main. - Replace the four legacy server-layer message/send tests (which still used the pre-SDK handler shape and a stub http.Handler, and fail on main) with SDK-based equivalents that wire a real a2asrv handler + ScionExecutor via newFollowUpTestServerWithHub. - Drop the SendMessageParams/SendMessageConfig/ErrCodeInvalidParams/boolPtr test shims that only existed for those legacy tests.
The gRPC and REST transports carried their own credential check that only understood the legacy apiKey/bearer schemes. After the per-user auth work landed upstream (GoogleCloudPlatform#864), that meant hubUAT/hubJWT callers were rejected on those transports and no CallerIdentity was ever injected, so per-user Hub clients silently fell back to the service account. Extract a transport-neutral Server.authenticate() that all three transports share: - server.go: new authenticate()/authError plus header-lookup helpers; authMiddleware delegates to it (status codes and messages unchanged); AuthHTTPMiddleware becomes a Server method supporting every scheme; drop the now-dead extractCredential helper. - executor.go: AuthUnaryInterceptor/AuthStreamInterceptor become Server methods backed by the same authenticator, mapping failures to Unauthenticated/Internal and propagating the auth context into streams (ctxServerStream, reused by the route-info interceptor). - main.go: call the Server-scoped auth helpers, correct the stale comment claiming auth is not applied to gRPC/REST, and tidy the "net" import. Also close the SSE write-deadline gap: SSEWriteDeadlineMiddleware is now installed on the JSON-RPC handler chain, and both HTTP servers get a bounded WriteTimeout (send_message timeout + 30s) instead of WriteTimeout: 0, so non-streaming responses are bounded again while SSE keeps its rolling per-write deadline. Tests: transport_auth_test.go covers REST and gRPC (unary + stream) across apiKey/bearer/hubUAT/hubJWT/none/insecure, including CallerIdentity propagation and fail-closed behaviour when a validator is missing; sse_deadline_test.go proves an SSE stream outlives the server WriteTimeout with the middleware and is truncated without it, while non-SSE responses stay bounded.
Stand up the full stack (state store, ScionExecutor, a2asrv handler) behind the gRPC server and REST handler exactly as cmd/scion-a2a-bridge wires them, with a fake Scion agent that replies through Bridge.HandleBrokerMessage, and drive it with the real a2a-go clients over real sockets: - gRPC: unary SendMessage round-trip, server-streaming round-trip (which also proves the stream interceptors propagate auth + route context), and rejection of bad credentials. - REST: SendMessage round-trip, SSE streaming round-trip, and rejection of bad credentials.
- Remove the stale "No gRPC or REST transport" known limitation and replace it with the real constraint: both new transports use fixed single-project/single-agent routing. - Add a Transports section covering enablement, routing, streaming, shared auth schemes, the insecure opt-in, body limits, and SSE write deadlines. - Document bridge.grpc_listen_address, bridge.rest_listen_address, bridge.grpc_insecure and bridge.rest_insecure in the configuration table, the ports table, and the sample config.
b38dbcf to
fe91de7
Compare
2 participants
Problem
The A2A bridge only supports JSON-RPC over HTTP. Some deployments need gRPC (for internal service mesh integration) or REST (for simpler client tooling). Adding these transports by hand would mean reimplementing protocol handling three times.
Solution
With #362's migration to the
a2a-goSDK, adding transports becomes straightforward — the SDK providesa2agrpc.NewHandleranda2asrv.NewRESTHandlerout of the box. Both share the sameRequestHandlerandScionExecutorfrom #362.Configuration (opt-in)
Non-goals
Risk
a2a-goSDK.