Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 47 additions & 2 deletions server/internal/notifications/a2a.go
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,9 @@ const (
// a2aMaxErrorBody caps how much of an error response is kept in the
// error message.
a2aMaxErrorBody = 512
// a2aMaxResponseBody caps every successful A2A response body (card,
// SendMessage, GetTask). Plenty for a card or a task with artifacts.
a2aMaxResponseBody = 1 << 20
// a2aLegacyMajor is the semver major of A2A 0.x (0.3), served through
// the SDK's a2av0 compat codec.
a2aLegacyMajor = "v0"
Expand Down Expand Up @@ -98,6 +101,9 @@ var (
ErrA2AServerError = errors.New("a2a server error")
// ErrA2ARequestFailed flags a 4xx or a JSON-RPC error. Permanent.
ErrA2ARequestFailed = errors.New("a2a request failed")
// ErrA2AResponseTooLarge flags a response body over a2aMaxResponseBody.
// Permanent: the same agent will answer the same way.
ErrA2AResponseTooLarge = errors.New("a2a agent response exceeds 1 MiB")
)

// A2ASender sends incident events to an AI agent over A2A.
Expand Down Expand Up @@ -196,8 +202,10 @@ func (t *a2aStatusTransport) RoundTrip(req *http.Request) (*http.Response, error
}

// 3xx goes back to the client, which follows it through the same
// guarded transport.
// guarded transport. The SDK reads the body whole, so it is capped here.
if resp.StatusCode < http.StatusBadRequest {
resp.Body = &a2aCappedBody{ReadCloser: resp.Body, remaining: a2aMaxResponseBody}

return resp, nil
}

Expand All @@ -207,6 +215,38 @@ func (t *a2aStatusTransport) RoundTrip(req *http.Request) (*http.Response, error
return nil, &a2aHTTPStatusError{StatusCode: resp.StatusCode, Body: strings.TrimSpace(string(body))}
}

// a2aCappedBody returns ErrA2AResponseTooLarge, never EOF, once the body goes
// past its cap, so a truncated card or JSON-RPC response fails instead of
// parsing.
type a2aCappedBody struct {
io.ReadCloser

remaining int64
}

func (b *a2aCappedBody) Read(buf []byte) (int, error) {
if b.remaining <= 0 {
// At the cap: one more byte means the body is too large.
var probe [1]byte

n, err := b.ReadCloser.Read(probe[:])
if n > 0 {
return 0, ErrA2AResponseTooLarge
}

return 0, err
}

if int64(len(buf)) > b.remaining {
buf = buf[:b.remaining]
}

n, err := b.ReadCloser.Read(buf)
b.remaining -= int64(n)

return n, err
}

// a2aHTTPClient is the guarded client every A2A request goes through: card
// fetch, SendMessage and GetTask. The guard checks every dial, so a redirect
// or a DNS answer to a private address is refused too.
Expand Down Expand Up @@ -243,6 +283,10 @@ func classifyA2AError(err error) error {
return fmt.Errorf("%w: %s", ErrSenderURLInvalid, denied.Error())
}

if errors.Is(err, ErrA2AResponseTooLarge) {
return ErrA2AResponseTooLarge
}

var rpcErr *a2a.Error
if errors.As(err, &rpcErr) {
return fmt.Errorf("%w: %s", ErrA2ARequestFailed, rpcErr.Error())
Expand Down Expand Up @@ -347,7 +391,8 @@ func fetchA2ACard(
}

var statusErr *a2aHTTPStatusError
if errors.As(err, &statusErr) || IsNetworkError(err) || errors.Is(err, context.DeadlineExceeded) {
if errors.As(err, &statusErr) || IsNetworkError(err) || errors.Is(err, context.DeadlineExceeded) ||
errors.Is(err, ErrA2AResponseTooLarge) {
return nil, classifyA2AError(err)
}

Expand Down
113 changes: 113 additions & 0 deletions server/internal/notifications/a2a_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"

Expand Down Expand Up @@ -713,3 +714,115 @@ func TestA2ATaskComment(t *testing.T) {
}}}
r.Equal("root cause: expired cert", A2ATaskComment(done))
}

// streamOversized answers with a JSON prefix and then padding inside a string,
// far past a2aMaxResponseBody, and records how many bytes it managed to write.
func streamOversized(prefix string, written *atomic.Int64) http.HandlerFunc {
return func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)

n, err := io.WriteString(w, prefix)
written.Add(int64(n))

