From 76d52bce073c5a2e52611c45b26f8a9972bd7fc0 Mon Sep 17 00:00:00 2001 From: PrashantBtkl Date: Wed, 30 Sep 2026 16:05:05 +0530 Subject: [PATCH 1/3] feat(health): report redacted datasource/SSH integration errors Add error_class to datasource health entries and record request-time failures so they surface in the next health report. Errors are redacted (credentials, keys, IPs, host:port) and truncated before leaving the host. Refs nudgebee/nudgebee-enterprise#39578 Co-Authored-By: Claude Sonnet 5.5 --- docs/connection-lifecycle.md | 2 ++ pkg/proxy/errors.go | 48 ++++++++++++++++++++++++++ pkg/proxy/errors_test.go | 52 ++++++++++++++++++++++++++++ pkg/proxy/registry.go | 67 +++++++++++++++++++++++++++++++----- pkg/ws/handler.go | 2 ++ 5 files changed, 163 insertions(+), 8 deletions(-) create mode 100644 pkg/proxy/errors.go create mode 100644 pkg/proxy/errors_test.go diff --git a/docs/connection-lifecycle.md b/docs/connection-lifecycle.md index e91b1a6..9b58f53 100644 --- a/docs/connection-lifecycle.md +++ b/docs/connection-lifecycle.md @@ -37,6 +37,8 @@ sequenceDiagram end ``` +**Integration errors:** Each entry in the health report carries `status`, and while `status` is `"error"` also `error` and `error_class` (`datasource_integration_failure` or `ssh_integration_failure`). Errors come from failed health checks and from failed proxied requests (kept for up to 15 minutes, cleared by the next successful request). They are redacted before leaving the host (credentials, private keys, IPs, `host:port`) and truncated to 512 chars. Nothing is stored server-side by the agent: once the integration recovers, the next report simply omits the error. + **Auto-reconnect:** On disconnect, the agent reconnects with exponential backoff (3s → 6s → 12s → ... → 30s max). ## Config Sync diff --git a/pkg/proxy/errors.go b/pkg/proxy/errors.go new file mode 100644 index 0000000..8125c32 --- /dev/null +++ b/pkg/proxy/errors.go @@ -0,0 +1,48 @@ +package proxy + +import ( + "regexp" + "strings" +) + +// Error classes surfaced on the Proxy Agent tab of the Agent Health page. +const ( + ErrorClassDatasource = "datasource_integration_failure" + ErrorClassSSH = "ssh_integration_failure" + + maxReportedErrorLen = 512 +) + +// ErrorClass maps a proxy type to the error class reported upstream. +func ErrorClass(proxyType string) string { + if proxyType == "ssh-proxy" { + return ErrorClassSSH + } + return ErrorClassDatasource +} + +var ( + // scheme://user:pass@host → scheme://[redacted]@host + urlUserinfoRe = regexp.MustCompile(`([a-zA-Z][a-zA-Z0-9+.-]*://)[^\s/@]+@`) + // password=..., token: ..., secret=... etc. + secretKVRe = regexp.MustCompile(`(?i)\b(password|passwd|pwd|secret|token|api[_-]?key|access[_-]?key|private[_-]?key)(\s*[=:]\s*)("[^"]*"|'[^']*'|\S+)`) + pemBlockRe = regexp.MustCompile(`(?s)-----BEGIN [A-Z ]*PRIVATE KEY-----.*?(-----END [A-Z ]*PRIVATE KEY-----|$)`) + ipv4Re = regexp.MustCompile(`\b\d{1,3}(?:\.\d{1,3}){3}\b`) + // host:port pairs such as db.internal.example.com:5432 + hostPortRe = regexp.MustCompile(`\b[a-zA-Z0-9-]+(?:\.[a-zA-Z0-9-]+)+:\d{1,5}\b`) +) + +// RedactError strips credentials and hostnames/IPs from an error message +// before it leaves the host, and bounds its length. +func RedactError(msg string) string { + msg = pemBlockRe.ReplaceAllString(msg, "[redacted-key]") + msg = urlUserinfoRe.ReplaceAllString(msg, "${1}[redacted]@") + msg = secretKVRe.ReplaceAllString(msg, "${1}${2}[redacted]") + msg = ipv4Re.ReplaceAllString(msg, "[redacted-host]") + msg = hostPortRe.ReplaceAllString(msg, "[redacted-host]") + msg = strings.Join(strings.Fields(msg), " ") + if len(msg) > maxReportedErrorLen { + msg = msg[:maxReportedErrorLen] + "…" + } + return msg +} diff --git a/pkg/proxy/errors_test.go b/pkg/proxy/errors_test.go new file mode 100644 index 0000000..a48e85c --- /dev/null +++ b/pkg/proxy/errors_test.go @@ -0,0 +1,52 @@ +package proxy + +import ( + "context" + "errors" + "strings" + "testing" +) + +func TestRedactError(t *testing.T) { + cases := map[string]string{ + "dial tcp 10.1.2.3:5432: connection refused": "10.1.2.3", + "connect to db.internal.example.com:5432 failed": "db.internal.example.com", + "postgres://admin:hunter2@db/x failed": "hunter2", + "auth failed password=hunter2 for user": "hunter2", + "key -----BEGIN RSA PRIVATE KEY-----\nabc\n-----END RSA PRIVATE KEY-----": "abc", + } + for in, leak := range cases { + if out := RedactError(in); strings.Contains(out, leak) { + t.Errorf("RedactError(%q) = %q leaks %q", in, out, leak) + } + } + if got := RedactError(strings.Repeat("x", 2000)); len(got) > maxReportedErrorLen+4 { + t.Errorf("not truncated: %d", len(got)) + } +} + +func TestErrorClass(t *testing.T) { + if ErrorClass("ssh-proxy") != ErrorClassSSH || ErrorClass("db-proxy") != ErrorClassDatasource { + t.Fatal("wrong class") + } +} + +type stubProxy struct{ Proxy } + +func (stubProxy) HealthCheck(context.Context) error { return nil } + +func TestRequestErrorReportedThenCleared(t *testing.T) { + r := NewRegistry() + r.Register("ssh1", DatasourceEntry{ID: "ssh1", ProxyType: "ssh-proxy"}, stubProxy{}) + + r.RecordRequestResult("ssh1", errors.New("dial tcp 10.0.0.5:22: refused")) + h := r.HealthReport(context.Background())["ssh1"] + if h.Status != "error" || h.ErrorClass != ErrorClassSSH || strings.Contains(h.Error, "10.0.0.5") { + t.Fatalf("unexpected: %+v", h) + } + + r.RecordRequestResult("ssh1", nil) + if h := r.HealthReport(context.Background())["ssh1"]; h.Status != "healthy" || h.ErrorClass != "" { + t.Fatalf("not cleared: %+v", h) + } +} diff --git a/pkg/proxy/registry.go b/pkg/proxy/registry.go index 5abb06a..60acfe3 100644 --- a/pkg/proxy/registry.go +++ b/pkg/proxy/registry.go @@ -70,6 +70,48 @@ type Registry struct { mu sync.RWMutex proxies map[string]Proxy // datasource ID → Proxy configs map[string]DatasourceEntry + + reqErrMu sync.Mutex + reqErrors map[string]requestError // datasource ID → last unrecovered request failure +} + +type requestError struct { + msg string + at time.Time +} + +// requestErrorTTL bounds how long a request failure is reported without a +// later successful request to the same datasource. +const requestErrorTTL = 15 * time.Minute + +// RecordRequestResult notes the outcome of a proxied request. A failure is +// reported on the next health report (deduplicated per datasource, latest +// wins); a success clears it. +func (r *Registry) RecordRequestResult(datasourceID string, err error) { + r.reqErrMu.Lock() + defer r.reqErrMu.Unlock() + if err == nil { + delete(r.reqErrors, datasourceID) + return + } + if r.reqErrors == nil { + r.reqErrors = make(map[string]requestError) + } + r.reqErrors[datasourceID] = requestError{msg: RedactError(err.Error()), at: time.Now()} +} + +func (r *Registry) recentRequestError(datasourceID string) (string, bool) { + r.reqErrMu.Lock() + defer r.reqErrMu.Unlock() + e, ok := r.reqErrors[datasourceID] + if !ok { + return "", false + } + if time.Since(e.at) > requestErrorTTL { + delete(r.reqErrors, datasourceID) + return "", false + } + return e.msg, true } // DatasourceEntry stores the proxy and its metadata. @@ -146,7 +188,10 @@ type DatasourceHealth struct { Name string `json:"name"` Status string `json:"status"` // "healthy", "error", "unknown" Error string `json:"error,omitempty"` - LastCheck string `json:"last_check"` // RFC3339 + // ErrorClass is set only while Status is "error"; it clears on the next + // successful check, so the report always reflects current state. + ErrorClass string `json:"error_class,omitempty"` + LastCheck string `json:"last_check"` // RFC3339 } const ( @@ -181,12 +226,13 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth now := time.Now().UTC().Format(time.RFC3339) for _, t := range targets { report[t.id] = DatasourceHealth{ - Type: t.cfg.Type, - ProxyType: t.cfg.ProxyType, - Name: t.cfg.Name, - Status: "error", - Error: err.Error(), - LastCheck: now, + Type: t.cfg.Type, + ProxyType: t.cfg.ProxyType, + Name: t.cfg.Name, + Status: "error", + Error: RedactError(err.Error()), + ErrorClass: ErrorClass(t.cfg.ProxyType), + LastCheck: now, } } return report @@ -243,7 +289,12 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth if err != nil { health.Status = "error" - health.Error = err.Error() + health.Error = RedactError(err.Error()) + health.ErrorClass = ErrorClass(t.cfg.ProxyType) + } else if msg, ok := r.recentRequestError(t.id); ok { + health.Status = "error" + health.Error = msg + health.ErrorClass = ErrorClass(t.cfg.ProxyType) } else { health.Status = "healthy" } diff --git a/pkg/ws/handler.go b/pkg/ws/handler.go index ba6d622..fc79aec 100644 --- a/pkg/ws/handler.go +++ b/pkg/ws/handler.go @@ -173,6 +173,7 @@ func (h *Handler) handleRequest(ctx context.Context, msg []byte, requestID, data } resp, err := p.HandleRequest(ctx, &req) + h.registry.RecordRequestResult(datasourceID, err) if err != nil { h.logger.Error("proxy request failed", "action", req.Action, "datasource", datasourceID, "err", err) return h.buildErrorResponse(requestID, 500, err.Error()), nil @@ -214,6 +215,7 @@ func (h *Handler) handleLegacyRequest(ctx context.Context, msg []byte, requestID } resp, err := p.HandleRequest(ctx, req) + h.registry.RecordRequestResult(datasourceID, err) if err != nil { h.logger.Error("legacy proxy request failed", "action", req.Action, "datasource", datasourceID, "err", err) return h.buildErrorResponse(requestID, 500, err.Error()), nil From 2cd19cfbbcb828e95ab1fdccb2806c83fcece057 Mon Sep 17 00:00:00 2001 From: PrashantBtkl Date: Wed, 30 Sep 2026 16:40:25 +0530 Subject: [PATCH 2/3] fix(health): address review feedback on integration error reporting - only record connectivity/auth failures; ignore caller errors and cancelled request contexts - clear recorded request errors on Register/Remove/CloseAll and ignore results for unregistered datasources - redaction: bearer/basic, JSON-style secrets, IPv6, lookup/dial hosts, UTF-8-safe truncation; redact outside the lock - add ErrorClass and redaction to the per-target ctx-error branch - tests assert exact redacted output; add lifecycle and race coverage Co-Authored-By: Claude Sonnet 5.5 --- docs/connection-lifecycle.md | 2 +- pkg/proxy/errors.go | 71 +++++++++++++-- pkg/proxy/errors_test.go | 162 +++++++++++++++++++++++++++++++---- pkg/proxy/registry.go | 45 ++++++++-- pkg/ws/handler.go | 4 +- 5 files changed, 254 insertions(+), 30 deletions(-) diff --git a/docs/connection-lifecycle.md b/docs/connection-lifecycle.md index 9b58f53..bdf08f6 100644 --- a/docs/connection-lifecycle.md +++ b/docs/connection-lifecycle.md @@ -37,7 +37,7 @@ sequenceDiagram end ``` -**Integration errors:** Each entry in the health report carries `status`, and while `status` is `"error"` also `error` and `error_class` (`datasource_integration_failure` or `ssh_integration_failure`). Errors come from failed health checks and from failed proxied requests (kept for up to 15 minutes, cleared by the next successful request). They are redacted before leaving the host (credentials, private keys, IPs, `host:port`) and truncated to 512 chars. Nothing is stored server-side by the agent: once the integration recovers, the next report simply omits the error. +**Integration errors:** Each entry in the health report carries `status`, and while `status` is `"error"` also `error` and `error_class` (`datasource_integration_failure` or `ssh_integration_failure`). Errors come from failed health checks and from proxied requests that fail with a connectivity or authentication error (kept for up to 15 minutes, cleared by the next successful request or when the datasource is re-registered or removed). Caller mistakes (bad SQL, disallowed commands, invalid input) and cancelled requests are not reported. Messages are redacted on a best-effort basis before entering the report (credentials, bearer tokens, private keys, IP addresses, `host:port`, and hosts named after `lookup`/`dial`) and truncated to 512 chars; a bare hostname or username in free-form text may not be caught. The raw error is still returned in the response to the failing request itself. Once the integration recovers, the next report simply omits the error. **Auto-reconnect:** On disconnect, the agent reconnects with exponential backoff (3s → 6s → 12s → ... → 30s max). diff --git a/pkg/proxy/errors.go b/pkg/proxy/errors.go index 8125c32..6b028df 100644 --- a/pkg/proxy/errors.go +++ b/pkg/proxy/errors.go @@ -1,8 +1,12 @@ package proxy import ( + "errors" + "io" + "net" "regexp" "strings" + "unicode/utf8" ) // Error classes surfaced on the Proxy Agent tab of the Agent Health page. @@ -21,28 +25,85 @@ func ErrorClass(proxyType string) string { return ErrorClassDatasource } +// integrationErrorMarkers are substrings (lowercase) of errors that mean the +// datasource itself is unreachable or rejecting our credentials, as opposed to +// the caller sending a bad query or hitting a policy rule. +var integrationErrorMarkers = []string{ + "connection refused", "connection reset", "broken pipe", "no such host", + "i/o timeout", "network is unreachable", "no route to host", + "handshake failed", "unable to authenticate", "authentication failed", + "password authentication", "access denied for user", "permission denied (publickey", + "host key", "x509:", "tls:", "sasl", "noauth", "wrongpass", +} + +// IsIntegrationError reports whether err looks like a connectivity or +// authentication failure of the integration, rather than a caller mistake +// (invalid input, disallowed command, SQL error) which must not mark a +// healthy datasource as broken. +func IsIntegrationError(err error) bool { + if err == nil { + return false + } + var netErr net.Error + if errors.As(err, &netErr) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + return true + } + msg := strings.ToLower(err.Error()) + for _, m := range integrationErrorMarkers { + if strings.Contains(msg, m) { + return true + } + } + return false +} + +const secretKeys = `password|passwd|pwd|passphrase|secret|token|api[_-]?key|access[_-]?key|private[_-]?key|authorization|credentials?` + var ( // scheme://user:pass@host → scheme://[redacted]@host urlUserinfoRe = regexp.MustCompile(`([a-zA-Z][a-zA-Z0-9+.-]*://)[^\s/@]+@`) - // password=..., token: ..., secret=... etc. - secretKVRe = regexp.MustCompile(`(?i)\b(password|passwd|pwd|secret|token|api[_-]?key|access[_-]?key|private[_-]?key)(\s*[=:]\s*)("[^"]*"|'[^']*'|\S+)`) + // password=..., "password":"...", token: ... (key may be followed by a closing quote) + secretKVRe = regexp.MustCompile(`(?i)\b(` + secretKeys + `)("?\s*[=:]\s*)("[^"]*"|'[^']*'|\S+)`) + // Authorization header schemes with an inline credential. + bearerRe = regexp.MustCompile(`(?i)\b(bearer|basic)\s+[A-Za-z0-9._~+/=-]{6,}`) pemBlockRe = regexp.MustCompile(`(?s)-----BEGIN [A-Z ]*PRIVATE KEY-----.*?(-----END [A-Z ]*PRIVATE KEY-----|$)`) - ipv4Re = regexp.MustCompile(`\b\d{1,3}(?:\.\d{1,3}){3}\b`) - // host:port pairs such as db.internal.example.com:5432 + + // [fd00::5]:5432 and bare IPv6 with at least five groups. + ipv6BracketRe = regexp.MustCompile(`\[[0-9a-fA-F:.]*:[0-9a-fA-F:.]*\]`) + ipv6BareRe = regexp.MustCompile(`\b(?:[0-9a-fA-F]{1,4}:){4,7}[0-9a-fA-F]{1,4}\b`) + ipv4Re = regexp.MustCompile(`\b\d{1,3}(?:\.\d{1,3}){3}\b`) + // host.domain:port pairs such as db.internal.example.com:5432 hostPortRe = regexp.MustCompile(`\b[a-zA-Z0-9-]+(?:\.[a-zA-Z0-9-]+)+:\d{1,5}\b`) + // Go's resolver and dialer name the host right after these words, with or + // without dots: "lookup db.internal: no such host", "dial tcp postgres:5432". + lookupHostRe = regexp.MustCompile(`(?i)(\blookup\s+)[^\s:]+`) + dialHostRe = regexp.MustCompile(`(?i)(\bdial\s+(?:tcp|udp)\d?\s+)[^\s:]+`) ) // RedactError strips credentials and hostnames/IPs from an error message // before it leaves the host, and bounds its length. +// +// It is best-effort: a bare hostname or a username in free-form error text +// that follows none of the patterns above is not detected. func RedactError(msg string) string { + msg = strings.ToValidUTF8(msg, "") msg = pemBlockRe.ReplaceAllString(msg, "[redacted-key]") msg = urlUserinfoRe.ReplaceAllString(msg, "${1}[redacted]@") + msg = bearerRe.ReplaceAllString(msg, "${1} [redacted]") msg = secretKVRe.ReplaceAllString(msg, "${1}${2}[redacted]") + msg = ipv6BracketRe.ReplaceAllString(msg, "[redacted-host]") + msg = ipv6BareRe.ReplaceAllString(msg, "[redacted-host]") msg = ipv4Re.ReplaceAllString(msg, "[redacted-host]") msg = hostPortRe.ReplaceAllString(msg, "[redacted-host]") + msg = lookupHostRe.ReplaceAllString(msg, "${1}[redacted-host]") + msg = dialHostRe.ReplaceAllString(msg, "${1}[redacted-host]") msg = strings.Join(strings.Fields(msg), " ") if len(msg) > maxReportedErrorLen { - msg = msg[:maxReportedErrorLen] + "…" + i := maxReportedErrorLen + for i > 0 && !utf8.RuneStart(msg[i]) { + i-- + } + msg = msg[:i] + "…" } return msg } diff --git a/pkg/proxy/errors_test.go b/pkg/proxy/errors_test.go index a48e85c..b1fea95 100644 --- a/pkg/proxy/errors_test.go +++ b/pkg/proxy/errors_test.go @@ -3,25 +3,49 @@ package proxy import ( "context" "errors" + "fmt" + "net" "strings" + "sync" "testing" + "unicode/utf8" ) func TestRedactError(t *testing.T) { - cases := map[string]string{ - "dial tcp 10.1.2.3:5432: connection refused": "10.1.2.3", - "connect to db.internal.example.com:5432 failed": "db.internal.example.com", - "postgres://admin:hunter2@db/x failed": "hunter2", - "auth failed password=hunter2 for user": "hunter2", - "key -----BEGIN RSA PRIVATE KEY-----\nabc\n-----END RSA PRIVATE KEY-----": "abc", - } - for in, leak := range cases { - if out := RedactError(in); strings.Contains(out, leak) { - t.Errorf("RedactError(%q) = %q leaks %q", in, out, leak) - } + cases := []struct{ name, in, want string }{ + {"ipv4", "dial tcp 10.1.2.3:5432: connection refused", "dial tcp [redacted-host]:5432: connection refused"}, + {"dotted host", "connect to db.internal.example.com:5432 failed", "connect to [redacted-host] failed"}, + {"url userinfo", "postgres://admin:hunter2@db/x failed", "postgres://[redacted]@db/x failed"}, + {"kv", "auth failed password=hunter2 for user", "auth failed password=[redacted] for user"}, + {"json", `body {"password":"hunter2","db":"x"}`, `body {"password":[redacted],"db":"x"}`}, + {"bearer", "401 Authorization: Bearer abc123def456", "401 Authorization: [redacted] [redacted]"}, + {"basic", "sent Basic dXNlcjpwYXNz rejected", "sent Basic [redacted] rejected"}, + {"ipv6 bracket", "dial tcp [fd00::5]:5432: refused", "dial tcp [redacted-host]:5432: refused"}, + {"lookup", "dial tcp: lookup db.internal.example.com: no such host", "dial tcp: lookup [redacted-host]: no such host"}, + {"dial single label", "dial tcp postgres:5432: connect: refused", "dial tcp [redacted-host]:5432: connect: refused"}, + {"pem", "key -----BEGIN RSA PRIVATE KEY-----\nabc\n-----END RSA PRIVATE KEY-----", "key [redacted-key]"}, + {"plain", "connection refused", "connection refused"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := RedactError(tc.in); got != tc.want { + t.Errorf("RedactError(%q)\n got %q\nwant %q", tc.in, got, tc.want) + } + }) + } +} + +func TestRedactErrorTruncatesOnRuneBoundary(t *testing.T) { + // 3-byte runes so byte 512 lands mid-rune. + got := RedactError(strings.Repeat("€", 400)) + if !utf8.ValidString(got) { + t.Fatalf("invalid UTF-8 after truncation: %q", got) } - if got := RedactError(strings.Repeat("x", 2000)); len(got) > maxReportedErrorLen+4 { - t.Errorf("not truncated: %d", len(got)) + if !strings.HasSuffix(got, "…") || len(got) > maxReportedErrorLen+len("…") { + t.Fatalf("not truncated properly: len=%d", len(got)) + } + if !utf8.ValidString(RedactError("bad \xff byte")) { + t.Fatal("invalid input bytes not scrubbed") } } @@ -31,22 +55,128 @@ func TestErrorClass(t *testing.T) { } } +func TestIsIntegrationError(t *testing.T) { + yes := []error{ + &net.OpError{Op: "dial", Err: errors.New("boom")}, + errors.New("dial tcp 1.2.3.4:22: connect: connection refused"), + errors.New("ssh: handshake failed: ssh: unable to authenticate"), + errors.New(`pq: password authentication failed for user "x"`), + } + no := []error{ + nil, + errors.New("command \"FLUSHALL\" not allowed"), + errors.New("datasource is read-only, execute not allowed"), + errors.New("invalid base64 content"), + errors.New(`ERROR: relation "typo_table" does not exist`), + } + for _, e := range yes { + if !IsIntegrationError(e) { + t.Errorf("want integration error: %v", e) + } + } + for _, e := range no { + if IsIntegrationError(e) { + t.Errorf("want caller error: %v", e) + } + } +} + type stubProxy struct{ Proxy } func (stubProxy) HealthCheck(context.Context) error { return nil } +func (stubProxy) Close() error { return nil } + +var errRefused = errors.New("dial tcp 10.0.0.5:22: connect: connection refused") func TestRequestErrorReportedThenCleared(t *testing.T) { r := NewRegistry() r.Register("ssh1", DatasourceEntry{ID: "ssh1", ProxyType: "ssh-proxy"}, stubProxy{}) - r.RecordRequestResult("ssh1", errors.New("dial tcp 10.0.0.5:22: refused")) + r.RecordRequestResult(context.Background(), "ssh1", errRefused) h := r.HealthReport(context.Background())["ssh1"] - if h.Status != "error" || h.ErrorClass != ErrorClassSSH || strings.Contains(h.Error, "10.0.0.5") { + if h.Status != "error" || h.ErrorClass != ErrorClassSSH || strings.Contains(h.Error, "10.0.0.5") || h.Error == "" { t.Fatalf("unexpected: %+v", h) } - r.RecordRequestResult("ssh1", nil) + r.RecordRequestResult(context.Background(), "ssh1", nil) if h := r.HealthReport(context.Background())["ssh1"]; h.Status != "healthy" || h.ErrorClass != "" { t.Fatalf("not cleared: %+v", h) } } + +func TestCallerErrorsAndCancelledRequestsIgnored(t *testing.T) { + r := NewRegistry() + r.Register("db1", DatasourceEntry{ID: "db1", ProxyType: "db-proxy"}, stubProxy{}) + + r.RecordRequestResult(context.Background(), "db1", errors.New(`relation "typo" does not exist`)) + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + r.RecordRequestResult(cancelled, "db1", errRefused) + + if h := r.HealthReport(context.Background())["db1"]; h.Status != "healthy" { + t.Fatalf("caller error flagged datasource: %+v", h) + } +} + +func TestRequestErrorClearedOnRegisterAndRemove(t *testing.T) { + r := NewRegistry() + r.Register("ssh1", DatasourceEntry{ID: "ssh1", ProxyType: "ssh-proxy"}, stubProxy{}) + r.RecordRequestResult(context.Background(), "ssh1", errRefused) + + // Re-registering (e.g. after a corrected password) discards the old failure. + r.Register("ssh1", DatasourceEntry{ID: "ssh1", ProxyType: "ssh-proxy"}, stubProxy{}) + if h := r.HealthReport(context.Background())["ssh1"]; h.Status != "healthy" { + t.Fatalf("stale error after Register: %+v", h) + } + + r.RecordRequestResult(context.Background(), "ssh1", errRefused) + if err := r.Remove("ssh1"); err != nil { + t.Fatal(err) + } + // A late result for a removed datasource must not recreate an entry. + r.RecordRequestResult(context.Background(), "ssh1", errRefused) + r.reqErrMu.Lock() + n := len(r.reqErrors) + r.reqErrMu.Unlock() + if n != 0 { + t.Fatalf("reqErrors leaked %d entries", n) + } + + r.Register("a", DatasourceEntry{ID: "a"}, stubProxy{}) + r.RecordRequestResult(context.Background(), "a", errRefused) + r.CloseAll() + r.reqErrMu.Lock() + n = len(r.reqErrors) + r.reqErrMu.Unlock() + if n != 0 { + t.Fatalf("CloseAll leaked %d entries", n) + } +} + +func TestRecordRequestResultConcurrent(t *testing.T) { + r := NewRegistry() + for i := 0; i < 4; i++ { + r.Register(fmt.Sprint("d", i), DatasourceEntry{ID: fmt.Sprint("d", i)}, stubProxy{}) + } + var wg sync.WaitGroup + for i := 0; i < 32; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + id := fmt.Sprint("d", i%4) + for j := 0; j < 50; j++ { + switch j % 4 { + case 0: + r.RecordRequestResult(context.Background(), id, errRefused) + case 1: + r.RecordRequestResult(context.Background(), id, nil) + case 2: + r.HealthReport(context.Background()) + default: + r.Register(id, DatasourceEntry{ID: id}, stubProxy{}) + } + } + }(i) + } + wg.Wait() +} diff --git a/pkg/proxy/registry.go b/pkg/proxy/registry.go index 60acfe3..58ed19b 100644 --- a/pkg/proxy/registry.go +++ b/pkg/proxy/registry.go @@ -84,10 +84,30 @@ type requestError struct { // later successful request to the same datasource. const requestErrorTTL = 15 * time.Minute -// RecordRequestResult notes the outcome of a proxied request. A failure is -// reported on the next health report (deduplicated per datasource, latest -// wins); a success clears it. -func (r *Registry) RecordRequestResult(datasourceID string, err error) { +// RecordRequestResult notes the outcome of a proxied request. A failure that +// looks like an integration problem (see IsIntegrationError) is reported on +// the next health report (deduplicated per datasource, latest wins); a +// success clears it. Failures caused by the caller — bad input, policy +// rejections, a cancelled or timed-out request context — are ignored. +// +// Lock order is mu → reqErrMu everywhere. +func (r *Registry) RecordRequestResult(ctx context.Context, datasourceID string, err error) { + var msg string + if err != nil { + if ctx.Err() != nil || !IsIntegrationError(err) { + return + } + msg = RedactError(err.Error()) // outside the locks: several regexes + } + + r.mu.RLock() + defer r.mu.RUnlock() + // Ignore results for datasources that were removed while the request was + // in flight, so they cannot resurrect an entry nothing will clean up. + if _, ok := r.proxies[datasourceID]; !ok { + return + } + r.reqErrMu.Lock() defer r.reqErrMu.Unlock() if err == nil { @@ -97,7 +117,15 @@ func (r *Registry) RecordRequestResult(datasourceID string, err error) { if r.reqErrors == nil { r.reqErrors = make(map[string]requestError) } - r.reqErrors[datasourceID] = requestError{msg: RedactError(err.Error()), at: time.Now()} + r.reqErrors[datasourceID] = requestError{msg: msg, at: time.Now()} +} + +// clearRequestError drops any recorded request failure for a datasource. +// Callers hold r.mu. +func (r *Registry) clearRequestError(datasourceID string) { + r.reqErrMu.Lock() + delete(r.reqErrors, datasourceID) + r.reqErrMu.Unlock() } func (r *Registry) recentRequestError(datasourceID string) (string, bool) { @@ -152,6 +180,8 @@ func (r *Registry) Register(id string, entry DatasourceEntry, p Proxy) { r.proxies[id] = p r.configs[id] = entry + // A replaced proxy is a new configuration; the old failure no longer applies. + r.clearRequestError(id) } // Remove removes and closes a proxy by datasource ID. @@ -165,6 +195,7 @@ func (r *Registry) Remove(id string) error { } delete(r.proxies, id) delete(r.configs, id) + r.clearRequestError(id) } return nil } @@ -271,7 +302,8 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth if err := ctx.Err(); err != nil { health.Status = "error" - health.Error = err.Error() + health.Error = RedactError(err.Error()) + health.ErrorClass = ErrorClass(t.cfg.ProxyType) results[idx] = health continue } @@ -386,5 +418,6 @@ func (r *Registry) CloseAll() { p.Close() // nolint:errcheck delete(r.proxies, id) delete(r.configs, id) + r.clearRequestError(id) } } diff --git a/pkg/ws/handler.go b/pkg/ws/handler.go index fc79aec..b717df3 100644 --- a/pkg/ws/handler.go +++ b/pkg/ws/handler.go @@ -173,7 +173,7 @@ func (h *Handler) handleRequest(ctx context.Context, msg []byte, requestID, data } resp, err := p.HandleRequest(ctx, &req) - h.registry.RecordRequestResult(datasourceID, err) + h.registry.RecordRequestResult(ctx, datasourceID, err) if err != nil { h.logger.Error("proxy request failed", "action", req.Action, "datasource", datasourceID, "err", err) return h.buildErrorResponse(requestID, 500, err.Error()), nil @@ -215,7 +215,7 @@ func (h *Handler) handleLegacyRequest(ctx context.Context, msg []byte, requestID } resp, err := p.HandleRequest(ctx, req) - h.registry.RecordRequestResult(datasourceID, err) + h.registry.RecordRequestResult(ctx, datasourceID, err) if err != nil { h.logger.Error("legacy proxy request failed", "action", req.Action, "datasource", datasourceID, "err", err) return h.buildErrorResponse(requestID, 500, err.Error()), nil From c3671eae6158b2ccbf2485d063bef2f278b4e93f Mon Sep 17 00:00:00 2001 From: PrashantBtkl Date: Mon, 12 Oct 2026 13:10:55 +0530 Subject: [PATCH 3/3] fix(deps): bump Go to 1.26.9 and x/net, grpc to clear govulncheck findings Co-Authored-By: Claude Sonnet 5.5 --- Dockerfile | 4 ++-- go.mod | 14 +++++++------- go.sum | 28 ++++++++++++++-------------- 3 files changed, 23 insertions(+), 23 deletions(-) diff --git a/Dockerfile b/Dockerfile index bd74f93..0030811 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,6 @@ -# golang 1.26.5: fixes the Go stdlib pair CVE-2026-39822 (HIGH, os.Root symlink +# golang 1.26.9: fixes the Go stdlib pair CVE-2026-39822 (HIGH, os.Root symlink # traversal) / CVE-2026-42505 (crypto/tls ECH) that 1.26.4 builds carry. -FROM golang:1.26.6-bookworm@sha256:116d58cbd88c1297624acc6e967a060012422bacf9930927e23fb719189c6f36 AS build-stage +FROM golang:1.26.9-bookworm@sha256:d9c68c2c51161e12fd77e4c6320687c9cd86e1af1e3ad6e6cd63ff970641453c AS build-stage # Always compile with the base image's Go, never an auto-downloaded toolchain. ENV GOTOOLCHAIN=local diff --git a/go.mod b/go.mod index bfe3f69..90d5344 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module nudgebee/forager -go 1.26.6 +go 1.26.9 require ( cloud.google.com/go/auth v0.23.1 @@ -25,8 +25,8 @@ require ( github.com/spf13/viper v1.21.0 github.com/xdg-go/scram v1.2.0 go.mongodb.org/mongo-driver/v2 v2.8.0 - golang.org/x/crypto v0.56.0 - golang.org/x/sys v0.47.0 + golang.org/x/crypto v0.57.0 + golang.org/x/sys v0.48.0 golang.org/x/time v0.15.0 google.golang.org/api v0.293.0 gopkg.in/yaml.v3 v3.0.1 @@ -113,13 +113,13 @@ require ( go.uber.org/atomic v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/exp v0.0.0-20250506013437-ce4c2cf36ca6 // indirect - golang.org/x/net v0.57.0 // indirect + golang.org/x/net v0.60.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sync v0.22.0 // indirect - golang.org/x/text v0.41.0 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/text v0.42.0 // indirect google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260807164820-c8921c73eeea // indirect - google.golang.org/grpc v1.83.1 // indirect + google.golang.org/grpc v1.83.2 // indirect google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/go.sum b/go.sum index d86c542..4c79640 100644 --- a/go.sum +++ b/go.sum @@ -279,8 +279,8 @@ go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= -golang.org/x/crypto v0.56.0 h1:GUh5Ii4J5jtcseSMiRqr1jXCNHoxjeV9Fmekc2oLy6Y= -golang.org/x/crypto v0.56.0/go.mod h1:OMW5y6CY9l38uPLmxU6l6pwcXp1obtLo3e6gT7gQR2I= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= golang.org/x/exp v0.0.0-20250506013437-ce4c2cf36ca6 h1:y5zboxd6LQAqYIhHnB48p0ByQ/GnQx2BE33L8BOHQkI= golang.org/x/exp v0.0.0-20250506013437-ce4c2cf36ca6/go.mod h1:U6Lno4MTRCDY+Ba7aCcauB9T60gsv5s4ralQzP72ZoQ= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= @@ -290,14 +290,14 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= -golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= +golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU= +golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= -golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -305,20 +305,20 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= -golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= -golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= -golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= +golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= +golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= -golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= -golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= @@ -335,8 +335,8 @@ google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7 h1: google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7/go.mod h1:KqHwBx2upmfa1XSi1WuRvC+2VGCLtooKkfmyvRbUmqA= google.golang.org/genproto/googleapis/rpc v0.0.0-20260807164820-c8921c73eeea h1:kVhQEPTpKQahD5+JSBTfBB19wcgQTTjAIn45MBqnyHk= google.golang.org/genproto/googleapis/rpc v0.0.0-20260807164820-c8921c73eeea/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.83.1 h1:HIO0+BEtBP6soyqvqC8sNUjZ7bTs+0hFQuFF+RAy++Y= -google.golang.org/grpc v1.83.1/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=