diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 85a40f9..e283083 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,6 +28,9 @@ jobs: - name: Test run: go test -race -coverprofile=coverage.out -covermode=atomic ./... + - name: End-to-end integration + run: go test -tags=integration ./tests/integration/... + - name: Upload coverage to Codecov uses: codecov/codecov-action@v5 with: diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 7115ca0..8e3d9a0 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -11,6 +11,13 @@ go build ./... go test ./... ``` +The full-binary suite is opt-in because it compiles subprocess fixtures and +opens local ports. Run it with: + +```bash +go test -tags=integration ./tests/integration/... +``` + ## Before opening a PR - `go build ./...` passes diff --git a/tests/integration/http_test.go b/tests/integration/http_test.go new file mode 100644 index 0000000..36562af --- /dev/null +++ b/tests/integration/http_test.go @@ -0,0 +1,396 @@ +//go:build integration + +package integration_test + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/rsa" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/base64" + "encoding/json" + "encoding/pem" + "fmt" + "io" + "math/big" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/P4ST4S/mcp-audit/internal/audit" + "github.com/golang-jwt/jwt/v5" +) + +const staticBearerToken = "abcdef0123456789abcdef0123456789" + +func TestHTTPBinarySecurityGatewayFlow(t *testing.T) { + var upstreamCalls atomic.Int32 + var forwardedRevision atomic.Value + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + upstreamCalls.Add(1) + forwardedRevision.Store(request.Header.Get("Mcp-Protocol-Version")) + body, _ := io.ReadAll(request.Body) + var envelope struct { + ID json.RawMessage `json:"id"` + } + _ = json.Unmarshal(body, &envelope) + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":{"ok":true}}`, envelope.ID) + })) + defer upstream.Close() + + directory := t.TempDir() + auditPath := filepath.Join(directory, "audit.jsonl") + port := freePort(t) + config := fmt.Sprintf(`proxy: + transport: http + upstream: %q + bind_address: 127.0.0.1 + port: %d + client_id: legacy-client + server_id: integration-upstream + http: + max_request_body_bytes: 512 +auth: + mode: static_bearer + static: + subject: alice + client_id: authenticated-client + issuer: integration + roles: [operator] + scopes: [mcp:read] +audit: + storage: jsonl + path: %q + sign: true +policy: + enabled: true + default_action: allow + scope: all_operations + rules: + - action: deny + role: operator + method: resources/read + name: file:///secret +dashboard: + enabled: false +metrics: + enabled: false +`, upstream.URL, port, auditPath) + command, stderr := startHTTPProxy(t, writeConfig(t, config), port, map[string]string{ + "MCP_AUDIT_STATIC_BEARER_TOKEN": staticBearerToken, + }) + baseURL := fmt.Sprintf("http://127.0.0.1:%d/mcp", port) + + unauthorized := postMCP(t, http.DefaultClient, baseURL, `{"jsonrpc":"2.0","id":1,"method":"tools/list"}`, nil) + if unauthorized.StatusCode != http.StatusUnauthorized { + t.Fatalf("unauthorized status = %d", unauthorized.StatusCode) + } + _ = unauthorized.Body.Close() + + inconsistent := postMCP(t, http.DefaultClient, baseURL, `{"jsonrpc":"2.0","id":2,"method":"tools/list"}`, map[string]string{ + "Authorization": "Bearer " + staticBearerToken, + "Mcp-Method": "resources/read", + "Mcp-Protocol-Version": "2026-07-28", + }) + if inconsistent.StatusCode != http.StatusBadRequest { + t.Fatalf("inconsistent metadata status = %d", inconsistent.StatusCode) + } + _ = inconsistent.Body.Close() + + denied := postMCP(t, http.DefaultClient, baseURL, `{"jsonrpc":"2.0","id":3,"method":"resources/read","params":{"uri":"file:///secret"}}`, map[string]string{ + "Authorization": "Bearer " + staticBearerToken, + }) + var deniedBody struct { + Error *audit.RPCError `json:"error"` + } + decodeHTTPJSON(t, denied, &deniedBody) + if deniedBody.Error == nil || deniedBody.Error.Code != -32030 { + t.Fatalf("policy response = %#v", deniedBody) + } + + accepted := postMCP(t, http.DefaultClient, baseURL, `{"jsonrpc":"2.0","id":4,"method":"tools/list","params":{"_meta":{"protocolRevision":"2026-07-28"}}}`, map[string]string{ + "Authorization": "Bearer " + staticBearerToken, + "Mcp-Method": "tools/list", + "Mcp-Protocol-Version": "2026-07-28", + }) + var acceptedBody struct { + Result map[string]any `json:"result"` + } + decodeHTTPJSON(t, accepted, &acceptedBody) + if acceptedBody.Result["ok"] != true || forwardedRevision.Load() != "2026-07-28" { + t.Fatalf("accepted response/revision = %#v/%v", acceptedBody, forwardedRevision.Load()) + } + + oversized := postMCP(t, http.DefaultClient, baseURL, `{"jsonrpc":"2.0","id":5,"method":"tools/call","params":{"name":"echo","arguments":{"payload":"`+strings.Repeat("x", 600)+`"}}}`, map[string]string{ + "Authorization": "Bearer " + staticBearerToken, + }) + if oversized.StatusCode != http.StatusRequestEntityTooLarge { + t.Fatalf("oversized status = %d", oversized.StatusCode) + } + _ = oversized.Body.Close() + + stopProcess(t, command, stderr) + if upstreamCalls.Load() != 1 { + t.Fatalf("upstream calls = %d, want 1", upstreamCalls.Load()) + } + entries := readAuditEntries(t, auditPath) + if len(entries) != 2 { + t.Fatalf("audit entries = %#v, want deny and success", entries) + } + if entries[0].Outcome != audit.OutcomeDenied || entries[0].Principal == nil || entries[0].Principal.Subject != "alice" { + t.Fatalf("denied principal audit = %#v", entries[0]) + } + if entries[1].Outcome != audit.OutcomeSuccess || entries[1].Integrity == nil { + t.Fatalf("accepted audit = %#v", entries[1]) + } +} + +func TestHTTPBinaryIncomingTLS(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"jsonrpc":"2.0","id":1,"result":{}}`) + })) + defer upstream.Close() + certFile, keyFile := writeServerCertificate(t) + port := freePort(t) + config := fmt.Sprintf(`proxy: + transport: http + upstream: %q + bind_address: 127.0.0.1 + port: %d + tls: + enabled: true + cert_file: %q + key_file: %q +audit: + path: %q + sign: true +dashboard: + enabled: false +metrics: + enabled: false +`, upstream.URL, port, certFile, keyFile, filepath.Join(t.TempDir(), "audit.jsonl")) + command, stderr := startHTTPProxy(t, writeConfig(t, config), port, nil) + + client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: true}}} //nolint:gosec -- test certificate + response := postMCP(t, client, fmt.Sprintf("https://127.0.0.1:%d/mcp", port), `{"jsonrpc":"2.0","id":1,"method":"tools/list"}`, nil) + if response.StatusCode != http.StatusOK || response.TLS == nil || response.TLS.Version < tls.VersionTLS12 { + t.Fatalf("TLS status/state = %d/%#v", response.StatusCode, response.TLS) + } + _ = response.Body.Close() + stopProcess(t, command, stderr) +} + +func TestHTTPBinaryOIDCPrincipalReachesPolicy(t *testing.T) { + privateKey, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatalf("generate RSA key: %v", err) + } + const keyID = "integration-key" + var issuerURL string + identityProvider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "keys": []map[string]any{{ + "kty": "RSA", + "kid": keyID, + "use": "sig", + "alg": "RS256", + "n": base64.RawURLEncoding.EncodeToString(privateKey.PublicKey.N.Bytes()), + "e": base64.RawURLEncoding.EncodeToString(big.NewInt(int64(privateKey.PublicKey.E)).Bytes()), + }}, + }) + })) + defer identityProvider.Close() + issuerURL = identityProvider.URL + + upstreamCalls := atomic.Int32{} + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstreamCalls.Add(1) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"jsonrpc":"2.0","id":1,"result":{}}`) + })) + defer upstream.Close() + + auditPath := filepath.Join(t.TempDir(), "audit.jsonl") + port := freePort(t) + config := fmt.Sprintf(`proxy: + transport: http + upstream: %q + bind_address: 127.0.0.1 + port: %d + server_id: integration-upstream +auth: + mode: oidc + oidc: + issuer: %q + audience: mcp-audit + jwks_uri: %q + allowed_methods: [RS256] + client_id_claim: client_id + roles_claim: roles + scopes_claim: scope +audit: + path: %q + sign: true +policy: + enabled: true + default_action: allow + scope: all_operations + rules: + - action: deny + role: auditor + method: resources/read + name: file:///secret +dashboard: + enabled: false +metrics: + enabled: false +`, upstream.URL, port, issuerURL, issuerURL, auditPath) + command, stderr := startHTTPProxy(t, writeConfig(t, config), port, nil) + baseURL := fmt.Sprintf("http://127.0.0.1:%d/mcp", port) + + invalidToken := signJWT(t, privateKey, keyID, issuerURL, "wrong-audience") + invalid := postMCP(t, http.DefaultClient, baseURL, `{"jsonrpc":"2.0","id":1,"method":"tools/list"}`, map[string]string{ + "Authorization": "Bearer " + invalidToken, + }) + if invalid.StatusCode != http.StatusUnauthorized { + t.Fatalf("invalid audience status = %d", invalid.StatusCode) + } + _ = invalid.Body.Close() + + validToken := signJWT(t, privateKey, keyID, issuerURL, "mcp-audit") + denied := postMCP(t, http.DefaultClient, baseURL, `{"jsonrpc":"2.0","id":2,"method":"resources/read","params":{"uri":"file:///secret"}}`, map[string]string{ + "Authorization": "Bearer " + validToken, + }) + var deniedBody struct { + Error *audit.RPCError `json:"error"` + } + decodeHTTPJSON(t, denied, &deniedBody) + if deniedBody.Error == nil || deniedBody.Error.Code != -32030 { + t.Fatalf("OIDC policy response = %#v", deniedBody) + } + + stopProcess(t, command, stderr) + if upstreamCalls.Load() != 0 { + t.Fatalf("upstream calls = %d, want 0", upstreamCalls.Load()) + } + entries := readAuditEntries(t, auditPath) + if len(entries) != 1 || entries[0].Principal == nil || entries[0].Principal.Subject != "bob" || entries[0].Principal.ClientID != "oidc-client" || entries[0].Principal.Issuer != issuerURL { + t.Fatalf("OIDC principal audit = %#v", entries) + } +} + +func signJWT(t *testing.T, privateKey *rsa.PrivateKey, keyID, issuer, audience string) string { + t.Helper() + now := time.Now() + token := jwt.NewWithClaims(jwt.SigningMethodRS256, jwt.MapClaims{ + "iss": issuer, + "aud": audience, + "sub": "bob", + "client_id": "oidc-client", + "roles": []string{"auditor"}, + "scope": "mcp:read", + "iat": now.Unix(), + "nbf": now.Add(-time.Minute).Unix(), + "exp": now.Add(time.Hour).Unix(), + }) + token.Header["kid"] = keyID + signed, err := token.SignedString(privateKey) + if err != nil { + t.Fatalf("sign JWT: %v", err) + } + return signed +} + +func startHTTPProxy(t *testing.T, configPath string, port int, env map[string]string) (*exec.Cmd, *lockedBuffer) { + t.Helper() + if env == nil { + env = make(map[string]string) + } + env["MCP_AUDIT_SIGNING_SECRET"] = signingSecret + command := exec.Command(binaryPath, "--config", configPath, "--log-level", "error") + command.Env = commandEnv(env, "AUDIT_SECRET") + stderr := newLockedBuffer() + command.Stderr = stderr + if err := command.Start(); err != nil { + t.Fatalf("start HTTP proxy: %v", err) + } + t.Cleanup(func() { + if command.Process != nil && command.ProcessState == nil { + _ = command.Process.Kill() + } + }) + waitForPort(t, port, command, stderr) + return command, stderr +} + +func postMCP(t *testing.T, client *http.Client, url, body string, headers map[string]string) *http.Response { + t.Helper() + request, err := http.NewRequest(http.MethodPost, url, strings.NewReader(body)) + if err != nil { + t.Fatalf("create request: %v", err) + } + request.Header.Set("Content-Type", "application/json") + for name, value := range headers { + request.Header.Set(name, value) + } + response, err := client.Do(request) + if err != nil { + t.Fatalf("send request: %v", err) + } + return response +} + +func decodeHTTPJSON(t *testing.T, response *http.Response, target any) { + t.Helper() + defer response.Body.Close() + if err := json.NewDecoder(response.Body).Decode(target); err != nil { + t.Fatalf("decode HTTP JSON: %v", err) + } +} + +func writeServerCertificate(t *testing.T) (string, string) { + t.Helper() + privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("generate private key: %v", err) + } + template := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "localhost"}, + IPAddresses: nil, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + certificateDER, err := x509.CreateCertificate(rand.Reader, &template, &template, &privateKey.PublicKey, privateKey) + if err != nil { + t.Fatalf("create certificate: %v", err) + } + keyDER, err := x509.MarshalECPrivateKey(privateKey) + if err != nil { + t.Fatalf("marshal private key: %v", err) + } + directory := t.TempDir() + certFile := filepath.Join(directory, "server.crt") + keyFile := filepath.Join(directory, "server.key") + if err := os.WriteFile(certFile, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certificateDER}), 0o600); err != nil { + t.Fatalf("write certificate: %v", err) + } + if err := os.WriteFile(keyFile, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600); err != nil { + t.Fatalf("write key: %v", err) + } + return certFile, keyFile +} diff --git a/tests/integration/main_test.go b/tests/integration/main_test.go new file mode 100644 index 0000000..01bf67f --- /dev/null +++ b/tests/integration/main_test.go @@ -0,0 +1,155 @@ +//go:build integration + +package integration_test + +import ( + "bytes" + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +const signingSecret = "0123456789abcdef0123456789abcdef" + +var ( + binaryPath string + upstreamPath string + suiteDir string +) + +func TestMain(m *testing.M) { + var err error + suiteDir, err = os.MkdirTemp("", "mcp-audit-integration-") + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + binaryPath = filepath.Join(suiteDir, "mcp-audit") + upstreamPath = filepath.Join(suiteDir, "stdio-upstream") + if err := buildBinary(binaryPath, "../../cmd/mcp-audit"); err != nil { + fmt.Fprintln(os.Stderr, err) + _ = os.RemoveAll(suiteDir) + os.Exit(1) + } + if err := buildBinary(upstreamPath, "./testdata/upstream"); err != nil { + fmt.Fprintln(os.Stderr, err) + _ = os.RemoveAll(suiteDir) + os.Exit(1) + } + exitCode := m.Run() + _ = os.RemoveAll(suiteDir) + os.Exit(exitCode) +} + +func buildBinary(output, source string) error { + command := exec.Command("go", "build", "-o", output, source) + var stderr bytes.Buffer + command.Stderr = &stderr + if err := command.Run(); err != nil { + return fmt.Errorf("build %s: %w\n%s", source, err, stderr.String()) + } + return nil +} + +func writeConfig(t *testing.T, contents string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(path, []byte(contents), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + return path +} + +func commandEnv(overrides map[string]string, remove ...string) []string { + removed := make(map[string]bool, len(remove)) + for _, name := range remove { + removed[name] = true + } + env := make([]string, 0, len(os.Environ())+len(overrides)) + for _, item := range os.Environ() { + name, _, _ := strings.Cut(item, "=") + if !removed[name] { + env = append(env, item) + } + } + for name, value := range overrides { + env = append(env, name+"="+value) + } + return env +} + +func freePort(t *testing.T) int { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("reserve port: %v", err) + } + defer listener.Close() + return listener.Addr().(*net.TCPAddr).Port +} + +func waitForPort(t *testing.T, port int, process *exec.Cmd, stderr *lockedBuffer) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + address := fmt.Sprintf("127.0.0.1:%d", port) + for time.Now().Before(deadline) { + connection, err := net.DialTimeout("tcp", address, 50*time.Millisecond) + if err == nil { + _ = connection.Close() + return + } + if process.ProcessState != nil && process.ProcessState.Exited() { + t.Fatalf("proxy exited before listening: %s", stderr.String()) + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("proxy did not listen on %s: %s", address, stderr.String()) +} + +func stopProcess(t *testing.T, command *exec.Cmd, stderr *lockedBuffer) { + t.Helper() + if command.Process == nil || command.ProcessState != nil { + return + } + if err := command.Process.Signal(os.Interrupt); err != nil { + t.Fatalf("signal process: %v", err) + } + wait := make(chan error, 1) + go func() { wait <- command.Wait() }() + select { + case err := <-wait: + if err != nil { + t.Fatalf("process shutdown: %v\n%s", err, stderr.String()) + } + case <-time.After(7 * time.Second): + _ = command.Process.Kill() + t.Fatalf("process did not stop after interrupt: %s", stderr.String()) + } +} + +type lockedBuffer struct { + buffer bytes.Buffer + mu sync.Mutex +} + +func newLockedBuffer() *lockedBuffer { + return &lockedBuffer{} +} + +func (b *lockedBuffer) Write(data []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buffer.Write(data) +} + +func (b *lockedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buffer.String() +} diff --git a/tests/integration/stdio_test.go b/tests/integration/stdio_test.go new file mode 100644 index 0000000..1d3aae0 --- /dev/null +++ b/tests/integration/stdio_test.go @@ -0,0 +1,291 @@ +//go:build integration + +package integration_test + +import ( + "bufio" + "bytes" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/P4ST4S/mcp-audit/internal/audit" +) + +func TestStdioBinaryHappyPathRedactionAndGracefulShutdown(t *testing.T) { + auditPath := filepath.Join(t.TempDir(), "audit.jsonl") + configPath := writeStdioConfig(t, auditPath, "") + command, stdin, scanner, stderr := startStdioProxy(t, configPath) + + sendJSONRPC(t, stdin, `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2026-07-28","capabilities":{},"clientInfo":{"name":"integration","version":"1.0"}}}`) + assertResponseID(t, scanner, 1) + sendJSONRPC(t, stdin, `{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"message":"hello","token":"top-secret"}}}`) + assertResponseID(t, scanner, 2) + stopProcess(t, command, stderr) + + entries := readAuditEntries(t, auditPath) + var toolEntry *audit.Entry + for index := range entries { + if entries[index].Method == "tools/call" { + toolEntry = &entries[index] + } + } + if toolEntry == nil { + t.Fatalf("tools/call audit entry not found: %#v", entries) + } + if toolEntry.Outcome != audit.OutcomeSuccess || toolEntry.Integrity == nil || toolEntry.Signature == "" { + t.Fatalf("terminal signed entry = %#v", toolEntry) + } + if !bytes.Contains(toolEntry.Params, []byte(`"token":"[REDACTED]"`)) || bytes.Contains(toolEntry.Params, []byte("top-secret")) { + t.Fatalf("redacted params = %s", toolEntry.Params) + } + + verify := exec.Command(binaryPath, "verify", auditPath, "--json") + verify.Env = commandEnv(map[string]string{"MCP_AUDIT_SIGNING_SECRET": signingSecret}, "AUDIT_SECRET") + output, err := verify.CombinedOutput() + if err != nil { + t.Fatalf("verify audit artifact: %v\n%s", err, output) + } + var result struct { + Verified int `json:"verified"` + Invalid int `json:"invalid"` + Unsigned int `json:"unsigned"` + } + if err := json.Unmarshal(output, &result); err != nil { + t.Fatalf("decode verification output: %v\n%s", err, output) + } + if result.Verified < 2 || result.Invalid != 0 || result.Unsigned != 0 { + t.Fatalf("verification result = %+v", result) + } +} + +func TestStdioBinaryPolicyDeny(t *testing.T) { + auditPath := filepath.Join(t.TempDir(), "audit.jsonl") + policy := `policy: + enabled: true + default_action: allow + scope: all_operations + rules: + - action: deny + method: tools/call + name: delete_file + reason: blocked by integration policy +` + configPath := writeStdioConfig(t, auditPath, policy) + command, stdin, scanner, stderr := startStdioProxy(t, configPath) + sendJSONRPC(t, stdin, `{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"delete_file","arguments":{}}}`) + response := scanResponse(t, scanner) + errorObject := response["error"].(map[string]any) + if errorObject["code"] != float64(-32030) { + t.Fatalf("response = %#v", response) + } + _ = stdin.Close() + waitProcess(t, command, stderr) + + entries := readAuditEntries(t, auditPath) + if len(entries) != 1 || entries[0].Outcome != audit.OutcomeDenied || entries[0].Error == nil || entries[0].Error.Code != -32030 { + t.Fatalf("policy audit entries = %#v", entries) + } +} + +func TestStdioBinaryRateLimitsSixtyFirstCall(t *testing.T) { + auditPath := filepath.Join(t.TempDir(), "audit.jsonl") + configPath := writeStdioConfig(t, auditPath, "") + command, stdin, scanner, stderr := startStdioProxy(t, configPath) + for id := 1; id <= 61; id++ { + sendJSONRPC(t, stdin, fmt.Sprintf(`{"jsonrpc":"2.0","id":%d,"method":"tools/call","params":{"name":"echo","arguments":{}}}`, id)) + response := scanResponse(t, scanner) + if id <= 60 { + if response["result"] == nil { + t.Fatalf("request %d was rejected: %#v", id, response) + } + continue + } + errorObject := response["error"].(map[string]any) + if errorObject["code"] != float64(-32029) { + t.Fatalf("61st response = %#v", response) + } + } + _ = stdin.Close() + waitProcess(t, command, stderr) + + entries := readAuditEntries(t, auditPath) + if len(entries) != 61 || entries[len(entries)-1].Outcome != audit.OutcomeRateLimited { + t.Fatalf("rate-limit audit count/outcome = %d/%q", len(entries), entries[len(entries)-1].Outcome) + } +} + +func TestBinaryFailsClosedBeforeStartingUpstream(t *testing.T) { + directory := t.TempDir() + marker := filepath.Join(directory, "started") + upstream := fmt.Sprintf("touch %s", marker) + config := fmt.Sprintf(`proxy: + transport: stdio + upstream: %q +audit: + sign: true +dashboard: + enabled: false +metrics: + enabled: false +`, upstream) + configPath := writeConfig(t, config) + command := exec.Command(binaryPath, "--config", configPath, "--log-level", "error") + command.Env = commandEnv(nil, "MCP_AUDIT_SIGNING_SECRET", "AUDIT_SECRET") + output, err := command.CombinedOutput() + if err == nil || !bytes.Contains(output, []byte("signing is enabled but no signing secret")) { + t.Fatalf("exit error/output = %v/%s", err, output) + } + if _, err := os.Stat(marker); !os.IsNotExist(err) { + t.Fatalf("upstream marker exists or stat failed: %v", err) + } +} + +func TestVerifySQLiteArtifactProducedByBinary(t *testing.T) { + databasePath := filepath.Join(t.TempDir(), "audit.db") + config := fmt.Sprintf(`proxy: + transport: stdio + upstream: %q +audit: + storage: sqlite + sqlite_path: %q + sign: true +dashboard: + enabled: false +metrics: + enabled: false +`, upstreamPath, databasePath) + command, stdin, _, stderr := startStdioProxy(t, writeConfig(t, config)) + sendJSONRPC(t, stdin, `{"jsonrpc":"2.0","method":"notifications/initialized"}`) + _ = stdin.Close() + waitProcess(t, command, stderr) + + verify := exec.Command(binaryPath, "verify", databasePath, "--format", "sqlite", "--json") + verify.Env = commandEnv(map[string]string{"MCP_AUDIT_SIGNING_SECRET": signingSecret}, "AUDIT_SECRET") + output, err := verify.CombinedOutput() + if err != nil { + t.Fatalf("verify SQLite artifact: %v\n%s", err, output) + } + var result struct { + Verified int `json:"verified"` + Invalid int `json:"invalid"` + } + if err := json.Unmarshal(output, &result); err != nil || result.Verified != 1 || result.Invalid != 0 { + t.Fatalf("SQLite verification = %+v, err %v\n%s", result, err, output) + } +} + +func writeStdioConfig(t *testing.T, auditPath, extra string) string { + t.Helper() + config := fmt.Sprintf(`proxy: + transport: stdio + upstream: %q + client_id: integration-client + server_id: integration-upstream +audit: + storage: jsonl + path: %q + sign: true +middleware: + rate_limit: + enabled: true + requests_per_minute: 60 + redact: + enabled: true + patterns: [token, secret, authorization, bearer] +dashboard: + enabled: false +metrics: + enabled: false +%s`, upstreamPath, auditPath, extra) + return writeConfig(t, config) +} + +func startStdioProxy(t *testing.T, configPath string) (*exec.Cmd, *os.File, *bufio.Scanner, *lockedBuffer) { + t.Helper() + command := exec.Command(binaryPath, "--config", configPath, "--log-level", "error") + command.Env = commandEnv(map[string]string{"MCP_AUDIT_SIGNING_SECRET": signingSecret}, "AUDIT_SECRET") + stdin, err := command.StdinPipe() + if err != nil { + t.Fatalf("stdin pipe: %v", err) + } + stdout, err := command.StdoutPipe() + if err != nil { + t.Fatalf("stdout pipe: %v", err) + } + stderr := newLockedBuffer() + command.Stderr = stderr + if err := command.Start(); err != nil { + t.Fatalf("start proxy: %v", err) + } + t.Cleanup(func() { + if command.Process != nil && command.ProcessState == nil { + _ = command.Process.Kill() + } + }) + return command, stdin.(*os.File), bufio.NewScanner(stdout), stderr +} + +func sendJSONRPC(t *testing.T, stdin *os.File, message string) { + t.Helper() + if _, err := fmt.Fprintln(stdin, message); err != nil { + t.Fatalf("write request: %v", err) + } +} + +func assertResponseID(t *testing.T, scanner *bufio.Scanner, id int) { + t.Helper() + response := scanResponse(t, scanner) + if response["id"] != float64(id) || response["result"] == nil { + t.Fatalf("response = %#v, want id %d result", response, id) + } +} + +func scanResponse(t *testing.T, scanner *bufio.Scanner) map[string]any { + t.Helper() + if !scanner.Scan() { + t.Fatalf("read response: %v", scanner.Err()) + } + var response map[string]any + if err := json.Unmarshal(scanner.Bytes(), &response); err != nil { + t.Fatalf("decode response: %v\n%s", err, scanner.Bytes()) + } + return response +} + +func waitProcess(t *testing.T, command *exec.Cmd, stderr *lockedBuffer) { + t.Helper() + wait := make(chan error, 1) + go func() { wait <- command.Wait() }() + select { + case err := <-wait: + if err != nil { + t.Fatalf("process exit: %v\n%s", err, stderr.String()) + } + case <-time.After(5 * time.Second): + _ = command.Process.Kill() + t.Fatalf("process did not exit: %s", stderr.String()) + } +} + +func readAuditEntries(t *testing.T, path string) []audit.Entry { + t.Helper() + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read audit log: %v", err) + } + var entries []audit.Entry + for _, line := range strings.Split(strings.TrimSpace(string(raw)), "\n") { + var entry audit.Entry + if err := json.Unmarshal([]byte(line), &entry); err != nil { + t.Fatalf("decode audit entry: %v\n%s", err, line) + } + entries = append(entries, entry) + } + return entries +} diff --git a/tests/integration/testdata/upstream/main.go b/tests/integration/testdata/upstream/main.go new file mode 100644 index 0000000..32a59e9 --- /dev/null +++ b/tests/integration/testdata/upstream/main.go @@ -0,0 +1,30 @@ +package main + +import ( + "bufio" + "encoding/json" + "fmt" + "os" +) + +type request struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` +} + +func main() { + scanner := bufio.NewScanner(os.Stdin) + for scanner.Scan() { + var message request + if err := json.Unmarshal(scanner.Bytes(), &message); err != nil || message.Method == "" || len(message.ID) == 0 { + continue + } + result := `{"ok":true}` + if message.Method == "initialize" { + result = `{"protocolVersion":"2026-07-28","capabilities":{"tools":{}},"serverInfo":{"name":"integration-upstream","version":"1.0.0"}}` + } else if message.Method == "tools/call" { + result = `{"content":[{"type":"text","text":"ok"}]}` + } + fmt.Printf("{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":%s}\n", message.ID, result) + } +}