if err != nil {
return
}

chunk := []byte(strings.Repeat("a", 32<<10))
for range (64 << 20) / len(chunk) {
n, err := w.Write(chunk)
written.Add(int64(n))

if err != nil {
return
}
}
}
}

// a2aOversizeMargin is how far past the cap the server may get: what sits in
// the loopback socket buffers when the client stops reading.
const a2aOversizeMargin = 8 << 20

func TestA2A_CardTooLarge(t *testing.T) {
t.Parallel()
r := require.New(t)

var written atomic.Int64

srv := httptest.NewServer(streamOversized(`{"name":"`, &written))
t.Cleanup(srv.Close)

_, err := ResolveA2AAgent(t.Context(), nil, models.JSONMap{A2ASettingAgentCardURL: srv.URL})
r.ErrorIs(err, ErrA2AResponseTooLarge)
r.Contains(err.Error(), "exceeds 1 MiB")
requirePermanentA2A(r, classifyA2AError(err))

srv.Close() // waits for the handler, so the count is final
t.Logf("server wrote %d bytes", written.Load())
r.Less(written.Load(), int64(a2aMaxResponseBody+a2aOversizeMargin))
}

func TestA2ASender_SendResultTooLarge(t *testing.T) {
t.Parallel()
r := require.New(t)

var written atomic.Int64

oversized := streamOversized(`{"jsonrpc":"2.0","id":1,"result":{"message":{"messageId":"m","role":"ROLE_AGENT",`+
`"parts":[{"text":"`, &written)
mux := http.NewServeMux()
srv := httptest.NewServer(mux)
t.Cleanup(srv.Close)

mux.HandleFunc("GET "+a2aWellKnownCardPath, func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(w, a2aCardV1(srv.URL))
})
mux.HandleFunc("POST /rpc", oversized)

h := newA2AHarness(nil)
payload := a2aTestPayload(models.JSONMap{A2ASettingAgentCardURL: srv.URL}, eventTypeIncidentCreated)

err := (&A2ASender{}).Send(t.Context(), h.jctx, payload)
requirePermanentA2A(r, err)
r.ErrorIs(err, ErrA2AResponseTooLarge)
r.Empty(h.comments.all())

srv.Close()
t.Logf("server wrote %d bytes", written.Load())
r.Less(written.Load(), int64(a2aMaxResponseBody+a2aOversizeMargin))
}

// Positive control for the two tests above: a card and an answer just under
// the cap still go through.
func TestA2ASender_ResponseJustUnderTheCap(t *testing.T) {
t.Parallel()
r := require.New(t)

text := strings.Repeat("x", a2aMaxResponseBody-1024)
agent := newFakeA2AAgent(t, func(agent *fakeA2AAgent) {
agent.card = func(base string) string {
card := a2aCardV1(base)

return strings.Replace(card, `"Investigates incidents"`,
`"`+strings.Repeat("d", a2aMaxResponseBody-len(card)-64)+`"`, 1)
}
agent.result = func(fakeA2ACall) any {
return map[string]any{"message": map[string]any{
"messageId": "agent-reply-big", "role": "ROLE_AGENT",
"parts": []any{map[string]any{"text": text}},
}}
}
})

info, err := ResolveA2AAgent(t.Context(), nil, models.JSONMap{A2ASettingAgentCardURL: agent.base()})
r.NoError(err)
r.Equal("acme investigator", info.Name)

h := newA2AHarness(nil)
payload := a2aTestPayload(models.JSONMap{A2ASettingAgentCardURL: agent.base()}, eventTypeIncidentCreated)
r.NoError((&A2ASender{}).Send(t.Context(), h.jctx, payload))
r.Equal([]string{text}, h.comments.all())
}
44 changes: 44 additions & 0 deletions specs/done/2026/10/2026-10-19-01-a2a-response-size-cap.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
---
model: opus
effort: medium
---

# A2A client reads Agent Card and JSON-RPC responses with no size cap: a user's own agent can exhaust the server's memory

## Problem

Found by the security scan of 2026-10-19 (finding J1). Severity: medium. Any member with write access to an org (role `user` or above) can trigger it, since notification integrations are self-service. On SaaS that is anyone: any account can create its own org.

