From 0ed2935f04faa11dae3accbc3b5fa2a24a58ffce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mats=20T=C3=B6rnberg?= Date: Mon, 21 Sep 2026 18:59:08 +0200 Subject: [PATCH] alot more changes --- Dockerfile | 2 +- README.md | 22 +- build.sh | 39 +- cmd/server/main.go | 49 ++ instrumentation.md | 1 + internal/abuseipdb/abuseipdb.go | 192 +++++ internal/abuseipdb/abuseipdb_test.go | 70 ++ internal/blocker/blocker.go | 92 +++ internal/collector/collector.go | 55 +- internal/collector/collector_test.go | 81 ++ internal/collector/handlers.go | 120 ++- internal/collector/issue_traces.go | 22 + internal/collector/issue_traces_test.go | 4 + internal/collector/source_ips.go | 110 +++ internal/collector/source_ips_test.go | 30 + internal/investigation/investigation.go | 509 +++++++++++++ internal/investigation/investigation_test.go | 37 + internal/investigation/store.go | 66 ++ internal/sourcemaps/store.go | 120 ++- internal/sourcemaps/store_test.go | 26 + internal/storage/clickhouse.go | 3 + internal/storage/clickhouse_traces.go | 46 ++ internal/storage/source_ips.go | 198 +++++ internal/storage/store.go | 12 + k8s/deployment.yaml | 24 + otel-ui/src/App.tsx | 2 +- otel-ui/src/components/IssuesVisualizer.tsx | 36 +- .../src/components/SourceIpsVisualizer.tsx | 693 ++++++++++++++++++ otel-ui/src/components/Tabs.tsx | 15 +- otel-ui/src/components/TraceVisualization.tsx | 28 +- otel-ui/src/components/common/DetailField.tsx | 15 +- otel-ui/src/components/common/IpButton.tsx | 99 +-- otel-ui/src/components/layout/IconRail.tsx | 2 + otel-ui/src/hooks/useDetectedIps.ts | 41 ++ otel-ui/src/hooks/useIssueSources.ts | 45 ++ otel-ui/src/hooks/useSourceIps.ts | 21 + otel-ui/src/lib/issue-utils.ts | 41 +- otel-ui/src/lib/routes.ts | 1 + otel-ui/src/lib/types.ts | 33 + scripts/upload-sourcemap.sh | 2 + 40 files changed, 2889 insertions(+), 115 deletions(-) create mode 100644 internal/abuseipdb/abuseipdb.go create mode 100644 internal/abuseipdb/abuseipdb_test.go create mode 100644 internal/collector/source_ips.go create mode 100644 internal/collector/source_ips_test.go create mode 100644 internal/investigation/investigation.go create mode 100644 internal/investigation/investigation_test.go create mode 100644 internal/investigation/store.go create mode 100644 internal/storage/source_ips.go create mode 100644 otel-ui/src/components/SourceIpsVisualizer.tsx create mode 100644 otel-ui/src/hooks/useDetectedIps.ts create mode 100644 otel-ui/src/hooks/useIssueSources.ts create mode 100644 otel-ui/src/hooks/useSourceIps.ts diff --git a/Dockerfile b/Dockerfile index ad2b5fe..f8cd707 100644 --- a/Dockerfile +++ b/Dockerfile @@ -36,7 +36,7 @@ FROM alpine:latest ARG PUBLIC_SOURCEMAPS -RUN apk --no-cache add ca-certificates +RUN apk --no-cache add ca-certificates nmap nmap-scripts # Create non-root user RUN addgroup -S appgroup && adduser -S appuser -G appgroup diff --git a/README.md b/README.md index 31ab8fe..1a64ec3 100644 --- a/README.md +++ b/README.md @@ -145,6 +145,9 @@ Errors seen in traces (spans with an error status) and logs (severity Supports `limit`. - `GET /api/issues/{id}` - Fetch a single issue, including up to 5 recent sample occurrences (timestamp, trace/span id, message). +- `GET /api/issues/{id}/sources` - Return every distinct source IP recorded + for the issue, with occurrence count and first/last seen timestamps. Requires + the ClickHouse occurrence store; returns `501` with file/in-memory storage. - `GET /api/issues/{id}/traces` - Every trace the issue occurred in, most recently hit first, with each trace's summary and its occurrences, in one request (`limit`, `before` = `nextBefore` from the previous page). @@ -161,8 +164,9 @@ Errors seen in traces (spans with an error status) and logs (severity - `GET /api/rollouts/{id}` - Fetch one normalized rollout. - `GET /api/rollouts/{id}/impact` - Return optional Prometheus error-rate points around the rollout. -- `POST /api/sourcemaps` - Upload a JavaScript source map. Public multipart upload; - the release build script sends the source-map file contents directly. +- `POST /api/sourcemaps` - Upload a JavaScript source map. Requires the shared + `OTEL_SOURCEMAP_KEY` in an `X-Sourcemap-Key` header (uploads are refused while + it is unset). Multipart upload; multipart fields are `release`, `source` (the generated bundle URL/path), and `file`. Uploaded maps are used server-side to resolve frontend stacks. @@ -286,6 +290,7 @@ npm --prefix otel-ui run build # Upload after the new UI is deployed. Use the generated URL/path exactly as # it appears in browser stack traces. curl -fS -X POST https://otel.example.com/api/sourcemaps \ + -H "X-Sourcemap-Key: $OTEL_SOURCEMAP_KEY" \ -F "release=$RELEASE" \ -F "source=/assets/index-ABC123.js" \ -F "file=@otel-ui/dist/assets/index-ABC123.js.map" @@ -330,8 +335,21 @@ to Docker or stored in the image. - `OTELLOG_PATH` - path to the persistent telemetry log file (default `otel-agent.log`). Only used when ClickHouse is not configured. +- `OTEL_SOURCEMAP_KEY` - shared key required to upload source maps. +- `OTEL_SOURCEMAP_CACHE_MB` - size budget (on-disk MB, default 64) for maps kept + parsed in memory; maps are indexed at start and parsed lazily on first use. - `OTEL_SOURCEMAP_DIR` - directory for uploaded release source maps (default `./sourcemaps`; mount persistent storage there in production). +- `OTEL_IP_INVESTIGATION_NMAP` - set to `true` to enable the authenticated, + rate-limited active service probe. It is off by default because Nmap sends + active network traffic to the investigated public IP. +- `ABUSE_KEY` - optional AbuseIPDB API key. When set, the authenticated IP + detail view can check a public address and submit a report without exposing + the key to the browser. Keep it in a secret, never in the frontend build. +- `OTEL_PPROF_ADDR` - address for the Go pprof server (default + `127.0.0.1:6060`). It serves `/debug/pprof/` separately from the public UI + port; use a port-forward or set this explicitly while diagnosing memory or + CPU usage. - `OTEL_KUBERNETES_ENABLED` - set to `false` to disable rollout observation; otherwise it activates in-cluster or `KUBECONFIG` observation when available. - `OTEL_KUBERNETES_NAMESPACE` - optional namespace restriction; empty watches diff --git a/build.sh b/build.sh index dd5dc30..72459fd 100755 --- a/build.sh +++ b/build.sh @@ -17,9 +17,36 @@ docker build \ docker push matst80/otel-debug:latest +# Shared key that only keeps strangers from uploading source maps (it is not a +# real credential). Taken from the environment, else from the cluster secret, +# else generated and stored there so the deployment can mount it. +ensure_sourcemap_key() { + if [[ -z "${OTEL_SOURCEMAP_KEY:-}" ]]; then + OTEL_SOURCEMAP_KEY="$(kubectl get secret otel-sourcemap-key -n monitoring \ + -o jsonpath='{.data.key}' 2>/dev/null | base64 -d || true)" + fi + if [[ -z "${OTEL_SOURCEMAP_KEY}" ]]; then + OTEL_SOURCEMAP_KEY="$(openssl rand -hex 24)" + fi + kubectl create secret generic otel-sourcemap-key -n monitoring \ + --from-literal=key="${OTEL_SOURCEMAP_KEY}" --dry-run=client -o yaml | kubectl apply -f - >/dev/null + export OTEL_SOURCEMAP_KEY +} +ensure_sourcemap_key + +# Roll out first: the new image requires OTEL_SOURCEMAP_KEY, so the manifest has +# to be applied before uploads, or any pod restart in between rejects them. +# Only the app manifests. Not `-f k8s/`: clickhouse.yaml carries a CHANGE_ME +# password Secret that would overwrite the live one, and auth-dev.yaml is a +# partial AuthServer that must be applied by hand with +# `kubectl apply --server-side --field-manager=otel-dev-auth -f k8s/auth-dev.yaml`. +kubectl apply -f k8s/deployment.yaml -f k8s/service.yaml -f k8s/ingress.yaml +kubectl rollout restart deployment otel-debug-receiver -n monitoring +kubectl rollout status deployment/otel-debug-receiver -n monitoring + # Docker builds intentionally omit public source maps. Upload maps from the -# exact UI build that went into the image before rolling it out; otherwise RUM -# events can arrive with a release that the resolver cannot map. +# exact UI build that went into the image right after rolling it out; until then +# RUM events can arrive with a release that the resolver cannot map. if [[ "${UPLOAD_SOURCEMAPS}" == "true" ]]; then : "${OTEL_URL:?set OTEL_URL (for example https://otel.k6n.net), or set UPLOAD_SOURCEMAPS=false}" maps_dir="$(mktemp -d)" @@ -40,12 +67,4 @@ if [[ "${UPLOAD_SOURCEMAPS}" == "true" ]]; then [[ "$found" == 1 ]] || { echo "no source maps found in UI build" >&2; exit 1; } fi -# Only the app manifests. Not `-f k8s/`: clickhouse.yaml carries a CHANGE_ME -# password Secret that would overwrite the live one, and auth-dev.yaml is a -# partial AuthServer that must be applied by hand with -# `kubectl apply --server-side --field-manager=otel-dev-auth -f k8s/auth-dev.yaml`. -kubectl apply -f k8s/deployment.yaml -f k8s/service.yaml -f k8s/ingress.yaml -kubectl rollout restart deployment otel-debug-receiver -n monitoring -kubectl rollout status deployment/otel-debug-receiver -n monitoring - kubectl logs -f deployment/otel-debug-receiver -n monitoring diff --git a/cmd/server/main.go b/cmd/server/main.go index 0522c8f..a0e9f7c 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -6,15 +6,18 @@ import ( "log" "net" "net/http" + "net/http/pprof" "os" "strings" + "git.tornberg.me/go-otel/internal/abuseipdb" "git.tornberg.me/go-otel/internal/alerts" "git.tornberg.me/go-otel/internal/auth" "git.tornberg.me/go-otel/internal/blocker" "git.tornberg.me/go-otel/internal/collector" "git.tornberg.me/go-otel/internal/geoip" "git.tornberg.me/go-otel/internal/honeypot" + "git.tornberg.me/go-otel/internal/investigation" kube "git.tornberg.me/go-otel/internal/kubernetes" "git.tornberg.me/go-otel/internal/prometheus" "git.tornberg.me/go-otel/internal/sourcemaps" @@ -213,6 +216,14 @@ func setupStore(ctx context.Context) storage.Store { } func main() { + // Keep profiling off the public UI/API listener. The dedicated listener is + // bound to loopback by default; set OTEL_PPROF_ADDR to expose it through a + // port-forward or to bind it elsewhere when diagnosing a deployment. + pprofAddr := os.Getenv("OTEL_PPROF_ADDR") + if pprofAddr == "" { + pprofAddr = "127.0.0.1:6060" + } + go startPprofServer(pprofAddr) // Configure history size (default 100, can be made configurable) historySize := 100 @@ -240,6 +251,9 @@ func main() { if sq, ok := store.(storage.StatsQuerier); ok { coll.SetStatsQuerier(sq) } + if sq, ok := store.(storage.SourceIPQuerier); ok { + coll.SetSourceIPQuerier(sq) + } wsServer := websocket.New(coll) honeypotTracker := honeypot.New(honeypot.ConfigFromEnv()) coll.SetHoneypotRecorder(honeypotTracker) @@ -263,9 +277,20 @@ func main() { log.Fatalf("failed to open geoip database: %v", err) } defer geo.Close() + var investigationStore investigation.Store + if chStore, ok := store.(*storage.ClickHouseStore); ok { + if st, err := investigation.NewClickHouseStore(context.Background(), chStore.Conn(), chStore.Database()); err != nil { + log.Printf("ip investigation results will not be persisted: %v", err) + } else { + investigationStore = st + } + } + investigator := investigation.NewHandler(investigation.ConfigFromEnv(), investigationStore) + abuse := abuseipdb.NewHandler(abuseipdb.ConfigFromEnv()) ingestKey := os.Getenv("OTEL_INGEST_KEY") coll.SetIngestKey(ingestKey) + coll.SetSourceMapKey(os.Getenv("OTEL_SOURCEMAP_KEY")) oidcAuth := auth.NewOidcAuth(ingestKey) log.Printf("auth: OIDC issuer=%s client=%s disabled=%v", oidcAuth.Issuer(), oidcAuth.ClientID(), oidcAuth.IsDisabled()) @@ -320,6 +345,10 @@ func main() { uiMux.HandleFunc("POST /api/auth/token", oidcAuth.HandleTokenExchange) uiMux.HandleFunc("GET /api/auth/config", oidcAuth.HandleConfig) uiMux.Handle("GET /api/ip/{ip}", oidcAuth.RequireAuth(http.HandlerFunc(geo.Handler))) + uiMux.Handle("POST /api/source-ips/{ip}/investigate", oidcAuth.RequireAuth(http.HandlerFunc(investigator.Start))) + uiMux.Handle("GET /api/source-ips/{ip}/investigation", oidcAuth.RequireAuth(http.HandlerFunc(investigator.Get))) + uiMux.Handle("GET /api/source-ips/{ip}/abuseipdb", oidcAuth.RequireAuth(http.HandlerFunc(abuse.Check))) + uiMux.Handle("POST /api/source-ips/{ip}/abuseipdb/report", oidcAuth.RequireAuth(http.HandlerFunc(abuse.Report))) uiMux.Handle("/api/honeypot/ips", oidcAuth.RequireAuth(http.HandlerFunc(honeypotTracker.HandleIPsForUser))) if alertRunner != nil { uiMux.Handle("GET /api/alerts", oidcAuth.RequireAuth(http.HandlerFunc(alertRunner.HandleStatus))) @@ -327,6 +356,7 @@ func main() { } if ipBlocker != nil { uiMux.Handle("GET /api/blocker/status", oidcAuth.RequireAuth(http.HandlerFunc(ipBlocker.HandleStatus))) + uiMux.Handle("POST /api/blocker/ips/{ip}", oidcAuth.RequireAuth(http.HandlerFunc(ipBlocker.HandleAdd))) } // Source-map uploads are intentionally public so the release build script // can upload maps without an OIDC session or bearer token. @@ -345,10 +375,12 @@ func main() { uiMux.Handle("/api/history/wait", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleWait))) uiMux.Handle("GET /api/traces/summaries", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleTraceSummaries))) uiMux.Handle("GET /api/issues/{id}/traces", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueTraces))) + uiMux.Handle("GET /api/issues/{id}/sources", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueSources))) uiMux.Handle("GET /api/traces/{traceId}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleTraceByID))) uiMux.Handle("/api/issues", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssues))) uiMux.Handle("POST /api/ml/test-spike", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleMLDetectorTest))) uiMux.Handle("GET /api/issues/{id}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueByID))) + uiMux.Handle("PATCH /api/issues/{id}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueByID))) uiMux.Handle("GET /api/issues/{id}/similar", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueSimilar))) uiMux.Handle("POST /api/issues/{id}/similar", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleSimilarityFeedback))) uiMux.Handle("GET /api/patterns", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandlePatterns))) @@ -358,6 +390,8 @@ func main() { uiMux.Handle("GET /api/rollouts/{id}/impact", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleRolloutImpact))) // Aggregate stats (ClickHouse-backed); return 501 with an explanatory // JSON body when only the file store is configured. + uiMux.Handle("GET /api/source-ips", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleSourceIPs))) + uiMux.Handle("GET /api/source-ips/{ip}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleSourceIPActivity))) uiMux.Handle("/api/stats/errors", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleStatsErrors))) uiMux.Handle("/api/stats/releases", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleStatsReleases))) uiMux.Handle("/api/stats/services", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleStatsServices))) @@ -414,6 +448,21 @@ func main() { log.Fatalf("failed to start HTTP server on :%s: %v", httpPort, err) } } + +func startPprofServer(addr string) { + mux := http.NewServeMux() + mux.HandleFunc("GET /debug/pprof/", pprof.Index) + mux.HandleFunc("GET /debug/pprof/cmdline", pprof.Cmdline) + mux.HandleFunc("GET /debug/pprof/profile", pprof.Profile) + mux.HandleFunc("GET /debug/pprof/symbol", pprof.Symbol) + mux.HandleFunc("GET /debug/pprof/trace", pprof.Trace) + + log.Printf("Starting pprof server on %s", addr) + if err := http.ListenAndServe(addr, mux); err != nil { + log.Printf("pprof server stopped: %v", err) + } +} + func spaHandler(root http.FileSystem) http.Handler { fileServer := http.FileServer(root) return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/instrumentation.md b/instrumentation.md index a320e5a..0d46ccf 100644 --- a/instrumentation.md +++ b/instrumentation.md @@ -139,6 +139,7 @@ stack, while `file=@...` reads the local source-map file contents: ```sh curl --fail-with-body --silent --show-error \ + -H "X-Sourcemap-Key: $OTEL_SOURCEMAP_KEY" \ -X POST "https://otel.example.com/api/sourcemaps" \ -F "release=$RELEASE" \ -F "source=/assets/app-HASH.js" \ diff --git a/internal/abuseipdb/abuseipdb.go b/internal/abuseipdb/abuseipdb.go new file mode 100644 index 0000000..0c62f16 --- /dev/null +++ b/internal/abuseipdb/abuseipdb.go @@ -0,0 +1,192 @@ +// Package abuseipdb proxies the AbuseIPDB API for authenticated operator actions. +// The API key is deliberately kept on the server; browsers only receive the +// selected IP's response data. +package abuseipdb + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/netip" + "net/url" + "os" + "strconv" + "strings" + "time" +) + +const defaultBaseURL = "https://api.abuseipdb.com/api/v2" + +type Config struct { + Key string + BaseURL string +} + +func ConfigFromEnv() Config { + base := strings.TrimRight(os.Getenv("ABUSE_API_URL"), "/") + if base == "" { + base = defaultBaseURL + } + return Config{Key: strings.TrimSpace(os.Getenv("ABUSE_KEY")), BaseURL: base} +} + +type Handler struct { + cfg Config + client *http.Client +} + +func NewHandler(cfg Config) *Handler { + return &Handler{cfg: cfg, client: &http.Client{Timeout: 10 * time.Second}} +} + +type Report struct { + ReportedAt string `json:"reportedAt,omitempty"` + Comment string `json:"comment,omitempty"` + Categories []int `json:"categories,omitempty"` + ReporterCountryCode string `json:"reporterCountryCode,omitempty"` +} + +type CheckData struct { + IPAddress string `json:"ipAddress,omitempty"` + IsPublic bool `json:"isPublic"` + IPVersion int `json:"ipVersion,omitempty"` + IsWhitelisted bool `json:"isWhitelisted"` + AbuseConfidenceScore int `json:"abuseConfidenceScore"` + CountryCode string `json:"countryCode,omitempty"` + UsageType string `json:"usageType,omitempty"` + ISP string `json:"isp,omitempty"` + Domain string `json:"domain,omitempty"` + TotalReports int `json:"totalReports"` + NumDistinctUsers int `json:"numDistinctUsers"` + LastReportedAt string `json:"lastReportedAt,omitempty"` + Reports []Report `json:"reports,omitempty"` +} + +type checkResponse struct { + Data CheckData `json:"data"` +} + +type reportResponse struct { + Data struct { + IPAddress string `json:"ipAddress,omitempty"` + AbuseConfidenceScore int `json:"abuseConfidenceScore"` + } `json:"data"` +} + +func (h *Handler) Check(w http.ResponseWriter, r *http.Request) { + ip, ok := h.publicIP(w, r) + if !ok { + return + } + query := url.Values{"ipAddress": {ip.String()}, "maxAgeInDays": {"90"}} + var out checkResponse + if err := h.request(r.Context(), http.MethodGet, "/check?"+query.Encode(), nil, &out); err != nil { + h.writeError(w, err) + return + } + h.writeJSON(w, out.Data) +} + +type reportRequest struct { + Categories []int `json:"categories"` + Comment string `json:"comment"` +} + +func (h *Handler) Report(w http.ResponseWriter, r *http.Request) { + ip, ok := h.publicIP(w, r) + if !ok { + return + } + var req reportRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 16<<10)).Decode(&req); err != nil { + http.Error(w, "invalid report request", http.StatusBadRequest) + return + } + if len(req.Categories) == 0 { + http.Error(w, "at least one category is required", http.StatusBadRequest) + return + } + if len(req.Comment) > 1000 { + http.Error(w, "comment is too long", http.StatusBadRequest) + return + } + form := url.Values{"ip": {ip.String()}, "comment": {strings.TrimSpace(req.Comment)}} + categories := make([]string, len(req.Categories)) + for i, category := range req.Categories { + if category < 1 || category > 30 { + http.Error(w, "category must be between 1 and 30", http.StatusBadRequest) + return + } + categories[i] = strconv.Itoa(category) + } + form.Set("categories", strings.Join(categories, ",")) + var out reportResponse + if err := h.request(r.Context(), http.MethodPost, "/report", strings.NewReader(form.Encode()), &out); err != nil { + h.writeError(w, err) + return + } + h.writeJSON(w, out.Data) +} + +func (h *Handler) publicIP(w http.ResponseWriter, r *http.Request) (netip.Addr, bool) { + if h.cfg.Key == "" { + http.Error(w, "AbuseIPDB is not configured (set ABUSE_KEY)", http.StatusNotImplemented) + return netip.Addr{}, false + } + ip, err := netip.ParseAddr(strings.TrimSpace(r.PathValue("ip"))) + if err != nil || !ip.IsGlobalUnicast() || ip.IsPrivate() || ip.IsLoopback() { + http.Error(w, "only public unicast IPs can be checked or reported", http.StatusBadRequest) + return netip.Addr{}, false + } + return ip, true +} + +func (h *Handler) request(ctx context.Context, method, path string, body io.Reader, dst any) error { + req, err := http.NewRequestWithContext(ctx, method, h.cfg.BaseURL+path, body) + if err != nil { + return err + } + req.Header.Set("Accept", "application/json") + req.Header.Set("Key", h.cfg.Key) + if body != nil { + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + } + resp, err := h.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + var apiErr struct { + Errors []struct { + Detail string `json:"detail"` + } `json:"errors"` + } + _ = json.NewDecoder(io.LimitReader(resp.Body, 64<<10)).Decode(&apiErr) + if len(apiErr.Errors) > 0 && apiErr.Errors[0].Detail != "" { + return fmt.Errorf("AbuseIPDB: %s", apiErr.Errors[0].Detail) + } + return fmt.Errorf("AbuseIPDB returned HTTP %d", resp.StatusCode) + } + if err := json.NewDecoder(io.LimitReader(resp.Body, 2<<20)).Decode(dst); err != nil { + return fmt.Errorf("invalid AbuseIPDB response: %w", err) + } + return nil +} + +func (h *Handler) writeJSON(w http.ResponseWriter, value any) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Cache-Control", "private, no-store") + _ = json.NewEncoder(w).Encode(value) +} + +func (h *Handler) writeError(w http.ResponseWriter, err error) { + status := http.StatusBadGateway + if errors.Is(err, context.DeadlineExceeded) { + status = http.StatusGatewayTimeout + } + http.Error(w, err.Error(), status) +} diff --git a/internal/abuseipdb/abuseipdb_test.go b/internal/abuseipdb/abuseipdb_test.go new file mode 100644 index 0000000..221fb40 --- /dev/null +++ b/internal/abuseipdb/abuseipdb_test.go @@ -0,0 +1,70 @@ +package abuseipdb + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestCheckAndReport(t *testing.T) { + var gotKey string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotKey = r.Header.Get("Key") + w.Header().Set("Content-Type", "application/json") + if r.URL.Path == "/check" { + if r.URL.Query().Get("ipAddress") != "8.8.8.8" || r.URL.Query().Get("maxAgeInDays") != "90" { + t.Fatalf("unexpected check query: %s", r.URL.RawQuery) + } + _, _ = w.Write([]byte(`{"data":{"ipAddress":"8.8.8.8","abuseConfidenceScore":75,"totalReports":4}}`)) + return + } + if r.URL.Path != "/report" || r.Method != http.MethodPost { + t.Fatalf("unexpected report request: %s %s", r.Method, r.URL.Path) + } + if err := r.ParseForm(); err != nil || r.Form.Get("ip") != "8.8.8.8" || r.Form.Get("categories") != "14,18" { + t.Fatalf("unexpected report form: %v", r.Form) + } + _, _ = w.Write([]byte(`{"data":{"ipAddress":"8.8.8.8","abuseConfidenceScore":80}}`)) + })) + defer server.Close() + h := NewHandler(Config{Key: "test-key", BaseURL: server.URL}) + + check := httptest.NewRecorder() + checkReq := httptest.NewRequest(http.MethodGet, "/api/source-ips/8.8.8.8/abuseipdb", nil).WithContext(t.Context()) + checkReq.SetPathValue("ip", "8.8.8.8") + h.Check(check, checkReq) + if check.Code != http.StatusOK || !strings.Contains(check.Body.String(), `"abuseConfidenceScore":75`) { + t.Fatalf("check response: %d %s", check.Code, check.Body.String()) + } + + report := httptest.NewRecorder() + body, _ := json.Marshal(reportRequest{Categories: []int{14, 18}, Comment: "scanner activity"}) + reportReq := httptest.NewRequest(http.MethodPost, "/api/source-ips/8.8.8.8/abuseipdb/report", strings.NewReader(string(body))).WithContext(t.Context()) + reportReq.SetPathValue("ip", "8.8.8.8") + h.Report(report, reportReq) + if report.Code != http.StatusOK || !strings.Contains(report.Body.String(), `"abuseConfidenceScore":80`) { + t.Fatalf("report response: %d %s", report.Code, report.Body.String()) + } + if gotKey != "test-key" { + t.Fatalf("key header = %q", gotKey) + } +} + +func TestRejectsPrivateAndUnconfigured(t *testing.T) { + private := httptest.NewRecorder() + privateReq := httptest.NewRequest(http.MethodGet, "/api/source-ips/192.168.1.2/abuseipdb", nil) + privateReq.SetPathValue("ip", "192.168.1.2") + NewHandler(Config{Key: "test-key", BaseURL: "http://invalid"}).Check(private, privateReq) + if private.Code != http.StatusBadRequest { + t.Fatalf("private status = %d", private.Code) + } + unconfigured := httptest.NewRecorder() + unconfiguredReq := httptest.NewRequest(http.MethodGet, "/api/source-ips/8.8.8.8/abuseipdb", nil) + unconfiguredReq.SetPathValue("ip", "8.8.8.8") + NewHandler(Config{}).Check(unconfigured, unconfiguredReq) + if unconfigured.Code != http.StatusNotImplemented { + t.Fatalf("unconfigured status = %d", unconfigured.Code) + } +} diff --git a/internal/blocker/blocker.go b/internal/blocker/blocker.go index fce7a0d..771cafc 100644 --- a/internal/blocker/blocker.go +++ b/internal/blocker/blocker.go @@ -166,6 +166,79 @@ type Status struct { Error string `json:"error,omitempty"` } +// Add manually adds one public IP to the blocker-managed set. It uses the +// same expiry, allowlist, ConfigMap ownership, and dry-run rules as automatic +// blocks. +func (b *Blocker) Add(ctx context.Context, address string) (Status, error) { + addr, err := netip.ParseAddr(strings.TrimSpace(address)) + if err != nil || honeypot.Untrackable(address) { + return Status{}, fmt.Errorf("only public unicast IPs can be blocked") + } + for _, p := range b.cfg.Allow { + if p.Contains(addr) { + return Status{}, fmt.Errorf("IP is on the operator allowlist") + } + } + + b.mu.Lock() + defer b.mu.Unlock() + now := b.now() + cms := b.client.CoreV1().ConfigMaps(b.cfg.Namespace) + cm, err := cms.Get(ctx, b.cfg.ConfigMap, metav1.GetOptions{}) + if err != nil { + return Status{}, fmt.Errorf("get configmap: %w", err) + } + previous := map[string]int64{} + if raw := cm.Annotations[AnnotationManaged]; raw != "" { + _ = json.Unmarshal([]byte(raw), &previous) + } + managed := map[string]int64{} + for ip, exp := range previous { + if exp > now.Unix() { + managed[ip] = exp + } + } + if _, exists := managed[addr.String()]; !exists && len(managed) >= b.cfg.MaxEntries { + return Status{}, fmt.Errorf("managed block list is full") + } + managed[addr.String()] = now.Add(b.cfg.BlockTTL).Unix() + + status := Status{Enabled: true, DryRun: b.cfg.DryRun, Evaluated: now, Managed: managed, LastWrite: b.lastWrite} + if b.cfg.DryRun { + b.last = status + return status, nil + } + if !b.lastWrite.IsZero() && now.Sub(b.lastWrite) < b.cfg.MinWrite { + return status, fmt.Errorf("block list was recently updated; try again later") + } + + current := splitCIDRs(cm.Data[b.cfg.Key]) + manual := make([]string, 0, len(current)) + for _, entry := range current { + if _, wasOurs := previous[entry]; !wasOurs { + manual = append(manual, entry) + } + } + desired := append(manual, managedList(managed)...) + if cm.Annotations == nil { + cm.Annotations = map[string]string{} + } + if cm.Data == nil { + cm.Data = map[string]string{} + } + state, _ := json.Marshal(managed) + cm.Annotations[AnnotationManaged] = string(state) + cm.Data[b.cfg.Key] = strings.Join(dedupSorted(desired), ",") + if _, err := cms.Update(ctx, cm, metav1.UpdateOptions{}); err != nil { + return Status{}, fmt.Errorf("update configmap: %w", err) + } + b.lastWrite, b.lastErr = now, "" + status.Wrote, status.LastWrite = true, now + b.last = status + log.Printf("blocker: manually added %s (expires %s)", addr, time.Unix(managed[addr.String()], 0).UTC().Format(time.RFC3339)) + return status, nil +} + func New(cfg Config, client kubernetes.Interface, source Source) *Blocker { d := DefaultConfig() if cfg.Interval <= 0 { @@ -350,6 +423,25 @@ func (b *Blocker) HandleStatus(w http.ResponseWriter, r *http.Request) { _ = json.NewEncoder(w).Encode(s) } +// HandleAdd serves POST /api/blocker/ips/{ip}. Register it behind +// authentication; it never permits private or allowlisted addresses. +func (b *Blocker) HandleAdd(w http.ResponseWriter, r *http.Request) { + status, err := b.Add(r.Context(), r.PathValue("ip")) + if err != nil { + code := http.StatusBadRequest + if strings.Contains(err.Error(), "recently updated") { + code = http.StatusTooManyRequests + } + http.Error(w, err.Error(), code) + return + } + w.Header().Set("Content-Type", "application/json") + if status.DryRun { + w.WriteHeader(http.StatusAccepted) + } + _ = json.NewEncoder(w).Encode(status) +} + func splitCIDRs(v string) []string { var out []string for _, p := range strings.Split(v, ",") { diff --git a/internal/collector/collector.go b/internal/collector/collector.go index f10465c..3fdc8f5 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -22,6 +22,7 @@ import ( "git.tornberg.me/go-otel/internal/websocket" "go.opentelemetry.io/collector/consumer" + "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/pdata/pmetric" "go.opentelemetry.io/collector/pdata/ptrace" @@ -42,6 +43,7 @@ type Collector struct { ingestLimiter *ipRateLimiter ingestKey string + sourceMapKey string notifyIssuesFn func() @@ -49,6 +51,7 @@ type Collector struct { // The stats endpoints return 501 when it's nil so the UI can hide those // views instead of erroring. statsQuerier storage.StatsQuerier + sourceIPs storage.SourceIPQuerier sourceMaps *sourcemaps.Resolver kubeWatcher *kube.Watcher prometheus *prometheus.Client @@ -111,6 +114,9 @@ func (c *Collector) SetIngestKey(key string) { c.ingestKey = key } +// SetSourceMapKey sets the shared key required to upload source maps. +func (c *Collector) SetSourceMapKey(key string) { c.sourceMapKey = key } + // HoneypotRecorder receives suspicious request observations extracted from // telemetry. It is intentionally a small interface so the collector does not // depend on the HTTP honeypot implementation. @@ -390,7 +396,9 @@ func (c *Collector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error { Data: buf, } - c.broadcaster.Broadcast(&entry) + if c.broadcaster != nil { + c.broadcaster.Broadcast(&entry) + } return nil } @@ -411,7 +419,9 @@ func (c *Collector) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) erro Data: buf, } - c.broadcaster.Broadcast(&entry) + if c.broadcaster != nil { + c.broadcaster.Broadcast(&entry) + } return nil } @@ -436,11 +446,50 @@ func (c *Collector) ConsumeLogs(ctx context.Context, ld plog.Logs) error { Data: buf, } - c.broadcaster.Broadcast(&entry) + if c.broadcaster != nil { + c.broadcaster.Broadcast(&entry) + } return nil } +// EmitInternalLog records a diagnostic generated by this server as an OTLP +// log. Keeping these in the same pipeline as customer telemetry makes deploy +// and ingestion problems visible in the UI's Logs view. Values are deliberately +// limited to scalar attributes so callers cannot accidentally attach payloads. +func (c *Collector) EmitInternalLog(ctx context.Context, severity plog.SeverityNumber, body string, attrs map[string]any, traceID pcommon.TraceID, spanID pcommon.SpanID) error { + logs := plog.NewLogs() + rl := logs.ResourceLogs().AppendEmpty() + rl.Resource().Attributes().PutStr("service.name", "go-otel") + rl.Resource().Attributes().PutStr("telemetry.sdk.name", "go-otel-internal") + sl := rl.ScopeLogs().AppendEmpty() + sl.Scope().SetName("git.tornberg.me/go-otel/internal") + record := sl.LogRecords().AppendEmpty() + record.SetTimestamp(pcommon.NewTimestampFromTime(time.Now())) + record.SetObservedTimestamp(pcommon.NewTimestampFromTime(time.Now())) + record.SetSeverityNumber(severity) + record.Body().SetStr(body) + if !traceID.IsEmpty() { + record.SetTraceID(traceID) + } + if !spanID.IsEmpty() { + record.SetSpanID(spanID) + } + for key, value := range attrs { + switch value := value.(type) { + case string: + record.Attributes().PutStr(key, value) + case int: + record.Attributes().PutInt(key, int64(value)) + case int64: + record.Attributes().PutInt(key, value) + case bool: + record.Attributes().PutBool(key, value) + } + } + return c.ConsumeLogs(ctx, logs) +} + // Capabilities returns the capabilities of the collector. func (c *Collector) Capabilities() consumer.Capabilities { return consumer.Capabilities{MutatesData: false} diff --git a/internal/collector/collector_test.go b/internal/collector/collector_test.go index 4f39d98..f7f7fe0 100644 --- a/internal/collector/collector_test.go +++ b/internal/collector/collector_test.go @@ -383,6 +383,21 @@ func TestHandleIssuesAndByID(t *testing.T) { t.Fatalf("expected 200 for known issue id, got %d", rec.Code) } + req = httptest.NewRequest(http.MethodPatch, "/api/issues/"+id, bytes.NewBufferString(`{"status":"resolved"}`)) + req.SetPathValue("id", id) + rec = httptest.NewRecorder() + coll.HandleIssueByID(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 when resolving known issue, got %d: %s", rec.Code, rec.Body.String()) + } + var updated map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &updated); err != nil { + t.Fatalf("failed to decode updated issue response: %v", err) + } + if updated["status"] != "resolved" { + t.Fatalf("expected resolved issue, got %+v", updated["status"]) + } + req = httptest.NewRequest(http.MethodGet, "/api/issues/missing", nil) req.SetPathValue("id", "missing") rec = httptest.NewRecorder() @@ -530,6 +545,72 @@ func TestHandleLogsIngestKey(t *testing.T) { } } +func TestHandleLogsPodGuardianIntegration(t *testing.T) { + coll := New(noopBroadcaster{}, 10, nil) + coll.SetIngestKey("ingest-secret") + + // Payload matching what fervent-franklin (k8s-pod-guardian) sends + payload := `{ + "resourceLogs": [ + { + "resource": { + "attributes": [ + { "key": "service.name", "value": { "stringValue": "payment-service" } }, + { "key": "service.environment", "value": { "stringValue": "production" } }, + { "key": "telemetry.source", "value": { "stringValue": "k8s-pod-guardian" } }, + { "key": "k8s.namespace.name", "value": { "stringValue": "default" } }, + { "key": "k8s.pod.name", "value": { "stringValue": "payment-api-6789-xyz" } }, + { "key": "k8s.container.name", "value": { "stringValue": "payment-api" } }, + { "key": "host.name", "value": { "stringValue": "worker-1" } } + ] + }, + "scopeLogs": [ + { + "scope": { "name": "k8s-pod-guardian/crashloop-harvester" }, + "logRecords": [ + { + "timeUnixNano": "1779362400000000000", + "severityNumber": 17, + "severityText": "ERROR", + "body": { + "stringValue": "panic: database connection failed: dial tcp 10.96.0.10:5432: i/o timeout" + }, + "attributes": [ + { "key": "k8s.container.restart_count", "value": { "stringValue": "5" } }, + { "key": "crashloop.reason", "value": { "stringValue": "CrashLoopBackOff" } } + ] + } + ] + } + ] + } + ] + }` + + req := httptest.NewRequest(http.MethodPost, "/v1/logs", bytes.NewReader([]byte(payload))) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Otel-Ingest-Key", "ingest-secret") + req.RemoteAddr = "192.0.2.1:12345" + + rec := httptest.NewRecorder() + coll.HandleLogs(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected status 200, got %d (body=%s)", rec.Code, rec.Body.String()) + } + + issues := coll.ListIssues(10) + if len(issues) != 1 { + t.Fatalf("expected 1 issue created from crash log, got %d", len(issues)) + } + if issues[0].Service != "payment-service" { + t.Errorf("expected issue service 'payment-service', got %q", issues[0].Service) + } + if !strings.Contains(issues[0].Title, "panic: database connection failed") { + t.Errorf("unexpected issue title: %q", issues[0].Title) + } +} + func TestHandleLogsIngestRateLimited(t *testing.T) { coll := New(noopBroadcaster{}, 10, nil) body := []byte(`{"resourceLogs":[]}`) diff --git a/internal/collector/handlers.go b/internal/collector/handlers.go index c814ff8..b71e248 100644 --- a/internal/collector/handlers.go +++ b/internal/collector/handlers.go @@ -4,7 +4,9 @@ package collector import ( "context" + "crypto/subtle" "encoding/base64" + "encoding/hex" "encoding/json" "fmt" "git.tornberg.me/go-otel/internal/sourcemaps" @@ -20,6 +22,7 @@ import ( "git.tornberg.me/go-otel/internal/issues" "git.tornberg.me/go-otel/internal/storage" "git.tornberg.me/go-otel/internal/websocket" + "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/pdata/pmetric" "go.opentelemetry.io/collector/pdata/ptrace" @@ -477,32 +480,116 @@ func encodeIssueCursor(lastSeen int64, id string) string { // HandleSourceMap accepts a JavaScript source map for a deployed release. // Multipart fields: release, source, file. func (c *Collector) HandleSourceMap(w http.ResponseWriter, r *http.Request) { + traceID, spanID := traceContextFromRequest(r) + logUpload := func(severity plog.SeverityNumber, body string, attrs map[string]any) { + if err := c.EmitInternalLog(r.Context(), severity, body, attrs, traceID, spanID); err != nil { + log.Printf("failed to emit sourcemap diagnostic: %v", err) + } + } + logUpload(plog.SeverityNumberInfo, "sourcemap upload received", map[string]any{ + "event.name": "sourcemap.upload", + "http.route": "/api/sourcemaps", + "http.request.method": r.Method, + }) if r.Method != http.MethodPost { + logUpload(plog.SeverityNumberWarn, "sourcemap upload rejected: method not allowed", map[string]any{ + "event.name": "sourcemap.upload.rejected", + "sourcemap.failure_stage": "method", + }) http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } + if !c.checkSourceMapKey(r) { + logUpload(plog.SeverityNumberWarn, "sourcemap upload rejected: invalid or missing key", map[string]any{ + "event.name": "sourcemap.upload.rejected", + "sourcemap.failure_stage": "auth", + "client.address": clientIP(r), + }) + http.Error(w, "invalid or missing sourcemap key", http.StatusUnauthorized) + return + } if c.sourceMaps == nil { + logUpload(plog.SeverityNumberError, "sourcemap upload failed: resolver is not configured", map[string]any{ + "event.name": "sourcemap.upload.failed", + "sourcemap.failure_stage": "configuration", + }) http.Error(w, "source maps are not configured", http.StatusNotImplemented) return } if err := r.ParseMultipartForm(50 << 20); err != nil { + logUpload(plog.SeverityNumberError, "sourcemap upload failed: invalid multipart body", map[string]any{ + "event.name": "sourcemap.upload.failed", + "sourcemap.failure_stage": "multipart", + "error.type": "multipart_parse", + }) http.Error(w, "invalid multipart body", http.StatusBadRequest) return } release, source := r.FormValue("release"), r.FormValue("source") - file, _, err := r.FormFile("file") + file, fileHeader, err := r.FormFile("file") if err != nil { + logUpload(plog.SeverityNumberError, "sourcemap upload failed: file is missing", map[string]any{ + "event.name": "sourcemap.upload.failed", + "sourcemap.failure_stage": "file", + "sourcemap.release": release, + "sourcemap.source": source, + "error.type": "missing_file", + }) http.Error(w, "file is required", http.StatusBadRequest) return } defer file.Close() if err := c.sourceMaps.Upload(release, source, file); err != nil { + logUpload(plog.SeverityNumberError, "sourcemap upload failed", map[string]any{ + "event.name": "sourcemap.upload.failed", + "sourcemap.failure_stage": "store", + "sourcemap.release": release, + "sourcemap.source": source, + "sourcemap.size_bytes": fileHeader.Size, + "error.type": "source_map_invalid_or_unwritable", + "error.message": err.Error(), + }) http.Error(w, err.Error(), http.StatusBadRequest) return } + logUpload(plog.SeverityNumberInfo, "sourcemap upload stored", map[string]any{ + "event.name": "sourcemap.upload.stored", + "sourcemap.release": release, + "sourcemap.source": source, + "sourcemap.size_bytes": fileHeader.Size, + }) w.WriteHeader(http.StatusCreated) } +// checkSourceMapKey authenticates a source map upload against the shared +// OTEL_SOURCEMAP_KEY (X-Sourcemap-Key header). Uploads are refused while no key +// is configured so the endpoint is never open by accident. +func (c *Collector) checkSourceMapKey(r *http.Request) bool { + if c.sourceMapKey == "" { + return false + } + got := r.Header.Get("X-Sourcemap-Key") + return subtle.ConstantTimeCompare([]byte(got), []byte(c.sourceMapKey)) == 1 +} + +// traceContextFromRequest preserves the caller's trace linkage for internal +// logs when the upload came through an instrumented client or proxy. +func traceContextFromRequest(r *http.Request) (pcommon.TraceID, pcommon.SpanID) { + var traceID pcommon.TraceID + var spanID pcommon.SpanID + parts := strings.Split(r.Header.Get("Traceparent"), "-") + if len(parts) != 4 || len(parts[1]) != 32 || len(parts[2]) != 16 { + return traceID, spanID + } + if _, err := hex.Decode(traceID[:], []byte(parts[1])); err != nil { + return pcommon.NewTraceIDEmpty(), pcommon.NewSpanIDEmpty() + } + if _, err := hex.Decode(spanID[:], []byte(parts[2])); err != nil { + return pcommon.NewTraceIDEmpty(), pcommon.NewSpanIDEmpty() + } + return traceID, spanID +} + // HandleListSourceMaps lists the release/source pairs currently loaded, so a // deploy can verify its uploads (and that they survived a restart). func (c *Collector) HandleListSourceMaps(w http.ResponseWriter, r *http.Request) { @@ -518,6 +605,18 @@ func (c *Collector) HandleListSourceMaps(w http.ResponseWriter, r *http.Request) // uploaded source maps. Body: {"items":[{"release":"...","stacktrace":"..."}]}; // the response has one resolved stacktrace per item, in order. func (c *Collector) HandleResolveStacks(w http.ResponseWriter, r *http.Request) { + traceID, spanID := traceContextFromRequest(r) + logResolve := func(severity plog.SeverityNumber, body string, attrs map[string]any) { + if err := c.EmitInternalLog(r.Context(), severity, body, attrs, traceID, spanID); err != nil { + log.Printf("failed to emit sourcemap diagnostic: %v", err) + } + } + logResolve(plog.SeverityNumberInfo, "sourcemap resolution received", map[string]any{ + "event.name": "sourcemap.resolve", + "http.route": "/api/sourcemaps/resolve", + "http.request.method": r.Method, + "sourcemap.resolver_ready": c.sourceMaps != nil, + }) var req struct { Items []struct { Release string `json:"release"` @@ -525,11 +624,17 @@ func (c *Collector) HandleResolveStacks(w http.ResponseWriter, r *http.Request) } `json:"items"` } if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&req); err != nil { + logResolve(plog.SeverityNumberError, "sourcemap resolution failed: invalid body", map[string]any{ + "event.name": "sourcemap.resolve.failed", + "sourcemap.failure_stage": "decode", + "error.type": "invalid_json", + }) http.Error(w, "invalid body", http.StatusBadRequest) return } out := make([]string, len(req.Items)) frames := make([][]sourcemaps.Frame, len(req.Items)) + statusCounts := map[string]int{} for i, it := range req.Items { out[i] = it.Stacktrace frames[i] = []sourcemaps.Frame{} @@ -537,7 +642,20 @@ func (c *Collector) HandleResolveStacks(w http.ResponseWriter, r *http.Request) out[i] = c.sourceMaps.ResolveStack(it.Release, it.Stacktrace) frames[i] = c.sourceMaps.ResolveFrames(it.Release, it.Stacktrace) } + for _, frame := range frames[i] { + statusCounts[frame.Status]++ + } } + logResolve(plog.SeverityNumberInfo, "sourcemap resolution completed", map[string]any{ + "event.name": "sourcemap.resolve.completed", + "sourcemap.item_count": len(req.Items), + "sourcemap.frame_count": statusCounts["mapped"] + statusCounts["no-release"] + statusCounts["no-map"] + statusCounts["unmapped"], + "sourcemap.mapped": statusCounts["mapped"], + "sourcemap.no_release": statusCounts["no-release"], + "sourcemap.no_map": statusCounts["no-map"], + "sourcemap.unmapped": statusCounts["unmapped"], + "sourcemap.resolver_ready": c.sourceMaps != nil, + }) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{"resolved": out, "frames": frames}) } diff --git a/internal/collector/issue_traces.go b/internal/collector/issue_traces.go index 428be86..2e6a961 100644 --- a/internal/collector/issue_traces.go +++ b/internal/collector/issue_traces.go @@ -175,6 +175,28 @@ func (c *Collector) HandleIssueTraces(w http.ResponseWriter, r *http.Request) { } } +// HandleIssueSources returns all distinct source IPs recorded for an issue. +// This is separate from /traces because trace pages intentionally bound the +// number of raw samples returned per trace. +func (c *Collector) HandleIssueSources(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if c.occStore == nil { + http.Error(w, "issue source history requires ClickHouse storage", http.StatusNotImplemented) + return + } + sources, err := c.occStore.IssueSources(r.Context(), id) + if err != nil { + log.Printf("failed to load issue sources %s: %v", id, err) + http.Error(w, "failed to load issue sources", http.StatusInternalServerError) + return + } + if sources == nil { + sources = []storage.IssueSource{} + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(sources) +} + // groupInMemory builds the same shape from the tracker's bounded samples, // newest trace first. func groupInMemory(samples []issues.Sample) ([]issueTraceJSON, []issues.Sample) { diff --git a/internal/collector/issue_traces_test.go b/internal/collector/issue_traces_test.go index da5de8a..7ab9f5f 100644 --- a/internal/collector/issue_traces_test.go +++ b/internal/collector/issue_traces_test.go @@ -34,6 +34,10 @@ func (o *occStore) IssueTraces(context.Context, string, storage.IssueTracesOptio return o.page, nil } +func (o *occStore) IssueSources(context.Context, string) ([]storage.IssueSource, error) { + return nil, nil +} + func (o *occStore) TraceEntries(context.Context, string, int) ([]websocket.HistoryEntry, error) { return nil, nil } diff --git a/internal/collector/source_ips.go b/internal/collector/source_ips.go new file mode 100644 index 0000000..a9c0b01 --- /dev/null +++ b/internal/collector/source_ips.go @@ -0,0 +1,110 @@ +package collector + +import ( + "encoding/json" + "log" + "net/http" + "net/netip" + "strings" + "time" + + "git.tornberg.me/go-otel/internal/features" + "git.tornberg.me/go-otel/internal/storage" + "git.tornberg.me/go-otel/internal/triage" + "git.tornberg.me/go-otel/internal/uaclass" +) + +// SetSourceIPQuerier enables the source address endpoints. +func (c *Collector) SetSourceIPQuerier(q storage.SourceIPQuerier) { c.sourceIPs = q } + +func (c *Collector) sourceIPsEnabled(w http.ResponseWriter) bool { + if c.sourceIPs == nil { + statsNotEnabled(w) + return false + } + return true +} + +// HandleSourceIPs lists every unique source address ever recorded. +// Query: q (substring), hours (only addresses seen within), limit. +func (c *Collector) HandleSourceIPs(w http.ResponseWriter, r *http.Request) { + if !c.sourceIPsEnabled(w) { + return + } + var since time.Time + if _, hours := parseStatsCommon(r); r.URL.Query().Get("hours") != "" { + since = time.Now().Add(-time.Duration(hours) * time.Hour) + } + ips, err := c.sourceIPs.SourceIPs(r.Context(), strings.TrimSpace(r.URL.Query().Get("q")), since, parseLimit(r, 200)) + if err != nil { + log.Printf("failed to list source ips: %v", err) + http.Error(w, "failed to list source ips", http.StatusInternalServerError) + return + } + // Judge scanning behaviour over the last week for the whole page in one + // query. A failure only drops the hints; the list itself is still useful. + addrs := make([]string, len(ips)) + for i := range ips { + addrs[i] = ips[i].IP + } + if routes, err := c.sourceIPs.SourceIPRoutes(r.Context(), addrs, time.Now().Add(-7*24*time.Hour)); err != nil { + log.Printf("failed to assess source ips: %v", err) + } else { + for i := range ips { + ips[i].BadActor, ips[i].ProbePaths = assessSourceIP(ips[i].IP, routes[ips[i].IP]) + } + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"ips": ips}) +} + +// HandleSourceIPActivity lists the recent requests made from one address. +// Query: hours (default 168), limit. +func (c *Collector) HandleSourceIPActivity(w http.ResponseWriter, r *http.Request) { + if !c.sourceIPsEnabled(w) { + return + } + addr, err := netip.ParseAddr(r.PathValue("ip")) + if err != nil { + http.Error(w, "invalid ip", http.StatusBadRequest) + return + } + _, hours := parseStatsCommon(r) + if r.URL.Query().Get("hours") == "" { + hours = 24 * 7 + } + events, err := c.sourceIPs.SourceIPActivity(r.Context(), addr.String(), time.Now().Add(-time.Duration(hours)*time.Hour), parseLimit(r, 200)) + if err != nil { + log.Printf("failed to load source ip activity: %v", err) + http.Error(w, "failed to load source ip activity", http.StatusInternalServerError) + return + } + reason, probes := assessSourceIP(addr.String(), events) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "ip": addr.String(), "hours": hours, "events": events, + // badActor is the first matching features.MarkBadActors rule over the + // returned window; probePaths is how many distinct probe paths it hit. + "badActor": reason, "probePaths": probes, + }) +} + +// assessSourceIP applies the same bad-actor rules the detector uses to one +// address's requests, so the hint in the UI never disagrees with triage. +func assessSourceIP(ip string, events []storage.SourceIPEvent) (reason string, probePaths int) { + fe := make([]features.Event, len(events)) + probes := map[string]struct{}{} + for i, e := range events { + fe[i] = features.Event{Route: e.Route, StatusCode: int(e.StatusCode), ClientIP: ip, UAClass: uaclass.Of(e.UserAgent)} + if triage.IsProbePath(e.Route) { + probes[e.Route] = struct{}{} + } + } + features.MarkBadActors(fe, features.BadActorConfig{}) + for _, e := range fe { + if e.BadActor != "" { + return e.BadActor, len(probes) + } + } + return "", len(probes) +} diff --git a/internal/collector/source_ips_test.go b/internal/collector/source_ips_test.go new file mode 100644 index 0000000..7367c03 --- /dev/null +++ b/internal/collector/source_ips_test.go @@ -0,0 +1,30 @@ +package collector + +import ( + "fmt" + "testing" + + "git.tornberg.me/go-otel/internal/storage" +) + +func TestAssessSourceIP(t *testing.T) { + ev := func(route string, status uint16, ua string) storage.SourceIPEvent { + return storage.SourceIPEvent{Route: route, StatusCode: status, UserAgent: ua} + } + browser := "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/120 Safari/537.36" + + if r, _ := assessSourceIP("1.2.3.4", []storage.SourceIPEvent{ev("/", 200, browser), ev("/cart", 200, browser)}); r != "" { + t.Errorf("normal traffic flagged as %q", r) + } + probes := []storage.SourceIPEvent{ev("/.env", 404, browser), ev("/.git/config", 404, browser), ev("/wp-admin", 404, browser)} + if r, n := assessSourceIP("1.2.3.4", probes); r != "probe-scanner" || n != 3 { + t.Errorf("probes: got %q, %d", r, n) + } + var walk []storage.SourceIPEvent + for i := 0; i < 15; i++ { + walk = append(walk, ev(fmt.Sprintf("/missing-%d", i), 404, browser)) + } + if r, _ := assessSourceIP("1.2.3.4", walk); r != "enumerator" { + t.Errorf("enumeration: got %q", r) + } +} diff --git a/internal/investigation/investigation.go b/internal/investigation/investigation.go new file mode 100644 index 0000000..9ecef56 --- /dev/null +++ b/internal/investigation/investigation.go @@ -0,0 +1,509 @@ +// Package investigation provides bounded, authenticated IP enrichment for the +// operator UI. Passive lookups use PTR and RDAP. Active probing is deliberately +// disabled unless the operator enables it explicitly. +package investigation + +import ( + "bytes" + "context" + "encoding/json" + "encoding/xml" + "fmt" + "io" + "log" + "net" + "net/http" + "net/netip" + "os" + "os/exec" + "strings" + "sync" + "time" +) + +type Config struct { + NmapEnabled bool + NmapBinary string + // NmapTimeout bounds one whole scan (OTEL_IP_INVESTIGATION_NMAP_TIMEOUT, + // a Go duration; default 5m). nmap's own --host-timeout is set below it. + NmapTimeout time.Duration +} + +func ConfigFromEnv() Config { + binary := os.Getenv("OTEL_IP_INVESTIGATION_NMAP_BINARY") + if binary == "" { + binary = "nmap" + } + timeout := 5 * time.Minute + if d, err := time.ParseDuration(os.Getenv("OTEL_IP_INVESTIGATION_NMAP_TIMEOUT")); err == nil && d >= 30*time.Second { + timeout = d + } + return Config{NmapEnabled: strings.EqualFold(os.Getenv("OTEL_IP_INVESTIGATION_NMAP"), "true"), NmapBinary: binary, NmapTimeout: timeout} +} + +type Service struct { + Port uint16 `json:"port"` + Protocol string `json:"protocol"` + State string `json:"state"` + Name string `json:"name,omitempty"` + Product string `json:"product,omitempty"` + Version string `json:"version,omitempty"` + // Scripts holds selected NSE output keyed by script id (http-title, ssl-cert). + Scripts map[string]string `json:"scripts,omitempty"` +} + +// Debug describes how an active scan ran, so the UI can explain failures. +type Debug struct { + Command string `json:"command"` + DurationMs int64 `json:"durationMs"` + ExitError string `json:"exitError,omitempty"` + TimedOut bool `json:"timedOut,omitempty"` + Stderr string `json:"stderr,omitempty"` + HostState string `json:"hostState,omitempty"` + Summary string `json:"summary,omitempty"` + Partial bool `json:"partial,omitempty"` +} + +type errorBody struct { + Error string `json:"error"` + Debug *Debug `json:"debug,omitempty"` +} + +type PassiveResult struct { + IP string `json:"ip"` + Hostnames []string `json:"hostnames,omitempty"` + RDAP *RDAP `json:"rdap,omitempty"` + Warnings []string `json:"warnings,omitempty"` + CompletedAt string `json:"completedAt"` +} + +type RDAP struct { + Name string `json:"name,omitempty"` + Handle string `json:"handle,omitempty"` + Start string `json:"start,omitempty"` + End string `json:"end,omitempty"` + Country string `json:"country,omitempty"` + Registry string `json:"registry,omitempty"` + Organization string `json:"organization,omitempty"` + AbuseEmail string `json:"abuseEmail,omitempty"` +} + +type ActiveResult struct { + IP string `json:"ip"` + Services []Service `json:"services,omitempty"` + Debug *Debug `json:"debug,omitempty"` + CompletedAt string `json:"completedAt"` +} + +// Record is one investigation of one address, as returned to the UI and +// persisted. Status is running, done or failed; only finished records are +// stored, so a restart drops in-flight jobs rather than leaving them running. +type Record struct { + IP string `json:"ip"` + Mode string `json:"mode"` + Status string `json:"status"` + StartedAt time.Time `json:"startedAt"` + FinishedAt *time.Time `json:"finishedAt,omitempty"` + Error string `json:"error,omitempty"` + Passive *PassiveResult `json:"passive,omitempty"` + Active *ActiveResult `json:"active,omitempty"` + // Debug is kept on failures, where there is no Active result to hold it. + Debug *Debug `json:"debug,omitempty"` +} + +// Store persists finished records. Nil means in-memory only. +type Store interface { + Save(ctx context.Context, rec Record) error + Load(ctx context.Context, ip string) ([]Record, error) +} + +type Handler struct { + cfg Config + client *http.Client + store Store + lastScan time.Time + mu sync.Mutex + jobs map[string]*Record // key: ip|mode; latest in-process record +} + +func NewHandler(cfg Config, store Store) *Handler { + if cfg.NmapTimeout <= 0 { + cfg.NmapTimeout = 5 * time.Minute + } + return &Handler{cfg: cfg, store: store, client: &http.Client{Timeout: 8 * time.Second}, jobs: map[string]*Record{}} +} + +func parsePublicIP(r *http.Request) (netip.Addr, bool) { + ip, err := netip.ParseAddr(strings.TrimSpace(r.PathValue("ip"))) + return ip, err == nil && ip.IsGlobalUnicast() && !ip.IsPrivate() && !ip.IsLoopback() +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Cache-Control", "private, no-store") + w.WriteHeader(code) + json.NewEncoder(w).Encode(v) +} + +// Start (POST) queues an investigation and returns immediately; poll Get. +func (h *Handler) Start(w http.ResponseWriter, r *http.Request) { + ip, ok := parsePublicIP(r) + if !ok { + writeJSON(w, http.StatusBadRequest, errorBody{Error: "only public unicast IPs can be investigated"}) + return + } + var req struct { + Mode string `json:"mode"` + } + if err := json.NewDecoder(io.LimitReader(r.Body, 256)).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errorBody{Error: "invalid investigation request"}) + return + } + if req.Mode != "passive" && req.Mode != "active" { + writeJSON(w, http.StatusBadRequest, errorBody{Error: "mode must be passive or active"}) + return + } + if req.Mode == "active" && !h.cfg.NmapEnabled { + writeJSON(w, http.StatusNotImplemented, errorBody{Error: "active investigation is disabled; set OTEL_IP_INVESTIGATION_NMAP=true"}) + return + } + key := ip.String() + "|" + req.Mode + h.mu.Lock() + if cur := h.jobs[key]; cur != nil && cur.Status == "running" { + rec := *cur + h.mu.Unlock() + writeJSON(w, http.StatusAccepted, rec) + return + } + if req.Mode == "active" && time.Since(h.lastScan) < 30*time.Second { + h.mu.Unlock() + writeJSON(w, http.StatusTooManyRequests, errorBody{Error: "active investigations are rate limited to one per 30s; retry shortly"}) + return + } + if req.Mode == "active" { + h.lastScan = time.Now() + } + rec := &Record{IP: ip.String(), Mode: req.Mode, Status: "running", StartedAt: time.Now().UTC()} + h.jobs[key] = rec + snapshot := *rec + h.mu.Unlock() + + go h.run(key, ip, rec) + writeJSON(w, http.StatusAccepted, snapshot) +} + +// Get returns the latest passive and active record for an address: running +// jobs from memory, otherwise whatever was last stored. +func (h *Handler) Get(w http.ResponseWriter, r *http.Request) { + ip, ok := parsePublicIP(r) + if !ok { + writeJSON(w, http.StatusBadRequest, errorBody{Error: "only public unicast IPs can be investigated"}) + return + } + out := map[string]Record{} + if h.store != nil { + if recs, err := h.store.Load(r.Context(), ip.String()); err != nil { + log.Printf("ip investigation: load %s: %v", ip, err) + } else { + for _, rec := range recs { + out[rec.Mode] = rec + } + } + } + h.mu.Lock() + for _, mode := range []string{"passive", "active"} { + if cur := h.jobs[ip.String()+"|"+mode]; cur != nil { + out[mode] = *cur + } + } + h.mu.Unlock() + writeJSON(w, http.StatusOK, map[string]any{"ip": ip.String(), "records": out}) +} + +func (h *Handler) run(key string, ip netip.Addr, rec *Record) { + timeout := 45 * time.Second + if rec.Mode == "active" { + timeout = h.cfg.NmapTimeout + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + var passive *PassiveResult + var active *ActiveResult + var err error + if rec.Mode == "passive" { + passive = h.passive(ctx, ip) + } else { + active, err = h.active(ctx, ip) + } + + now := time.Now().UTC() + h.mu.Lock() + rec.FinishedAt = &now + rec.Passive, rec.Active = passive, active + if err != nil { + rec.Status, rec.Error = "failed", err.Error() + if active != nil { + rec.Debug = active.Debug + } + log.Printf("ip investigation: nmap %s: %v (debug=%+v)", ip, err, rec.Debug) + } else { + rec.Status = "done" + } + final := *rec + h.mu.Unlock() + + if h.store != nil { + sctx, scancel := context.WithTimeout(context.Background(), 10*time.Second) + defer scancel() + if err := h.store.Save(sctx, final); err != nil { + log.Printf("ip investigation: save %s/%s: %v", ip, final.Mode, err) + } + } +} + +func (h *Handler) passive(ctx context.Context, ip netip.Addr) *PassiveResult { + result := &PassiveResult{IP: ip.String(), CompletedAt: time.Now().UTC().Format(time.RFC3339)} + if names, err := net.DefaultResolver.LookupAddr(ctx, ip.String()); err == nil { + for _, name := range names { + result.Hostnames = append(result.Hostnames, strings.TrimSuffix(name, ".")) + } + } else { + result.Warnings = append(result.Warnings, "reverse DNS lookup returned no name") + } + if rdap, err := h.lookupRDAP(ctx, ip); err == nil { + result.RDAP = rdap + } else { + result.Warnings = append(result.Warnings, "RDAP lookup unavailable") + } + return result +} + +// IANA's RDAP bootstrap format is an array of tuples, not an array of +// objects: [ ["prefix", ...], ["https://rdap.example/", ...] ]. +type bootstrap struct { + Services [][]json.RawMessage `json:"services"` +} + +func (b bootstrap) serviceFor(ip netip.Addr) (string, bool) { + for _, service := range b.Services { + if len(service) < 2 { + continue + } + var prefixes, urls []string + if json.Unmarshal(service[0], &prefixes) != nil || json.Unmarshal(service[1], &urls) != nil || len(urls) == 0 { + continue + } + for _, prefix := range prefixes { + p, err := netip.ParsePrefix(prefix) + if err == nil && p.Contains(ip) { + return urls[0], true + } + } + } + return "", false +} + +type rdapEntity struct { + Roles []string `json:"roles"` + VCard []any `json:"vcardArray"` + Entities []rdapEntity `json:"entities"` +} + +type rdapResponse struct { + Name string `json:"name"` + Handle string `json:"handle"` + Start string `json:"startAddress"` + End string `json:"endAddress"` + Country string `json:"country"` + Entities []rdapEntity `json:"entities"` +} + +func (h *Handler) lookupRDAP(ctx context.Context, ip netip.Addr) (*RDAP, error) { + bootstrapURL := "https://data.iana.org/rdap/ipv4.json" + if ip.Is6() { + bootstrapURL = "https://data.iana.org/rdap/ipv6.json" + } + var b bootstrap + if err := h.getJSON(ctx, bootstrapURL, &b); err != nil { + return nil, err + } + serviceURL, ok := b.serviceFor(ip) + if !ok { + return nil, fmt.Errorf("no RDAP service for %s", ip) + } + var out rdapResponse + if err := h.getJSON(ctx, strings.TrimRight(serviceURL, "/")+"/ip/"+ip.String(), &out); err != nil { + return nil, err + } + organization, abuseEmail := rdapContacts(out.Entities) + return &RDAP{ + Name: out.Name, Handle: out.Handle, Start: out.Start, End: out.End, + Country: out.Country, Registry: serviceURL, Organization: organization, AbuseEmail: abuseEmail, + }, nil + +} + +func rdapContacts(entities []rdapEntity) (organization, abuseEmail string) { + for _, entity := range entities { + fn, email := vcardFields(entity.VCard) + for _, role := range entity.Roles { + switch role { + case "registrant", "administrative", "technical": + if organization == "" { + organization = fn + } + case "abuse": + if abuseEmail == "" { + abuseEmail = email + } + } + } + if childOrg, childEmail := rdapContacts(entity.Entities); organization == "" { + organization = childOrg + if abuseEmail == "" { + abuseEmail = childEmail + } + } else if abuseEmail == "" { + _, abuseEmail = rdapContacts(entity.Entities) + } + } + return organization, abuseEmail +} + +func (h *Handler) getJSON(ctx context.Context, url string, dst any) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return err + } + req.Header.Set("Accept", "application/rdap+json, application/json") + resp, err := h.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + return fmt.Errorf("HTTP %s", resp.Status) + } + return json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(dst) +} + +// vcardFields extracts fn and email from a jCard ["vcard", [[name, params, type, value], ...]]. +func vcardFields(card []any) (fn, email string) { + if len(card) < 2 { + return + } + props, _ := card[1].([]any) + for _, p := range props { + prop, _ := p.([]any) + if len(prop) < 4 { + continue + } + name, _ := prop[0].(string) + val, _ := prop[3].(string) + switch name { + case "fn": + fn = val + case "email": + if email == "" { + email = val + } + } + } + return +} + +type nmapXML struct { + Hosts []struct { + Status struct { + State string `xml:"state,attr"` + Reason string `xml:"reason,attr"` + } `xml:"status"` + Ports struct { + ExtraPorts []struct { + State string `xml:"state,attr"` + Count int `xml:"count,attr"` + } `xml:"extraports"` + Ports []struct { + Protocol string `xml:"protocol,attr"` + Port uint16 `xml:"portid,attr"` + State struct { + Value string `xml:"state,attr"` + } `xml:"state"` + Service struct { + Name string `xml:"name,attr"` + Product string `xml:"product,attr"` + Version string `xml:"version,attr"` + } `xml:"service"` + Scripts []struct { + ID string `xml:"id,attr"` + Output string `xml:"output,attr"` + } `xml:"script"` + } `xml:"port"` + } `xml:"ports"` + } `xml:"host"` + Finished struct { + Summary string `xml:"summary,attr"` + } `xml:"runstats>finished"` +} + +func (h *Handler) active(ctx context.Context, ip netip.Addr) (*ActiveResult, error) { + // --host-timeout makes nmap stop itself and emit valid XML with whatever it + // found; the context deadline is only a backstop and stays well above it. + hostTimeout := fmt.Sprintf("%ds", int((h.cfg.NmapTimeout - 15*time.Second).Seconds())) + args := []string{"-Pn", "-n", "-T4", "--max-retries", "1", "--host-timeout", hostTimeout, "--version-light", "-sV", + "--script", "ssl-cert,http-title", "--top-ports", "100", "--open", "-oX", "-", ip.String()} + result := &ActiveResult{IP: ip.String(), Debug: &Debug{Command: h.cfg.NmapBinary + " " + strings.Join(args, " ")}} + start := time.Now() + cmd := exec.CommandContext(ctx, h.cfg.NmapBinary, args...) + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + err := cmd.Run() + result.Debug.DurationMs = time.Since(start).Milliseconds() + result.Debug.Stderr = truncate(strings.TrimSpace(stderr.String()), 2000) + result.CompletedAt = time.Now().UTC().Format(time.RFC3339) + if err != nil { + result.Debug.ExitError = err.Error() + result.Debug.TimedOut = ctx.Err() != nil + if result.Debug.TimedOut { + return result, fmt.Errorf("nmap exceeded the %s scan deadline and was killed", h.cfg.NmapTimeout) + } + return result, fmt.Errorf("nmap failed: %w", err) + } + var parsed nmapXML + if err := xml.Unmarshal(stdout.Bytes(), &parsed); err != nil { + result.Debug.ExitError = "unparseable nmap XML: " + err.Error() + return result, fmt.Errorf("could not parse nmap output: %w", err) + } + result.Debug.Summary = parsed.Finished.Summary + for _, host := range parsed.Hosts { + result.Debug.HostState = host.Status.State + if host.Status.Reason != "" { + result.Debug.HostState += " (" + host.Status.Reason + ")" + } + for _, x := range host.Ports.ExtraPorts { + result.Debug.Summary += fmt.Sprintf("; %d ports %s", x.Count, x.State) + } + for _, p := range host.Ports.Ports { + svc := Service{Port: p.Port, Protocol: p.Protocol, State: p.State.Value, Name: p.Service.Name, Product: p.Service.Product, Version: p.Service.Version} + for _, sc := range p.Scripts { + if svc.Scripts == nil { + svc.Scripts = map[string]string{} + } + svc.Scripts[sc.ID] = truncate(sc.Output, 600) + } + result.Services = append(result.Services, svc) + } + } + if strings.Contains(result.Debug.Summary, "timed out") || len(parsed.Hosts) == 0 { + result.Debug.Partial = true + } + return result, nil +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/internal/investigation/investigation_test.go b/internal/investigation/investigation_test.go new file mode 100644 index 0000000..95ad69d --- /dev/null +++ b/internal/investigation/investigation_test.go @@ -0,0 +1,37 @@ +package investigation + +import ( + "encoding/json" + "net/netip" + "testing" +) + +func TestBootstrapServiceFor(t *testing.T) { + var b bootstrap + err := json.Unmarshal([]byte(`{ + "services": [ + [["8.0.0.0/8"], ["https://rdap.example.test/"]], + [["2001:db8::/32"], ["https://rdap6.example.test/"]] + ] + }`), &b) + if err != nil { + t.Fatal(err) + } + if got, ok := b.serviceFor(netip.MustParseAddr("8.8.8.8")); !ok || got != "https://rdap.example.test/" { + t.Fatalf("IPv4 service = %q, %v", got, ok) + } + if got, ok := b.serviceFor(netip.MustParseAddr("2001:db8::1")); !ok || got != "https://rdap6.example.test/" { + t.Fatalf("IPv6 service = %q, %v", got, ok) + } +} + +func TestRdapContacts(t *testing.T) { + entities := []rdapEntity{ + {Roles: []string{"registrant"}, VCard: []any{"vcard", []any{[]any{"fn", map[string]any{}, "text", "Example Org"}}}}, + {Roles: []string{"abuse"}, VCard: []any{"vcard", []any{[]any{"email", map[string]any{}, "text", "abuse@example.test"}}}}, + } + org, email := rdapContacts(entities) + if org != "Example Org" || email != "abuse@example.test" { + t.Fatalf("contacts = %q, %q", org, email) + } +} diff --git a/internal/investigation/store.go b/internal/investigation/store.go new file mode 100644 index 0000000..ee8e2f1 --- /dev/null +++ b/internal/investigation/store.go @@ -0,0 +1,66 @@ +package investigation + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" +) + +// ClickHouseStore keeps the latest finished record per (ip, mode) as JSON. +// No TTL: what an address turned out to be is worth keeping. +type ClickHouseStore struct { + conn driver.Conn + db string +} + +func NewClickHouseStore(ctx context.Context, conn driver.Conn, db string) (*ClickHouseStore, error) { + err := conn.Exec(ctx, fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.ip_investigations ( + ip String, + mode LowCardinality(String), + status LowCardinality(String), + started_at DateTime64(3, 'UTC'), + finished_at DateTime64(3, 'UTC'), + record String + ) ENGINE = ReplacingMergeTree(finished_at) + ORDER BY (ip, mode)`, db)) + if err != nil { + return nil, err + } + return &ClickHouseStore{conn: conn, db: db}, nil +} + +func (s *ClickHouseStore) Save(ctx context.Context, rec Record) error { + body, err := json.Marshal(rec) + if err != nil { + return err + } + finished := time.Now().UTC() + if rec.FinishedAt != nil { + finished = *rec.FinishedAt + } + return s.conn.Exec(ctx, fmt.Sprintf(`INSERT INTO %s.ip_investigations (ip, mode, status, started_at, finished_at, record) VALUES (?, ?, ?, ?, ?, ?)`, s.db), + rec.IP, rec.Mode, rec.Status, rec.StartedAt.UTC(), finished, string(body)) +} + +func (s *ClickHouseStore) Load(ctx context.Context, ip string) ([]Record, error) { + rows, err := s.conn.Query(ctx, fmt.Sprintf(`SELECT record FROM %s.ip_investigations FINAL WHERE ip = ?`, s.db), ip) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Record + for rows.Next() { + var body string + if err := rows.Scan(&body); err != nil { + return nil, err + } + var rec Record + if err := json.Unmarshal([]byte(body), &rec); err == nil { + out = append(out, rec) + } + } + return out, rows.Err() +} diff --git a/internal/sourcemaps/store.go b/internal/sourcemaps/store.go index 5755ac7..f1d16b4 100644 --- a/internal/sourcemaps/store.go +++ b/internal/sourcemaps/store.go @@ -1,6 +1,7 @@ package sourcemaps import ( + "container/list" "encoding/base64" "fmt" "io" @@ -17,10 +18,30 @@ import ( var frameRE = regexp.MustCompile(`(.*?)(https?://[^ )]+|[^ )]+):(\d+):(\d+)(\)?\s*)$`) +// defaultCacheBytes bounds the on-disk size of the maps kept parsed in memory. +// A parsed map costs several times its file size, so this is a heap budget of +// roughly 4-5x. Override with OTEL_SOURCEMAP_CACHE_MB. +const defaultCacheBytes = 64 << 20 + +// Resolver indexes uploaded maps by file name at startup and parses them lazily +// on first use, keeping the most recently used ones within a size budget. A +// release ships hundreds of chunks, so parsing them all up front does not fit +// in memory. type Resolver struct { - dir string - mu sync.RWMutex - maps map[string]*sm.Consumer + dir string + + mu sync.Mutex + index map[string]string // key(release, source) -> file name + cache map[string]*list.Element + lru *list.List // front = most recently used; values are *cacheEntry + cacheBytes int64 + maxBytes int64 +} + +type cacheEntry struct { + key string + consumer *sm.Consumer + size int64 } func New(dir string) (*Resolver, error) { @@ -30,8 +51,18 @@ func New(dir string) (*Resolver, error) { if err := os.MkdirAll(dir, 0750); err != nil { return nil, err } - r := &Resolver{dir: dir, maps: make(map[string]*sm.Consumer)} - r.load() + maxBytes := int64(defaultCacheBytes) + if mb, err := strconv.Atoi(os.Getenv("OTEL_SOURCEMAP_CACHE_MB")); err == nil && mb > 0 { + maxBytes = int64(mb) << 20 + } + r := &Resolver{ + dir: dir, + index: make(map[string]string), + cache: make(map[string]*list.Element), + lru: list.New(), + maxBytes: maxBytes, + } + r.scan() return r, nil } @@ -55,8 +86,9 @@ func parseFileName(name string) (release, source string, ok bool) { return string(rel), string(src), true } -// load restores previously uploaded maps; unreadable or legacy-named files are skipped. -func (r *Resolver) load() { +// scan indexes previously uploaded maps by file name without reading them; +// legacy-named files are skipped. +func (r *Resolver) scan() { entries, err := os.ReadDir(r.dir) if err != nil { return @@ -66,15 +98,51 @@ func (r *Resolver) load() { if !ok || e.IsDir() { continue } - b, err := os.ReadFile(filepath.Join(r.dir, e.Name())) - if err != nil { - continue - } - consumer, err := sm.Parse(source+".map", b) - if err != nil { - continue - } - r.maps[key(release, source)] = consumer + r.index[key(release, source)] = e.Name() + } +} + +// get returns the parsed map for k, loading it from disk on a cache miss. It +// returns nil when no map is indexed or the file is unreadable. +func (r *Resolver) get(k string) *sm.Consumer { + r.mu.Lock() + defer r.mu.Unlock() + if el, ok := r.cache[k]; ok { + r.lru.MoveToFront(el) + return el.Value.(*cacheEntry).consumer + } + name, ok := r.index[k] + if !ok { + return nil + } + b, err := os.ReadFile(filepath.Join(r.dir, name)) + if err != nil { + return nil + } + _, source, _ := parseFileName(name) + consumer, err := sm.Parse(source+".map", b) + if err != nil { + return nil + } + r.put(k, consumer, int64(len(b))) + return consumer +} + +// put caches a parsed map and evicts least recently used ones over budget, +// always keeping the newest entry. Callers hold r.mu. +func (r *Resolver) put(k string, consumer *sm.Consumer, size int64) { + if el, ok := r.cache[k]; ok { + r.cacheBytes -= el.Value.(*cacheEntry).size + r.lru.Remove(el) + } + r.cache[k] = r.lru.PushFront(&cacheEntry{key: k, consumer: consumer, size: size}) + r.cacheBytes += size + for r.cacheBytes > r.maxBytes && r.lru.Len() > 1 { + last := r.lru.Back() + e := last.Value.(*cacheEntry) + r.lru.Remove(last) + delete(r.cache, e.key) + r.cacheBytes -= e.size } } @@ -86,10 +154,10 @@ type Entry struct { // List returns every loaded map, sorted by release then source. func (r *Resolver) List() []Entry { - r.mu.RLock() - defer r.mu.RUnlock() - out := make([]Entry, 0, len(r.maps)) - for k := range r.maps { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]Entry, 0, len(r.index)) + for k := range r.index { release, source, _ := strings.Cut(k, "\x00") out = append(out, Entry{Release: release, Source: source}) } @@ -128,8 +196,10 @@ func (r *Resolver) Upload(release, source string, body io.Reader) error { if err := os.WriteFile(name, b, 0600); err != nil { return err } + k := key(release, source) r.mu.Lock() - r.maps[key(release, source)] = consumer + r.index[k] = fileName(release, source) + r.put(k, consumer, int64(len(b))) r.mu.Unlock() return nil } @@ -149,8 +219,6 @@ func (r *Resolver) ResolveStack(release, stack string) string { if release == "" || stack == "" { return stack } - r.mu.RLock() - defer r.mu.RUnlock() lines := strings.Split(stack, "\n") for i, line := range lines { m := frameRE.FindStringSubmatch(line) @@ -159,7 +227,7 @@ func (r *Resolver) ResolveStack(release, stack string) string { } genLine, _ := strconv.Atoi(m[3]) genCol, _ := strconv.Atoi(m[4]) - consumer := r.maps[key(release, m[2])] + consumer := r.get(key(release, m[2])) if consumer == nil { continue } @@ -202,8 +270,6 @@ const frameContext = 6 // uploaded source maps, including surrounding original source when the map // embeds sourcesContent. func (r *Resolver) ResolveFrames(release, stack string) []Frame { - r.mu.RLock() - defer r.mu.RUnlock() var frames []Frame for _, line := range strings.Split(stack, "\n") { m := frameRE.FindStringSubmatch(line) @@ -214,7 +280,7 @@ func (r *Resolver) ResolveFrames(release, stack string) []Frame { col, _ := strconv.Atoi(m[4]) fn := strings.TrimSpace(strings.TrimSuffix(strings.TrimPrefix(strings.TrimSpace(m[1]), "at "), "(")) f := Frame{Raw: strings.TrimSpace(line), Function: fn, File: m[2], Line: ln, Column: col, Status: "mapped"} - switch consumer := r.maps[key(release, m[2])]; { + switch consumer := r.get(key(release, m[2])); { case release == "": f.Status = "no-release" case consumer == nil: diff --git a/internal/sourcemaps/store_test.go b/internal/sourcemaps/store_test.go index e00e6b9..0b9c0d4 100644 --- a/internal/sourcemaps/store_test.go +++ b/internal/sourcemaps/store_test.go @@ -22,3 +22,29 @@ func TestReloadFromDisk(t *testing.T) { t.Fatalf("frames = %+v", frames) } } + +func TestLazyLoadAndEviction(t *testing.T) { + dir := t.TempDir() + r, _ := New(dir) + for _, f := range []string{"a", "b", "c"} { + if err := r.Upload("rel", "/assets/"+f+".js", strings.NewReader(testMap)); err != nil { + t.Fatal(err) + } + } + r2, _ := New(dir) + if len(r2.cache) != 0 || len(r2.List()) != 3 { + t.Fatalf("startup should index without parsing: cached=%d listed=%d", len(r2.cache), len(r2.List())) + } + r2.maxBytes = int64(len(testMap)) * 2 // room for two parsed maps + for _, f := range []string{"a", "b", "c"} { + if fr := r2.ResolveFrames("rel", "at f (https://h/assets/"+f+".js:2:1)"); len(fr) != 1 || fr[0].Status != "mapped" { + t.Fatalf("%s not resolved: %+v", f, fr) + } + } + if len(r2.cache) != 2 { + t.Fatalf("cache should hold 2 entries, has %d", len(r2.cache)) + } + if r2.get(key("rel", "/assets/a.js")) == nil { // evicted, reloads from disk + t.Fatal("evicted map did not reload") + } +} diff --git a/internal/storage/clickhouse.go b/internal/storage/clickhouse.go index 10061c3..5c97713 100644 --- a/internal/storage/clickhouse.go +++ b/internal/storage/clickhouse.go @@ -156,6 +156,9 @@ func (s *ClickHouseStore) initSchema(ctx context.Context) error { if err := EnsureNormalizedSchema(ctx, s.conn, s.db); err != nil { return fmt.Errorf("normalized schema: %w", err) } + if err := EnsureSourceIPSchema(ctx, s.conn, s.db); err != nil { + return fmt.Errorf("source ip schema: %w", err) + } if err := s.conn.Exec(ctx, fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.entries ( timestamp DateTime64(9, 'UTC') NOT NULL, type LowCardinality(String) NOT NULL, diff --git a/internal/storage/clickhouse_traces.go b/internal/storage/clickhouse_traces.go index 9939bdd..dca9008 100644 --- a/internal/storage/clickhouse_traces.go +++ b/internal/storage/clickhouse_traces.go @@ -334,3 +334,49 @@ func (s *ClickHouseStore) IssueTraces(ctx context.Context, issueID string, opts } return page, nil } + +// IssueSources returns every distinct source IP recorded for an issue. Source +// addresses may be in the derived occurrence sample, or on any normalized span +// belonging to one of the issue's traces (for example, an ingress span's +// net.peer.ip when the failing child span has no client attributes). +func (s *ClickHouseStore) IssueSources(ctx context.Context, issueID string) ([]IssueSource, error) { + rows, err := s.conn.Query(ctx, fmt.Sprintf(` + SELECT ip, min(seen_at), max(seen_at), count() + FROM ( + -- Logs and older occurrences keep their source address in the + -- derived sample. Keep this path for data that has no trace. + SELECT JSONExtractString(sample, 'sourceIp') AS ip, timestamp AS seen_at + FROM %s.issue_occurrences + WHERE issue_id = ? AND ip != '' + + UNION ALL + + -- A span error's sample describes the failing span, but the public + -- peer is commonly recorded on the ingress span in the same trace. + -- Include every normalized span peer for the issue's traces. + SELECT peer_ip AS ip, timestamp AS seen_at + FROM %s.normalized_events + WHERE peer_ip != '' + AND trace_id IN ( + SELECT DISTINCT trace_id + FROM %s.issue_occurrences + WHERE issue_id = ? AND trace_id != '' + ) + ) + WHERE ip != '' + GROUP BY ip + ORDER BY count() DESC, max(seen_at) DESC`, s.db, s.db, s.db), issueID, issueID) + if err != nil { + return nil, fmt.Errorf("clickhouse issue sources: %w", err) + } + defer rows.Close() + var out []IssueSource + for rows.Next() { + var source IssueSource + if err := rows.Scan(&source.IP, &source.FirstSeen, &source.LastSeen, &source.Count); err != nil { + return nil, err + } + out = append(out, source) + } + return out, rows.Err() +} diff --git a/internal/storage/source_ips.go b/internal/storage/source_ips.go new file mode 100644 index 0000000..4659663 --- /dev/null +++ b/internal/storage/source_ips.go @@ -0,0 +1,198 @@ +package storage + +import ( + "context" + "fmt" + "time" + + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + + "git.tornberg.me/go-otel/internal/honeypot" + "git.tornberg.me/go-otel/internal/uaclass" +) + +// SourceIP is one unique client address ever seen on ingest, with lifetime +// totals. The address is the edge peer when the trace has one (the real +// client behind ingress), otherwise the span's own peer. +type SourceIP struct { + IP string `json:"ip"` + FirstSeen time.Time `json:"firstSeen"` + LastSeen time.Time `json:"lastSeen"` + Requests uint64 `json:"requests"` + Errors uint64 `json:"errors"` + Services []string `json:"services"` + UserAgent string `json:"userAgent,omitempty"` + UAClass string `json:"uaClass,omitempty"` + // Public is false for loopback/private/reserved addresses, which are + // usually a proxy hop rather than a real client. + Public bool `json:"public"` + // BadActor is the features.MarkBadActors rule the recent requests match + // ("" if none); ProbePaths counts the distinct probe paths they hit. Set by + // the caller from SourceIPRoutes, not by SourceIPs. + BadActor string `json:"badActor,omitempty"` + ProbePaths int `json:"probePaths,omitempty"` +} + +// SourceIPEvent is one request attributed to a source address. +type SourceIPEvent struct { + Timestamp time.Time `json:"timestamp"` + Service string `json:"service"` + Method string `json:"method,omitempty"` + Host string `json:"host,omitempty"` + Route string `json:"route,omitempty"` + SpanName string `json:"spanName,omitempty"` + StatusCode uint16 `json:"statusCode"` + IsError bool `json:"isError"` + TraceID string `json:"traceId"` + SpanID string `json:"spanId"` + UserAgent string `json:"userAgent,omitempty"` +} + +// SourceIPQuerier is implemented by stores that keep the global source +// address list (ClickHouseStore). +type SourceIPQuerier interface { + // SourceIPs lists unique addresses, most recently seen first. search is + // a substring match on the address; since (zero = all time) drops + // addresses last seen earlier. + SourceIPs(ctx context.Context, search string, since time.Time, limit int) ([]SourceIP, error) + // SourceIPActivity lists the requests attributed to one address, newest first. + SourceIPActivity(ctx context.Context, ip string, since time.Time, limit int) ([]SourceIPEvent, error) + // SourceIPRoutes returns, per address, the distinct (route, status, user + // agent) combinations requested since the given time - enough to judge + // scanning behaviour for many addresses in one query. + SourceIPRoutes(ctx context.Context, ips []string, since time.Time) (map[string][]SourceIPEvent, error) +} + +// sourceIPClient is the client-address expression shared by the rollup and +// the activity query, so both agree on who a request "came from". +const sourceIPClient = "if(edge_peer_ip != '', edge_peer_ip, peer_ip)" + +// sourceIPRequest keeps one row per request: a request produces several +// spans (ingress, upstream, ...) that all carry the same edge peer, so only +// the edge span itself is counted. +const sourceIPRequest = "kind = 'span' AND ip != '' AND (edge_peer_ip = '' OR span_id = edge_span_id)" + +const sourceIPSelect = `SELECT %s AS ip, min(timestamp) AS first_seen, max(timestamp) AS last_seen, + count() AS requests, sum(is_error) AS errors, groupUniqArray(service) AS services, + anyLast(user_agent) AS user_agent, anyLast(ua_class) AS ua_class + FROM %s.normalized_events WHERE %s GROUP BY ip` + +// EnsureSourceIPSchema creates the lifetime source_ips table and the +// materialized view feeding it from normalized_events. Unlike +// normalized_events it has no TTL: the list is meant to outlive raw data. +// An empty table is backfilled once from whatever normalized_events holds. +// Idempotent. +func EnsureSourceIPSchema(ctx context.Context, conn driver.Conn, db string) error { + if err := conn.Exec(ctx, fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.source_ips ( + ip String, + first_seen SimpleAggregateFunction(min, DateTime64(9, 'UTC')), + last_seen SimpleAggregateFunction(max, DateTime64(9, 'UTC')), + requests SimpleAggregateFunction(sum, UInt64), + errors SimpleAggregateFunction(sum, UInt64), + services SimpleAggregateFunction(groupUniqArrayArray, Array(String)), + user_agent SimpleAggregateFunction(anyLast, String), + ua_class SimpleAggregateFunction(anyLast, String) + ) ENGINE = AggregatingMergeTree + ORDER BY ip`, db)); err != nil { + return err + } + sel := fmt.Sprintf(sourceIPSelect, sourceIPClient, db, sourceIPRequest) + + var existing uint64 + if err := conn.QueryRow(ctx, fmt.Sprintf("SELECT count() FROM %s.source_ips", db)).Scan(&existing); err != nil { + return err + } + if existing == 0 { + if err := conn.Exec(ctx, fmt.Sprintf("INSERT INTO %s.source_ips %s", db, sel)); err != nil { + return fmt.Errorf("backfill: %w", err) + } + } + return conn.Exec(ctx, fmt.Sprintf("CREATE MATERIALIZED VIEW IF NOT EXISTS %s.source_ips_mv TO %s.source_ips AS %s", db, db, sel)) +} + +func (s *ClickHouseStore) SourceIPs(ctx context.Context, search string, since time.Time, limit int) ([]SourceIP, error) { + if limit <= 0 || limit > 1000 { + limit = 200 + } + where, args := "1", []interface{}{} + if search != "" { + where += " AND position(ip, ?) > 0" + args = append(args, search) + } + having := "" + if !since.IsZero() { + having = "HAVING last_seen >= ?" + args = append(args, since.UTC()) + } + args = append(args, limit) + rows, err := s.conn.Query(ctx, fmt.Sprintf(`SELECT ip, min(first_seen), max(last_seen) AS last_seen, sum(requests), sum(errors), + groupUniqArrayArray(services), anyLast(user_agent), anyLast(ua_class) + FROM %s.source_ips WHERE %s GROUP BY ip %s ORDER BY last_seen DESC LIMIT ?`, s.db, where, having), args...) + if err != nil { + return nil, fmt.Errorf("clickhouse source ips: %w", err) + } + defer rows.Close() + out := []SourceIP{} + for rows.Next() { + var v SourceIP + if err := rows.Scan(&v.IP, &v.FirstSeen, &v.LastSeen, &v.Requests, &v.Errors, &v.Services, &v.UserAgent, &v.UAClass); err != nil { + return nil, err + } + if v.UAClass == "" && v.UserAgent != "" { + v.UAClass = uaclass.Of(v.UserAgent) + } + v.Public = !honeypot.Untrackable(v.IP) + out = append(out, v) + } + return out, rows.Err() +} + +func (s *ClickHouseStore) SourceIPActivity(ctx context.Context, ip string, since time.Time, limit int) ([]SourceIPEvent, error) { + if limit <= 0 || limit > 1000 { + limit = 200 + } + if since.IsZero() { + since = time.Now().Add(-7 * 24 * time.Hour) + } + rows, err := s.conn.Query(ctx, fmt.Sprintf(`SELECT timestamp, service, method, host, route, span_name, status_code, is_error, trace_id, span_id, user_agent + FROM (SELECT *, %s AS ip FROM %s.normalized_events WHERE timestamp >= ?) + WHERE %s AND ip = ? ORDER BY timestamp DESC LIMIT ?`, sourceIPClient, s.db, sourceIPRequest), since.UTC(), ip, limit) + if err != nil { + return nil, fmt.Errorf("clickhouse source ip activity: %w", err) + } + defer rows.Close() + out := []SourceIPEvent{} + for rows.Next() { + var v SourceIPEvent + var isErr uint8 + if err := rows.Scan(&v.Timestamp, &v.Service, &v.Method, &v.Host, &v.Route, &v.SpanName, &v.StatusCode, &isErr, &v.TraceID, &v.SpanID, &v.UserAgent); err != nil { + return nil, err + } + v.IsError = isErr != 0 + out = append(out, v) + } + return out, rows.Err() +} + +func (s *ClickHouseStore) SourceIPRoutes(ctx context.Context, ips []string, since time.Time) (map[string][]SourceIPEvent, error) { + out := map[string][]SourceIPEvent{} + if len(ips) == 0 { + return out, nil + } + rows, err := s.conn.Query(ctx, fmt.Sprintf(`SELECT ip, route, status_code, anyLast(user_agent) + FROM (SELECT *, %s AS ip FROM %s.normalized_events WHERE timestamp >= ? AND kind = 'span') + WHERE %s AND ip IN ? GROUP BY ip, route, status_code`, sourceIPClient, s.db, sourceIPRequest), since.UTC(), ips) + if err != nil { + return nil, fmt.Errorf("clickhouse source ip routes: %w", err) + } + defer rows.Close() + for rows.Next() { + var ip string + var v SourceIPEvent + if err := rows.Scan(&ip, &v.Route, &v.StatusCode, &v.UserAgent); err != nil { + return nil, err + } + out[ip] = append(out[ip], v) + } + return out, rows.Err() +} diff --git a/internal/storage/store.go b/internal/storage/store.go index c2565b8..8824982 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -135,4 +135,16 @@ type IssueTracesPage struct { type OccurrenceStore interface { AppendOccurrence(ctx context.Context, occ Occurrence) error IssueTraces(ctx context.Context, issueID string, opts IssueTracesOptions) (IssueTracesPage, error) + IssueSources(ctx context.Context, issueID string) ([]IssueSource, error) +} + +// IssueSource is one source address observed for an issue across all persisted +// occurrences. It is deliberately an aggregate: the issue detail already +// has the recent occurrence samples, while this answers "which sources ever +// produced this issue?" without returning every raw occurrence. +type IssueSource struct { + IP string `json:"ip"` + FirstSeen time.Time `json:"firstSeen"` + LastSeen time.Time `json:"lastSeen"` + Count uint64 `json:"count"` } diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 10d0796..c9b2fa0 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -40,6 +40,8 @@ spec: name: grpc - containerPort: 4318 name: otlp-http + - containerPort: 6060 + name: pprof env: - name: GRPC_GO_LOG_SEVERITY_LEVEL value: "info" @@ -47,6 +49,21 @@ spec: value: "/logs/otel-agent.log" - name: OTEL_SOURCEMAP_DIR value: "/sourcemaps" + # Shared key required to POST /api/sourcemaps (X-Sourcemap-Key header); + # uploads are refused while it is unset. + - name: OTEL_SOURCEMAP_KEY + valueFrom: + secretKeyRef: + name: otel-sourcemap-key + key: key + # AbuseIPDB API key for authenticated IP checks/reports from the UI. + # Create out of band: kubectl create secret generic abuseipdb-key -n monitoring --from-literal=key=... + - name: ABUSE_KEY + valueFrom: + secretKeyRef: + name: abuseipdb-key + key: key + optional: true - name: OTEL_OIDC_ISSUER value: "https://users.k6n.net" - name: OTEL_OIDC_CLIENT_ID @@ -127,6 +144,13 @@ spec: # service account below. Leave Prometheus unset when unavailable. - name: OTEL_KUBERNETES_NAMESPACE value: "" + # Enable the explicitly requested, rate-limited Nmap service probe. + # This is kept opt-in because it generates active traffic to external IPs. + - name: OTEL_IP_INVESTIGATION_NMAP + value: "true" + # Whole-scan budget; scans run in the background and are stored. + - name: OTEL_IP_INVESTIGATION_NMAP_TIMEOUT + value: "5m" # - name: OTEL_PROMETHEUS_URL # value: "http://prometheus.monitoring.svc.cluster.local:9090" # Private token for an ingress-side synchronizer. If absent, the diff --git a/otel-ui/src/App.tsx b/otel-ui/src/App.tsx index bb6be85..73dd4e4 100644 --- a/otel-ui/src/App.tsx +++ b/otel-ui/src/App.tsx @@ -74,7 +74,7 @@ function App() { // The redesigned screens (issues, logs, RUM sessions, rollouts, metrics) own // their own filter bar and cards, so the legacy anomaly banner and volume // waveform above them would just duplicate that UI in a different visual language. - const isTriageOverview = activeTab === "tab0" || activeTab === "tab3" || activeTab === "tab7" || activeTab === "tab9" || activeTab === "tab2"; + const isTriageOverview = activeTab === "tab0" || activeTab === "tab3" || activeTab === "tab7" || activeTab === "tab9" || activeTab === "tab10" || activeTab === "tab2"; if (isCallback) { return ; diff --git a/otel-ui/src/components/IssuesVisualizer.tsx b/otel-ui/src/components/IssuesVisualizer.tsx index f2a93eb..888dc0b 100644 --- a/otel-ui/src/components/IssuesVisualizer.tsx +++ b/otel-ui/src/components/IssuesVisualizer.tsx @@ -21,6 +21,7 @@ import { X, } from "lucide-react" import { useObservabilityData } from "../hooks/useObservabilityData" +import { useIssueSources } from "../hooks/useIssueSources" import { authFetch } from "../lib/auth" import { formatRelativeTime } from "../lib/timeUtils" import { @@ -147,7 +148,7 @@ export const toneLine: Record = { /** The sparkline + expected-band + breach segment described by the anomaly presentation spec. */ export const IssueSparkline = ({ issue }: { issue: Issue }) => { - const { points, max, reference, breachIndex } = sparklineSeries(issue, 24) + const { points, hours, max, reference, breachIndex } = sparklineSeries(issue, 24) const width = 100 const height = 100 const step = width / (points.length - 1) @@ -167,8 +168,8 @@ export const IssueSparkline = ({ issue }: { issue: Issue }) => { const bandHeight = Math.max(height - refY - 4, 6) const description = hasBreach - ? `Hourly occurrence count over the last 24 hours, climbing above the period average roughly ${points.length - breachIndex} hours ago and still elevated` - : `Hourly occurrence count over the last 24 hours, staying within the period average` + ? `Hourly occurrence count over an adaptive ${hours}-hour activity window, climbing above the period average roughly ${points.length - breachIndex} hours ago and still elevated` + : `Hourly occurrence count over an adaptive ${hours}-hour activity window, staying within the period average` return ( (null) const [needleLoading, setNeedleLoading] = useState(false) + const { sources, loading: sourcesLoading, error: sourcesError } = useIssueSources(issue.id, expanded) const handleNeedleTriage = async () => { setNeedleLoading(true) @@ -545,6 +547,34 @@ const IssueCard = ({ )} +
+
+

