diff --git a/CHANGELOG.md b/CHANGELOG.md index 25f9b8f..649e929 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ All notable changes to mcp-audit are documented in this file. - Terminal audit `outcome` values and UUIDv7 `audit_operation_id` correlation, with an exactly-once operation finalizer shared by HTTP and stdio transports. +- Complete HTTP terminal auditing for upstream connection failures, timeouts, + malformed and incomplete responses, SSE failures, cancellations, and client + disconnects. HTTP notifications are finalized only after the upstream result. - Security invariants and release gates for v1.2.0. - Installation cookbook with platform-specific notes in `INSTALL.md`. - VS Code stdio configuration example under `examples/vscode/`. diff --git a/internal/proxy/http.go b/internal/proxy/http.go index 607b16c..1001150 100644 --- a/internal/proxy/http.go +++ b/internal/proxy/http.go @@ -5,6 +5,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "log/slog" @@ -154,6 +155,7 @@ func (p *HTTPProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { pending, reject := p.observeHTTPRequest(body, startedAt) if reject != nil { + p.finalizeHTTPPending(pending, audit.OutcomeCancelled, audit.DirectionClientToServer) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write(reject) @@ -162,6 +164,7 @@ func (p *HTTPProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { resp, err := p.doUpstreamRequest(r, body) if err != nil { + p.finalizeHTTPPending(pending, httpFailureOutcome(r.Context(), err), audit.DirectionServerToClient) http.Error(w, "upstream request failed", http.StatusBadGateway) p.log.Error("upstream request failed", "error", err) return @@ -171,19 +174,25 @@ func (p *HTTPProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { copyHeader(w.Header(), resp.Header) if isEventStream(resp.Header.Get("Content-Type")) { w.WriteHeader(resp.StatusCode) - p.streamSSE(w, resp.Body, pending) + p.streamSSE(r.Context(), w, resp.Body, pending, resp.StatusCode) return } respBody, err := io.ReadAll(resp.Body) if err != nil { + p.finalizeHTTPPending(pending, httpFailureOutcome(r.Context(), err), audit.DirectionServerToClient) http.Error(w, "failed to read upstream response", http.StatusBadGateway) p.log.Error("failed to read upstream response", "error", err) return } w.WriteHeader(resp.StatusCode) - _, _ = w.Write(respBody) - p.observeHTTPResponse(respBody, pending) + if _, err := w.Write(respBody); err != nil { + p.finalizeHTTPPending(pending, audit.OutcomeClientDisconnect, audit.DirectionServerToClient) + p.log.Warn("failed to write response to client", "error", err) + return + } + p.observeHTTPResponse(respBody, pending, resp.StatusCode) + p.finalizeHTTPRemainder(pending, resp.StatusCode) } func (p *HTTPProxy) doUpstreamRequest(r *http.Request, body []byte) (*http.Response, error) { @@ -309,7 +318,7 @@ func (p *HTTPProxy) observeHTTPRequest(raw []byte, startedAt time.Time) (map[str continue } toolName := toolNameFromParams(msg.Method, msg.Params) - call := p.newPendingCall(msg.Method, jsonRPCID(msg.ID), toolName, msg.Params, startedAt) + call := p.newPendingCall(msg.Method, jsonRPCID(msg.ID), toolName, msg.Params, len(msg.ID) > 0, startedAt) if msg.Method == "tools/call" { decision := p.evaluatePolicy(toolName) p.recordPolicyDecision(decision) @@ -335,48 +344,60 @@ func (p *HTTPProxy) observeHTTPRequest(raw []byte, startedAt time.Time) (map[str pending[string(msg.ID)] = call continue } - if err := p.record(call, audit.OutcomeSuccess, audit.DirectionClientToServer, nil, nil); err != nil { - p.log.Error("failed to audit http notification", "error", err) + if call.operation != nil { + pending["notification:"+call.operation.ID()] = call } } return pending, nil } -func (p *HTTPProxy) observeHTTPResponse(raw []byte, pending map[string]pendingCall) { +func (p *HTTPProxy) observeHTTPResponse(raw []byte, pending map[string]pendingCall, statusCode int) bool { if len(pending) == 0 || len(bytes.TrimSpace(raw)) == 0 { - return + return true } messages, err := decodeMessages(raw) if err != nil { p.log.Warn("failed to inspect http response", "error", err) - return + return false } for _, msg := range messages { call, ok := pending[string(msg.ID)] if !ok { continue } - if err := p.record(call, outcomeForRPCError(msg.Error), audit.DirectionServerToClient, msg.Result, msg.Error); err != nil { + outcome := outcomeForRPCError(msg.Error) + if statusCode >= http.StatusBadRequest { + outcome = audit.OutcomeUpstreamError + } + if err := p.record(call, outcome, audit.DirectionServerToClient, msg.Result, msg.Error); err != nil { p.log.Error("failed to audit http response", "error", err) } delete(pending, string(msg.ID)) } + return true } -func (p *HTTPProxy) streamSSE(w http.ResponseWriter, body io.Reader, pending map[string]pendingCall) { +func (p *HTTPProxy) streamSSE(ctx context.Context, w http.ResponseWriter, body io.Reader, pending map[string]pendingCall, statusCode int) { flusher, _ := w.(http.Flusher) reader := bufio.NewReader(body) var data strings.Builder for { lineBytes, err := reader.ReadBytes('\n') if len(lineBytes) > 0 { - _, _ = w.Write(lineBytes) + if _, writeErr := w.Write(lineBytes); writeErr != nil { + p.finalizeHTTPPending(pending, audit.OutcomeClientDisconnect, audit.DirectionServerToClient) + p.log.Warn("failed to stream response to client", "error", writeErr) + return + } } if flusher != nil { flusher.Flush() } if len(lineBytes) == 0 && err != nil { - if err != io.EOF { + if err == io.EOF { + p.finalizeHTTPRemainder(pending, statusCode) + } else { + p.finalizeHTTPPending(pending, httpFailureOutcome(ctx, err), audit.DirectionServerToClient) p.log.Error("failed to stream SSE response", "error", err) } return @@ -386,11 +407,16 @@ func (p *HTTPProxy) streamSSE(w http.ResponseWriter, body io.Reader, pending map data.WriteString(strings.TrimSpace(strings.TrimPrefix(line, "data:"))) } if line == "" && data.Len() > 0 { - p.observeHTTPResponse([]byte(data.String()), pending) + if !p.observeHTTPResponse([]byte(data.String()), pending, statusCode) { + p.finalizeHTTPRemainder(pending, statusCode) + } data.Reset() } if err != nil { - if err != io.EOF { + if err == io.EOF { + p.finalizeHTTPRemainder(pending, statusCode) + } else { + p.finalizeHTTPPending(pending, httpFailureOutcome(ctx, err), audit.DirectionServerToClient) p.log.Error("failed to stream SSE response", "error", err) } return @@ -398,7 +424,46 @@ func (p *HTTPProxy) streamSSE(w http.ResponseWriter, body io.Reader, pending map } } -func (p *HTTPProxy) newPendingCall(method, requestID, toolName string, params json.RawMessage, startedAt time.Time) pendingCall { +func (p *HTTPProxy) finalizeHTTPRemainder(pending map[string]pendingCall, statusCode int) { + for id, call := range pending { + outcome := audit.OutcomeMalformedUpstreamResponse + direction := audit.DirectionServerToClient + if statusCode >= http.StatusBadRequest { + outcome = audit.OutcomeUpstreamError + } else if !call.expectsResponse { + outcome = audit.OutcomeSuccess + direction = audit.DirectionClientToServer + } + if err := p.record(call, outcome, direction, nil, nil); err != nil { + p.log.Error("failed to finalize http audit operation", "outcome", outcome, "error", err) + } + delete(pending, id) + } +} + +func (p *HTTPProxy) finalizeHTTPPending(pending map[string]pendingCall, outcome audit.Outcome, direction string) { + for id, call := range pending { + if err := p.record(call, outcome, direction, nil, nil); err != nil { + p.log.Error("failed to finalize incomplete http audit operation", "outcome", outcome, "error", err) + } + delete(pending, id) + } +} + +func httpFailureOutcome(ctx context.Context, err error) audit.Outcome { + if errors.Is(err, context.DeadlineExceeded) { + return audit.OutcomeTimeout + } + if errors.Is(err, context.Canceled) { + if ctx != nil && errors.Is(ctx.Err(), context.Canceled) { + return audit.OutcomeClientDisconnect + } + return audit.OutcomeCancelled + } + return audit.OutcomeUpstreamError +} + +func (p *HTTPProxy) newPendingCall(method, requestID, toolName string, params json.RawMessage, expectsResponse bool, startedAt time.Time) pendingCall { operation, err := audit.NewOperation(p.config.Audit, audit.Entry{ Method: method, RequestID: requestID, @@ -410,7 +475,7 @@ func (p *HTTPProxy) newPendingCall(method, requestID, toolName string, params js if err != nil { p.log.Error("failed to start audit operation", "method", method, "error", err) } - return pendingCall{operation: operation, startedAt: startedAt} + return pendingCall{operation: operation, startedAt: startedAt, expectsResponse: expectsResponse} } func (p *HTTPProxy) record(call pendingCall, outcome audit.Outcome, direction string, result json.RawMessage, rpcErr *audit.RPCError) error { diff --git a/internal/proxy/http_test.go b/internal/proxy/http_test.go index b66636e..2b02fdd 100644 --- a/internal/proxy/http_test.go +++ b/internal/proxy/http_test.go @@ -2,11 +2,13 @@ package proxy import ( "bytes" + "context" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" + "strings" "testing" "time" @@ -241,16 +243,21 @@ func TestHTTPProxyAuditRedactsSensitiveJSONRPCParams(t *testing.T) { // TestHTTPProxyTimesOutUpstreamRequests verifies slow upstream requests use the // existing bad-gateway error path instead of hanging indefinitely. func TestHTTPProxyTimesOutUpstreamRequests(t *testing.T) { + store := &memoryAuditStore{} proxy, err := NewHTTPProxy(HTTPConfig{ Upstream: "http://upstream.local", UpstreamTimeoutMS: 10, + Audit: audit.NewLogger(audit.LoggerConfig{ + Store: store, + Transport: "http", + }), }) if err != nil { t.Fatalf("new http proxy: %v", err) } proxy.client.Transport = blockingRoundTripper{} - req := httptest.NewRequest(http.MethodPost, "http://proxy.local/rpc", nil) + req := httptest.NewRequest(http.MethodPost, "http://proxy.local/rpc", bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"resources/list"}`))) rec := httptest.NewRecorder() proxy.ServeHTTP(rec, req) @@ -258,6 +265,145 @@ func TestHTTPProxyTimesOutUpstreamRequests(t *testing.T) { if rec.Code != http.StatusBadGateway { t.Fatalf("status = %d, want %d", rec.Code, http.StatusBadGateway) } + if len(store.entries) != 1 || store.entries[0].Outcome != audit.OutcomeTimeout { + t.Fatalf("entries = %#v, want one timeout", store.entries) + } +} + +func TestHTTPProxyAuditsUpstreamConnectionFailure(t *testing.T) { + store := &memoryAuditStore{} + proxy, err := NewHTTPProxy(HTTPConfig{ + Upstream: "http://upstream.local", + Audit: audit.NewLogger(audit.LoggerConfig{Store: store, Transport: "http"}), + }) + if err != nil { + t.Fatalf("new http proxy: %v", err) + } + proxy.client.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, fmt.Errorf("connection refused") + }) + + req := httptest.NewRequest(http.MethodPost, "http://proxy.local/rpc", bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"resources/list"}`))) + rec := httptest.NewRecorder() + proxy.ServeHTTP(rec, req) + + if len(store.entries) != 1 || store.entries[0].Outcome != audit.OutcomeUpstreamError { + t.Fatalf("entries = %#v, want one upstream_error", store.entries) + } +} + +func TestHTTPProxyDoesNotMarkFailedNotificationSuccessful(t *testing.T) { + store := &memoryAuditStore{} + proxy, err := NewHTTPProxy(HTTPConfig{ + Upstream: "http://upstream.local", + Audit: audit.NewLogger(audit.LoggerConfig{Store: store, Transport: "http"}), + }) + if err != nil { + t.Fatalf("new http proxy: %v", err) + } + proxy.client.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, fmt.Errorf("connection refused") + }) + + req := httptest.NewRequest(http.MethodPost, "http://proxy.local/rpc", bytes.NewReader([]byte(`{"jsonrpc":"2.0","method":"notifications/initialized"}`))) + proxy.ServeHTTP(httptest.NewRecorder(), req) + + if len(store.entries) != 1 || store.entries[0].Outcome != audit.OutcomeUpstreamError { + t.Fatalf("entries = %#v, want one upstream_error", store.entries) + } +} + +func TestHTTPProxyAuditsMalformedUpstreamResponse(t *testing.T) { + store := &memoryAuditStore{} + proxy, err := NewHTTPProxy(HTTPConfig{ + Upstream: "http://upstream.local", + Audit: audit.NewLogger(audit.LoggerConfig{Store: store, Transport: "http"}), + }) + if err != nil { + t.Fatalf("new http proxy: %v", err) + } + proxy.client.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("not-json")), + Header: make(http.Header), + }, nil + }) + + req := httptest.NewRequest(http.MethodPost, "http://proxy.local/rpc", bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"resources/list"}`))) + rec := httptest.NewRecorder() + proxy.ServeHTTP(rec, req) + + if len(store.entries) != 1 || store.entries[0].Outcome != audit.OutcomeMalformedUpstreamResponse { + t.Fatalf("entries = %#v, want one malformed_upstream_response", store.entries) + } +} + +func TestHTTPProxyAuditsClientCancellation(t *testing.T) { + store := &memoryAuditStore{} + proxy, err := NewHTTPProxy(HTTPConfig{ + Upstream: "http://upstream.local", + Audit: audit.NewLogger(audit.LoggerConfig{Store: store, Transport: "http"}), + }) + if err != nil { + t.Fatalf("new http proxy: %v", err) + } + proxy.client.Transport = blockingRoundTripper{} + ctx, cancel := context.WithCancel(context.Background()) + cancel() + req := httptest.NewRequest(http.MethodPost, "http://proxy.local/rpc", bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"resources/list"}`))).WithContext(ctx) + + proxy.ServeHTTP(httptest.NewRecorder(), req) + + if len(store.entries) != 1 || store.entries[0].Outcome != audit.OutcomeClientDisconnect { + t.Fatalf("entries = %#v, want one client_disconnect", store.entries) + } +} + +func TestHTTPProxyAuditsClientWriteFailure(t *testing.T) { + store := &memoryAuditStore{} + proxy, err := NewHTTPProxy(HTTPConfig{ + Upstream: "http://upstream.local", + Audit: audit.NewLogger(audit.LoggerConfig{Store: store, Transport: "http"}), + }) + if err != nil { + t.Fatalf("new http proxy: %v", err) + } + proxy.client.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) { + return okJSONResponse(), nil + }) + req := httptest.NewRequest(http.MethodPost, "http://proxy.local/rpc", bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"resources/list"}`))) + + proxy.ServeHTTP(newFailingResponseWriter(), req) + + if len(store.entries) != 1 || store.entries[0].Outcome != audit.OutcomeClientDisconnect { + t.Fatalf("entries = %#v, want one client_disconnect", store.entries) + } +} + +func TestHTTPProxyAuditsIncompleteSSEResponse(t *testing.T) { + store := &memoryAuditStore{} + proxy, err := NewHTTPProxy(HTTPConfig{ + Upstream: "http://upstream.local", + Audit: audit.NewLogger(audit.LoggerConfig{Store: store, Transport: "http"}), + }) + if err != nil { + t.Fatalf("new http proxy: %v", err) + } + proxy.client.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("data: not-json\n\n")), + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + }, nil + }) + req := httptest.NewRequest(http.MethodPost, "http://proxy.local/rpc", bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"resources/list"}`))) + + proxy.ServeHTTP(httptest.NewRecorder(), req) + + if len(store.entries) != 1 || store.entries[0].Outcome != audit.OutcomeMalformedUpstreamResponse { + t.Fatalf("entries = %#v, want one malformed_upstream_response", store.entries) + } } func TestHTTPProxyRetriesSafeJSONRPCMethodOnServiceUnavailable(t *testing.T) { @@ -455,6 +601,22 @@ func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } +type failingResponseWriter struct { + header http.Header +} + +func newFailingResponseWriter() *failingResponseWriter { + return &failingResponseWriter{header: make(http.Header)} +} + +func (w *failingResponseWriter) Header() http.Header { return w.header } + +func (*failingResponseWriter) Write([]byte) (int, error) { + return 0, fmt.Errorf("client disconnected") +} + +func (*failingResponseWriter) WriteHeader(int) {} + func okJSONResponse() *http.Response { return &http.Response{ StatusCode: http.StatusOK, diff --git a/internal/proxy/stdio.go b/internal/proxy/stdio.go index 2669064..d4d6dbf 100644 --- a/internal/proxy/stdio.go +++ b/internal/proxy/stdio.go @@ -335,8 +335,9 @@ func (p *StdioProxy) recordPolicyDecision(decision policy.Decision) { } type pendingCall struct { - operation *audit.Operation - startedAt time.Time + operation *audit.Operation + startedAt time.Time + expectsResponse bool } type rpcState struct {