- The A2A channel (spec `2026-10-18-02`, `server/internal/notifications/a2a.go`) talks to a URL the user picks, through the `a2a-go/v2` v2.6.0 SDK. The SDK reads every response body whole, with no size limit:
- Agent Card: `agentcard.Resolver` (`a2a.go:340`) ends in the SDK's `fetchCard`, which does `io.ReadAll(resp.Body)` (`a2aclient/agentcard/resolver.go:173` in the module).
- `SendMessage` (`a2a.go:494`) and `GetTask` (`a2a.go:849`): the SDK's `jsonrpcTransport.sendRequest` decodes the body with `json.NewDecoder(httpResp.Body).Decode(&resp)` (`a2aclient/jsonrpc.go:129`). `resp.Result` is a `json.RawMessage`, so the whole result is buffered. The `a2av0` compat client does the same (`a2acompat/a2av0/jsonrpc_client.go:128`).
- On the SolidPing side, nothing limits the body either:
- `a2aHTTPClient` (`a2a.go:213`) is the guarded client plus `a2aStatusTransport` (`a2a.go:192`). The transport caps only error bodies (`a2aMaxErrorBody`, 512 bytes, `a2a.go:58`). A `200` body goes to the SDK untouched.
- The only bound is the 30 s client timeout (`a2aTimeout`, `a2a.go:52`). It covers reading the body, but a fast server can send several GB in 30 s.
- Every other sender caps what it reads: `webhook.go:264` (`models.DeliveryDetailsBodyCap`), `discord_webhook.go:139`, and `checkhttp` (`checker.go:31`, 10 MB).
- The egress guard does not help: the attacker's server is a public host.
- Reachable paths:
- `POST /orgs/:org/integrations/a2a/resolve-card` (`handlers/integrations/a2a.go:38`) and `POST /orgs/:org/integrations/:uid/test` (`testA2AConnection`, `a2a.go:90`). Both run synchronously on the API node, with no rate limit, and need nothing saved beyond the form values.
- The notification job (`A2ASender.Send`) and the `a2a_task_poll` chain, which runs every 60 s for up to 30 min per task, on the job workers.
- Impact: a server that answers the card request with an endless `200` JSON body makes the process buffer it until the timeout. A few parallel calls are enough to OOM-kill a pod. When the API and the workers share a process (the single-binary deployment), checks and alerts stop for every org until the restart, and the attacker can repeat it at will.

## Proposal

1. In `server/internal/notifications/a2a.go`, cap every A2A response body inside `a2aStatusTransport.RoundTrip`. For a status below 400, wrap `resp.Body` in a reader that returns an error after a byte cap. The error must not look like EOF: a truncated card has to fail, not parse.
- Add a const `a2aMaxResponseBody`: 1 MiB is plenty for a card or a task with artifacts.
- Since every request (card, `SendMessage`, `GetTask`, both codecs) goes through `a2aHTTPClient`, this one change covers all paths. The SDK cannot bypass it: `WithDefaultsDisabled` plus the two explicit JSON-RPC transports leave no other client in use.
2. Classify the "response too large" error as permanent in `classifyA2AError` (a dedicated sentinel, e.g. `ErrA2AResponseTooLarge`). Retrying the same oversized agent is pointless.
3. Check that `resolve-card` / test return a readable error ("agent response exceeds 1 MiB") instead of a JSON parse error.

## Tests

`server/internal/notifications/a2a_test.go`, next to `TestA2ASender_BadCardsArePermanent`:
- Card too large: an `httptest` server answers the card URL with a `200`, `Content-Type: application/json`, and a body larger than the cap (a valid JSON prefix followed by padding, streamed). `ResolveA2AAgent` must fail with `ErrA2AResponseTooLarge`, and the handler must have read at most cap + a small margin. Count the bytes written by the server. This test fails before the fix: the read succeeds or fails only on parsing, after reading everything.
- `SendMessage` result too large: a valid card, then a JSON-RPC endpoint whose `200` response is larger than the cap. `Send` must return a permanent (non-retryable) error.
- Positive control: a card and a `SendMessage` answer just under the cap (e.g. a task whose artifact text is close to 1 MiB) still go through. The comment is recorded, truncated to `MaxCommentLength` as today.
- Keep `TestA2ASender_ErrorClassification` green. Error bodies stay capped at 512 bytes.

## To verify

- Whether `json.Decoder` reports the capped reader's error, rather than `unexpected EOF`, so that classification can use `errors.Is`. If not, have the reader record that it hit the cap and check that flag.
- Whether a prod API pod also runs the job workers (single binary). That decides whether an OOM stops monitoring or only the API.
Loading