+ All sources ({sources.length}) +

+ {sourcesLoading && Loading…} +
+ {sourcesError &&

Failed to load sources: {sourcesError}

} + {!sourcesLoading && !sourcesError && sources.length === 0 && ( +

No source IP was recorded for these occurrences.

+ )} + {sources.length > 0 && ( +
+ {sources.map((source) => ( +
+ + + + ×{source.count} + + {new Date(source.firstSeen).toLocaleString()} → {new Date(source.lastSeen).toLocaleString()} + +
+ ))} +
+ )} +
+

Recent occurrences ({issue.samples.length}) diff --git a/otel-ui/src/components/SourceIpsVisualizer.tsx b/otel-ui/src/components/SourceIpsVisualizer.tsx new file mode 100644 index 0000000..2e1cb76 --- /dev/null +++ b/otel-ui/src/components/SourceIpsVisualizer.tsx @@ -0,0 +1,693 @@ +import { useEffect, useMemo, useState } from "react" +import useSWR from "swr" +import { Fingerprint, RefreshCw, ShieldAlert, Search, Radio, ShieldBan, ShieldCheck, Flag } from "lucide-react" +import { useSourceIps, useSourceIpActivity } from "../hooks/useSourceIps" +import { useDetectedIps, type DetectionKind } from "../hooks/useDetectedIps" +import { useObservabilityData } from "../hooks/useObservabilityData" +import { formatRelativeTime } from "../lib/timeUtils" +import { uaClassLabel } from "../lib/ua-class" +import { IpInfoPanel } from "./common/IpButton" +import { Segmented } from "./common/Segmented" +import { useSubRoute } from "../lib/routes" +import type { BadActorReason, SourceIp } from "../lib/types" +import { FilterBar, type FilterChip } from "./FilterBar" +import { MasterDetail } from "./MasterDetail" +import { DetailField, SectionLabel } from "./common/DetailField" +import { authFetch, authSWRFetcher, tokens } from "../lib/auth" + +type Scope = "public" | "flagged" | "all" +type SortMode = "recent" | "requests" | "errors" | "first" + +const sorters: Record number> = { + recent: (a, b) => ms(b.lastSeen) - ms(a.lastSeen), + requests: (a, b) => b.requests - a.requests || ms(b.lastSeen) - ms(a.lastSeen), + errors: (a, b) => b.errors - a.errors || ms(b.lastSeen) - ms(a.lastSeen), + first: (a, b) => ms(b.firstSeen) - ms(a.firstSeen), +} +const sortLabel: Record = { recent: "latest first", requests: "most requests", errors: "most errors", first: "newest addresses" } + +const SEEN_WINDOWS: Record = { + "1h": { label: "last hour", ms: 3600_000 }, + "24h": { label: "last 24 hours", ms: 24 * 3600_000 }, + "7d": { label: "last 7 days", ms: 7 * 24 * 3600_000 }, +} + +const detectionLabel: Record = { probing: "probing", "block-candidate": "block candidate", blocked: "blocked" } +const reasonLabel: Record, string> = { + "probe-scanner": "Probe scanner: hit several scanner paths", + enumerator: "Enumerator: walked many routes that do not exist", + "bot-4xx": "Self-declared bot getting client errors", +} + +/** Critical only for real detections; a hint, not a verdict. */ +const BadActorBadge = ({ label }: { label: string }) => ( + + +) + +const ms = (iso: string) => Date.parse(iso) +const dateTime = (iso: string) => + new Date(iso).toLocaleString([], { month: "short", day: "numeric", hour: "2-digit", minute: "2-digit", second: "2-digit" }) + +interface InvestigationRecord { + mode: "passive" | "active" + status: "running" | "done" | "failed" + startedAt: string + finishedAt?: string + error?: string + debug?: ScanDebug + passive?: { + hostnames?: string[] + rdap?: { name?: string; handle?: string; start?: string; end?: string; country?: string; registry?: string; organization?: string; abuseEmail?: string } + warnings?: string[] + completedAt: string + } + active?: { + services?: { port: number; protocol: string; state: string; name?: string; product?: string; version?: string; scripts?: Record }[] + debug?: ScanDebug + completedAt: string + } +} + +interface ScanDebug { + command: string + durationMs: number + exitError?: string + timedOut?: boolean + stderr?: string + hostState?: string + summary?: string + partial?: boolean +} + +const DebugPanel = ({ debug }: { debug: ScanDebug }) => ( +

+ Scan details · {(debug.durationMs / 1000).toFixed(1)}s{debug.hostState ? ` · host ${debug.hostState}` : ""} +
+
command:
{debug.command}
+ {debug.summary &&
nmap:
{debug.summary}
} + {debug.exitError &&
exit:
{debug.exitError}{debug.timedOut ? " (deadline)" : ""}
} + {debug.stderr &&
stderr:
{debug.stderr}
} +
+
+) + +const InvestigationActions = ({ ip }: { ip: string }) => { + const [starting, setStarting] = useState<"passive" | "active" | null>(null) + const [error, setError] = useState(null) + const [errorDebug, setErrorDebug] = useState(null) + const [, setNow] = useState(0) + + // Stored results load with the panel; while a job is running we poll. + const key = tokens() ? `/api/source-ips/${encodeURIComponent(ip)}/investigation` : null + const { data, mutate } = useSWR<{ records: Partial> }>(key, authSWRFetcher, { + revalidateOnFocus: false, + shouldRetryOnError: false, + refreshInterval: (d) => (d && Object.values(d.records ?? {}).some((r) => r?.status === "running") ? 3000 : 0), + }) + const passiveRec = data?.records?.passive + const activeRec = data?.records?.active + const running = (m: "passive" | "active") => data?.records?.[m]?.status === "running" || starting === m + const anyRunning = running("passive") || running("active") + + useEffect(() => { + if (!activeRec || activeRec.status !== "running") return + const t = setInterval(() => setNow((n) => n + 1), 1000) + return () => clearInterval(t) + }, [activeRec]) + + const investigate = async (mode: "passive" | "active") => { + setStarting(mode) + setError(null) + setErrorDebug(null) + try { + const response = await authFetch(`/api/source-ips/${encodeURIComponent(ip)}/investigate`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ mode }), + }) + if (!response.ok) { + const message = await response.text() + let parsed: { error?: string } | null = null + try { parsed = JSON.parse(message) } catch { /* plain text */ } + throw new Error(parsed?.error || message || `HTTP ${response.status}`) + } + await mutate() + } catch (e) { + setError(e instanceof Error ? e.message : String(e)) + } finally { + setStarting(null) + } + } + + const result = { passive: passiveRec?.passive, active: activeRec?.active } + const failed = [passiveRec, activeRec].filter((r): r is InvestigationRecord => r?.status === "failed") + const elapsed = (r?: InvestigationRecord) => (r ? Math.max(0, Math.round((Date.now() - new Date(r.startedAt).getTime()) / 1000)) : 0) + const passive = result.passive + const active = result.active + const rdap = passive?.rdap + return ( +
+ Investigate +
+ + +
+

Probing runs in the background (up to a few minutes) and results are saved. It is active network traffic and only available when enabled by the operator.

+ {error &&

Investigation failed: {error}

} + {errorDebug && } + {failed.map((r) => ( +
+

{r.mode === "active" ? "Probe" : "Lookup"} failed: {r.error}

+ {r.debug && } +
+ ))} + {(passiveRec?.finishedAt || activeRec?.finishedAt) && ( +

+ Saved results: {[passiveRec, activeRec].filter((r) => r?.status === "done" && r.finishedAt).map((r) => `${r!.mode} ${dateTime(r!.finishedAt!)}`).join(" · ") || "none successful"} +

+ )} + {passive && ( +
+ + + + + + + +
+ )} + {passive?.warnings?.map((warning) =>

{warning}

)} + {active && ( +
+

Open services

+ {active.services?.length ? ( +
    + {active.services.map((service) => ( +
  • + {service.port}/{service.protocol} · {service.name || "unknown"}{service.product ? ` · ${service.product}` : ""}{service.version ? ` ${service.version}` : ""} +
    • + ))} +
+ ) :

No open services found in the top 100 ports{active.debug?.hostState ? ` (host ${active.debug.hostState})` : ""}.

} + {active.services?.some((sv) => sv.scripts) && ( +
    + {active.services.flatMap((sv) => Object.entries(sv.scripts ?? {}).map(([id, out]) => ( +
  • {sv.port} {id}: {out}
    • + )))} +
+ )} + {active.debug && } +
+ )} +
+ ) +} + +const BlockIpAction = ({ ip, alreadyBlocked }: { ip: string; alreadyBlocked: boolean }) => { + const [loading, setLoading] = useState(false) + const [message, setMessage] = useState(null) + const [error, setError] = useState(null) + + const addToBlockList = async () => { + setLoading(true) + setMessage(null) + setError(null) + try { + const response = await authFetch(`/api/blocker/ips/${encodeURIComponent(ip)}`, { method: "POST" }) + const body = await response.text() + if (!response.ok && response.status !== 202) throw new Error(body || `HTTP ${response.status}`) + const status = body ? (JSON.parse(body) as { dryRun?: boolean }) : {} + setMessage(status.dryRun ? "Would be added (blocker is in dry-run mode)." : "Added to the block list.") + } catch (e) { + setError(e instanceof Error ? e.message : String(e)) + } finally { + setLoading(false) + } + } + + return ( +
+ Blocking + +

Blocks expire automatically according to the server policy.

+ {message &&

{message}

} + {error &&

Block failed: {error}

} +
+ ) +} + +interface AbuseCheck { + ipAddress?: string + isWhitelisted?: boolean + abuseConfidenceScore?: number + countryCode?: string + usageType?: string + isp?: string + domain?: string + totalReports?: number + numDistinctUsers?: number + lastReportedAt?: string +} + +interface AbuseReportResult { + ipAddress?: string + abuseConfidenceScore?: number +} + +const abuseCategories = [ + [14, "Port scan"], + [15, "Hacking"], + [18, "Brute-force"], + [19, "Bad web bot"], + [20, "Exploited host"], + [21, "Web app attack"], + [22, "SSH"], + [23, "IoT targeted"], +] as const + +const AbuseIpDbActions = ({ ip }: { ip: string }) => { + const [checking, setChecking] = useState(false) + const [check, setCheck] = useState(null) + const [category, setCategory] = useState("14") + const [comment, setComment] = useState("") + const [reporting, setReporting] = useState(false) + const [message, setMessage] = useState(null) + const [error, setError] = useState(null) + + const runCheck = async () => { + setChecking(true) + setError(null) + setMessage(null) + try { + const response = await authFetch(`/api/source-ips/${encodeURIComponent(ip)}/abuseipdb`) + const body = await response.text() + if (!response.ok) throw new Error(body || `HTTP ${response.status}`) + setCheck(JSON.parse(body) as AbuseCheck) + } catch (e) { + setError(e instanceof Error ? e.message : String(e)) + } finally { + setChecking(false) + } + } + + const report = async () => { + setReporting(true) + setError(null) + setMessage(null) + try { + const response = await authFetch(`/api/source-ips/${encodeURIComponent(ip)}/abuseipdb/report`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ categories: [Number(category)], comment }), + }) + const body = await response.text() + if (!response.ok) throw new Error(body || `HTTP ${response.status}`) + const result = JSON.parse(body) as AbuseReportResult + setMessage(`Reported${result.abuseConfidenceScore !== undefined ? ` · confidence now ${result.abuseConfidenceScore}%` : ""}.`) + setComment("") + void runCheck() + } catch (e) { + setError(e instanceof Error ? e.message : String(e)) + } finally { + setReporting(false) + } + } + + return ( +
+ AbuseIPDB +
+ +
+

Checks use the last 90 days. Reporting sends this address and your comment to AbuseIPDB.

+ {check && ( +
+ + + + + + + {check.isWhitelisted && } +
+ )} +
+ +