CSP violation reports: ingest endpoint, ClickHouse storage, UI view
Build and Publish / Metadata (push) Successful in 9s
Build and Publish / BuildAndDeployAmd64 (push) Successful in 2m3s

- POST /v1/csp collects browser CSP reports in both wire formats (legacy
  application/csp-report and Reporting API application/reports+json), with
  CORS limited to the operator's domains (any subdomain of k6n.net,
  tornberg.me, knatofs.se) and preflight handling.
- Reports are stored in a dedicated typed ClickHouse table (csp_reports,
  90-day TTL) with a JSONL file fallback (OTEL_CSP_PATH).
- New authenticated view endpoints /api/csp/reports and /api/csp/summary
  plus a 'CSP reports' screen in the UI (tab 12) with time window, site and
  directive filters and report drill-down.
- Documented in instrumentation.md; unit tests for parsing/handlers/file
  store and an env-gated ClickHouse integration test.
This commit is contained in:
developer committed 2026-09-23 16:50:43 +00:00
1 parent 9685de64da
commit 36f00d49b9
16 files changed
+1424 -1

No files matched your search

+1
View File
@@ -47,6 +47,7 @@ type Collector struct {
ingestKey string
todoStore todos.Store
sourceMapKey string
cspStore storage.CSPStore
notifyIssuesFn func()
+257
View File
@@ -0,0 +1,257 @@
package collector
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"time"
"git.tornberg.me/go-otel/internal/storage"
)
const (
// CSP reports are small; URLs and the echoed policy are the long parts.
maxCSPBody = 16 << 10
maxCSPURL = 512
maxCSPPolicy = 2048
maxCSPSample = 300
maxCSPBatch = 50
defaultCSPWindow = 24
)
// SetCSPStore enables CSP violation report collection. Both the ClickHouse
// store and the file fallback satisfy the interface; nil disables the
// endpoints.
func (c *Collector) SetCSPStore(s storage.CSPStore) { c.cspStore = s }
// cspReportBody mirrors the legacy report-uri payload
// (Content-Type: application/csp-report).
type cspReportBody struct {
CSPReport struct {
DocumentURI string `json:"document-uri"`
Referrer string `json:"referrer"`
BlockedURI string `json:"blocked-uri"`
ViolatedDirective string `json:"violated-directive"`
EffectiveDirective string `json:"effective-directive"`
OriginalPolicy string `json:"original-policy"`
Disposition string `json:"disposition"`
StatusCode int `json:"status-code"`
ScriptSample string `json:"script-sample"`
} `json:"csp-report"`
}
// cspAPIReport mirrors one entry of the Reporting API payload
// (Content-Type: application/reports+json); body repeats the csp-report
// fields in camelCase.
type cspAPIReport struct {
Type string `json:"type"`
UserAgent string `json:"user_agent"`
Body struct {
DocumentURL string `json:"documentURL"`
Referrer string `json:"referrer"`
BlockedURL string `json:"blockedURL"`
ViolatedDirective string `json:"violatedDirective"`
EffectiveDirective string `json:"effectiveDirective"`
OriginalPolicy string `json:"originalPolicy"`
Disposition string `json:"disposition"`
StatusCode int `json:"statusCode"`
Sample string `json:"sample"`
} `json:"body"`
}
func clipCSP(s string, max int) string {
if len(s) > max {
return s[:max]
}
return s
}
// parseCSPReports decodes either wire format into normalized reports; an
// unrecognized shape returns nil (the handler rejects those).
func parseCSPReports(body []byte) []storage.CSPReport {
var reports []storage.CSPReport
if len(body) > 0 && body[0] == '[' {
var api []cspAPIReport
if err := json.Unmarshal(body, &api); err != nil || len(api) == 0 {
return nil
}
if len(api) > maxCSPBatch {
api = api[:maxCSPBatch]
}
for _, r := range api {
// The endpoint only serves CSP; skip other report types instead
// of guessing at their bodies, and drop empty entries.
if r.Type != "" && r.Type != "csp-violation" {
continue
}
if r.Body.DocumentURL == "" && r.Body.BlockedURL == "" && r.Body.EffectiveDirective == "" &&
r.Body.ViolatedDirective == "" && r.Body.Sample == "" {
continue
}
reports = append(reports, storage.CSPReport{
DocumentURL: clipCSP(r.Body.DocumentURL, maxCSPURL),
Referrer: clipCSP(r.Body.Referrer, maxCSPURL),
BlockedURL: clipCSP(r.Body.BlockedURL, maxCSPURL),
ViolatedDirective: clipCSP(r.Body.ViolatedDirective, maxCSPSample),
EffectiveDirective: clipCSP(r.Body.EffectiveDirective, maxCSPSample),
OriginalPolicy: clipCSP(r.Body.OriginalPolicy, maxCSPPolicy),
Disposition: clipCSP(r.Body.Disposition, 32),
StatusCode: r.Body.StatusCode,
Sample: clipCSP(r.Body.Sample, maxCSPSample),
ReportType: clipCSP(r.Type, 64),
UserAgent: clipCSP(r.UserAgent, maxCSPSample),
})
}
return reports
}
var legacy cspReportBody
if err := json.Unmarshal(body, &legacy); err != nil {
return nil
}
v := legacy.CSPReport
if v.DocumentURI == "" && v.BlockedURI == "" && v.ViolatedDirective == "" {
return nil
}
return []storage.CSPReport{{
DocumentURL: clipCSP(v.DocumentURI, maxCSPURL),
Referrer: clipCSP(v.Referrer, maxCSPURL),
BlockedURL: clipCSP(v.BlockedURI, maxCSPURL),
ViolatedDirective: clipCSP(v.ViolatedDirective, maxCSPSample),
EffectiveDirective: clipCSP(v.EffectiveDirective, maxCSPSample),
OriginalPolicy: clipCSP(v.OriginalPolicy, maxCSPPolicy),
Disposition: clipCSP(v.Disposition, 32),
StatusCode: v.StatusCode,
Sample: clipCSP(v.ScriptSample, maxCSPSample),
}}
}
// finishCSPReport derives site/blocked host and a stable content id, and
// stamps server-side context (time, peer IP, request UA when the report
// carries none).
func finishCSPReport(r storage.CSPReport, now time.Time, peerIP, requestUA string) storage.CSPReport {
if r.Site = storage.URLLowerHost(r.DocumentURL); r.Site == "" {
r.Site = "unknown"
}
if r.ReportType == "" {
r.ReportType = "csp-report"
}
if r.UserAgent == "" {
r.UserAgent = clipCSP(requestUA, maxCSPSample)
}
r.SourceIP = peerIP
r.Timestamp = now.UTC()
// Deterministic id over the report content: a stable UI key that also
// makes duplicate re-deliveries of one violation identifiable.
hash := sha256.Sum256([]byte(fmt.Sprintf("%s|%s|%s|%s|%s|%s",
r.Site, r.DocumentURL, r.BlockedURL, r.EffectiveDirective, r.Sample, r.Timestamp.Format(time.RFC3339Nano))))
r.ID = hex.EncodeToString(hash[:8])
return r
}
// HandleCSPIngest collects Content-Security-Policy violation reports posted
// by browsers (report-uri / report-to). Both the legacy application/csp-report
// object and the Reporting API's application/reports+json array are accepted.
// Browsers cannot attach auth headers, so like /v1/vitals the optional ingest
// key is accepted as ?key= on the configured reporting URL; CORS is limited
// to the operator's own domains (withReportingCORS in cmd/server).
func (c *Collector) HandleCSPIngest(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if !c.checkIngest(w, r) {
return
}
if c.cspStore == nil {
http.Error(w, "csp report storage not configured", http.StatusServiceUnavailable)
return
}
// Browsers post application/csp-report or application/reports+json; some
// proxies rewrite the type, so decode regardless of content type.
raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxCSPBody))
if err != nil {
http.Error(w, "invalid csp report payload", http.StatusBadRequest)
return
}
reports := parseCSPReports(raw)
if len(reports) == 0 {
http.Error(w, "no csp reports in payload", http.StatusBadRequest)
return
}
now := time.Now()
peerIP := clientIP(r)
requestUA := r.UserAgent()
for i := range reports {
reports[i] = finishCSPReport(reports[i], now, peerIP, requestUA)
}
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
for _, report := range reports {
if err := c.cspStore.AppendCSPReport(ctx, report); err != nil {
log.Printf("csp: failed to store report: %v", err)
http.Error(w, "failed to store report", http.StatusInternalServerError)
return
}
}
w.WriteHeader(http.StatusNoContent)
}
// HandleCSPReports serves the view's list.
// Query: site, directive, q (substring), hours, limit.
func (c *Collector) HandleCSPReports(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if c.cspStore == nil {
http.Error(w, "csp report storage not configured", http.StatusNotImplemented)
return
}
q := storage.CSPReportsQuery{
Site: r.URL.Query().Get("site"),
Directive: r.URL.Query().Get("directive"),
Search: r.URL.Query().Get("q"),
Limit: parseLimit(r, 200),
}
if _, hours := parseStatsCommon(r); r.URL.Query().Get("hours") != "" {
q.Since = time.Now().Add(-time.Duration(hours) * time.Hour)
}
reports, err := c.cspStore.CSPReports(r.Context(), q)
if err != nil {
log.Printf("csp: list failed: %v", err)
http.Error(w, "failed to list csp reports", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{"reports": reports})
}
// HandleCSPSummary serves the view's aggregates. Query: hours (default 24).
func (c *Collector) HandleCSPSummary(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if c.cspStore == nil {
http.Error(w, "csp report storage not configured", http.StatusNotImplemented)
return
}
_, hours := parseStatsCommon(r)
if r.URL.Query().Get("hours") == "" {
hours = defaultCSPWindow
}
summary, err := c.cspStore.CSPSummary(r.Context(), hours)
if err != nil {
log.Printf("csp: summary failed: %v", err)
http.Error(w, "failed to summarize csp reports", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{"summary": summary})
}
+187
View File
@@ -0,0 +1,187 @@
package collector
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"git.tornberg.me/go-otel/internal/storage"
)
// cspMemStore records ingested reports and serves canned query results.
type cspMemStore struct {
reports []storage.CSPReport
}
func (s *cspMemStore) AppendCSPReport(_ context.Context, r storage.CSPReport) error {
s.reports = append(s.reports, r)
return nil
}
func (s *cspMemStore) CSPReports(_ context.Context, q storage.CSPReportsQuery) ([]storage.CSPReport, error) {
return s.reports, nil
}
func (s *cspMemStore) CSPSummary(_ context.Context, _ int) (storage.CSPSummary, error) {
return storage.CSPSummary{Total: uint64(len(s.reports))}, nil
}
func TestParseCSPReportsLegacy(t *testing.T) {
body := []byte(`{"csp-report":{
"document-uri":"https://app.k6n.net/page?x=1",
"referrer":"https://k6n.net/",
"blocked-uri":"https://evil.example.com/tracker.js",
"violated-directive":"script-src-elem",
"effective-directive":"script-src-elem",
"original-policy":"script-src 'self'; default-src 'self'",
"disposition":"enforce",
"status-code":200,
"script-sample":"evil()"}}`)
reports := parseCSPReports(body)
if len(reports) != 1 {
t.Fatalf("want 1 report, got %d", len(reports))
}
r := reports[0]
if r.DocumentURL != "https://app.k6n.net/page?x=1" || r.BlockedURL != "https://evil.example.com/tracker.js" {
t.Errorf("urls not parsed: %+v", r)
}
if r.EffectiveDirective != "script-src-elem" || r.Disposition != "enforce" || r.StatusCode != 200 {
t.Errorf("directives/disposition/status not parsed: %+v", r)
}
if r.OriginalPolicy != "script-src 'self'; default-src 'self'" || r.Sample != "evil()" {
t.Errorf("policy/sample not parsed: %+v", r)
}
}
func TestParseCSPReportsReportingAPI(t *testing.T) {
body := []byte(`[
{"age":1,"type":"csp-violation","url":"https://docs.tornberg.me/a","user_agent":"Mozilla/5.0",
"body":{"documentURL":"https://docs.tornberg.me/a","blockedURL":"inline","effectiveDirective":"style-src",
"disposition":"report","statusCode":200,"sample":"color:red","sourceFile":"x.js"}},
{"age":2,"type":"deprecation","body":{}}]`)
reports := parseCSPReports(body)
if len(reports) != 1 {
t.Fatalf("want only the csp-violation report, got %d", len(reports))
}
r := reports[0]
if r.EffectiveDirective != "style-src" || r.UserAgent != "Mozilla/5.0" || r.ReportType != "csp-violation" {
t.Errorf("reporting-api fields not parsed: %+v", r)
}
}
func TestParseCSPReportsRejectsGarbage(t *testing.T) {
for _, body := range []string{"", "not json", `{"nope":1}`, `[]`, `[{"body":{}}]`} {
if got := parseCSPReports([]byte(body)); got != nil {
t.Errorf("parseCSPReports(%q) = %+v, want nil", body, got)
}
}
}
func TestHandleCSPIngest(t *testing.T) {
store := &cspMemStore{}
c := New(nil, 10, nil)
c.SetCSPStore(store)
// Legacy payload via sendBeacon (text/plain, no UA in body).
req := httptest.NewRequest(http.MethodPost, "/v1/csp", strings.NewReader(
`{"csp-report":{"document-uri":"https://app.k6n.net/p","blocked-uri":"https://ads.knatofs.se/x.js",
"effective-directive":"script-src","disposition":"enforce"}}`))
req.Header.Set("User-Agent", "TestUA/1.0")
req.RemoteAddr = "203.0.113.7:1234"
rec := httptest.NewRecorder()
c.HandleCSPIngest(rec, req)
if rec.Code != http.StatusNoContent {
t.Fatalf("legacy ingest: code=%d body=%s", rec.Code, rec.Body.String())
}
if len(store.reports) != 1 {
t.Fatalf("legacy ingest: stored %d reports", len(store.reports))
}
got := store.reports[0]
if got.Site != "app.k6n.net" || got.SourceIP != "203.0.113.7" {
t.Errorf("site/source ip not derived: %+v", got)
}
if got.UserAgent != "TestUA/1.0" || got.ReportType != "csp-report" || got.ID == "" {
t.Errorf("ua/type/id not finished: %+v", got)
}
// Reporting API batch.
req = httptest.NewRequest(http.MethodPost, "/v1/csp", strings.NewReader(
`[{"type":"csp-violation","body":{"documentURL":"https://www.tornberg.me/","blockedURL":"https://b.example/"}}]`))
req.Header.Set("X-Forwarded-For", "198.51.100.9, 10.0.0.1")
rec = httptest.NewRecorder()
c.HandleCSPIngest(rec, req)
if rec.Code != http.StatusNoContent || len(store.reports) != 2 {
t.Fatalf("api ingest: code=%d stored=%d", rec.Code, len(store.reports))
}
if got := store.reports[1]; got.Site != "www.tornberg.me" || got.SourceIP != "198.51.100.9" {
t.Errorf("api ingest normalization: %+v", got)
}
// Garbage is rejected; reads stay 405.
if rec := post(c, "/v1/csp", "garbage"); rec.Code != http.StatusBadRequest {
t.Errorf("garbage: code=%d", rec.Code)
}
getRec := httptest.NewRecorder()
c.HandleCSPIngest(getRec, httptest.NewRequest(http.MethodGet, "/v1/csp", nil))
if getRec.Code != http.StatusMethodNotAllowed {
t.Errorf("GET: code=%d", getRec.Code)
}
}
func post(c *Collector, path, body string) *httptest.ResponseRecorder {
rec := httptest.NewRecorder()
c.HandleCSPIngest(rec, httptest.NewRequest(http.MethodPost, path, strings.NewReader(body)))
return rec
}
func TestHandleCSPViews(t *testing.T) {
store := &cspMemStore{reports: []storage.CSPReport{{
ID: "abc", Timestamp: time.Now(), Site: "app.k6n.net", EffectiveDirective: "script-src",
}}}
c := New(nil, 10, nil)
c.SetCSPStore(store)
rec := httptest.NewRecorder()
c.HandleCSPReports(rec, httptest.NewRequest(http.MethodGet, "/api/csp/reports?site=app.k6n.net&hours=24", nil))
if rec.Code != http.StatusOK {
t.Fatalf("reports: code=%d", rec.Code)
}
var list struct {
Reports []storage.CSPReport `json:"reports"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &list); err != nil || len(list.Reports) != 1 {
t.Fatalf("reports: err=%v body=%s", err, rec.Body.String())
}
rec = httptest.NewRecorder()
c.HandleCSPSummary(rec, httptest.NewRequest(http.MethodGet, "/api/csp/summary?hours=168", nil))
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), `"total":1`) {
t.Fatalf("summary: code=%d body=%s", rec.Code, rec.Body.String())
}
// Without a store both views answer 501 politely.
c2 := New(nil, 10, nil)
rec = httptest.NewRecorder()
c2.HandleCSPReports(rec, httptest.NewRequest(http.MethodGet, "/api/csp/reports", nil))
if rec.Code != http.StatusNotImplemented {
t.Fatalf("no store: code=%d", rec.Code)
}
}
func TestFinishCSPReportIDStable(t *testing.T) {
now := time.Date(2025, 1, 2, 3, 4, 5, 0, time.UTC)
a := finishCSPReport(storage.CSPReport{DocumentURL: "https://a.k6n.net/", BlockedURL: "https://b/"}, now, "1.2.3.4", "UA")
b := finishCSPReport(storage.CSPReport{DocumentURL: "https://a.k6n.net/", BlockedURL: "https://b/"}, now, "5.6.7.8", "Other")
if a.ID != b.ID {
// The id intentionally covers report content only, not delivery
// context, so duplicate re-deliveries share it.
t.Errorf("ids differ across delivery context: %s vs %s", a.ID, b.ID)
}
if a.Site != "a.k6n.net" || a.Timestamp != now.UTC() {
t.Errorf("finish: %+v", a)
}
}
+3
View File
@@ -159,6 +159,9 @@ func (s *ClickHouseStore) initSchema(ctx context.Context) error {
if err := EnsureSourceIPSchema(ctx, s.conn, s.db); err != nil {
return fmt.Errorf("source ip schema: %w", err)
}
if err := EnsureCSPSchema(ctx, s.conn, s.db); err != nil {
return fmt.Errorf("csp report 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,
+368
View File
@@ -0,0 +1,368 @@
package storage
import (
"bufio"
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"time"
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
)
// CSPReport is one Content-Security-Policy violation report, normalized from
// either the legacy report-uri format ({"csp-report": {...}}) or the newer
// Reporting API (application/reports+json).
type CSPReport struct {
// ID is a stable content hash, computed at ingest, usable as a UI key.
ID string `json:"id"`
Timestamp time.Time `json:"timestamp"`
Site string `json:"site"` // host of document_url, e.g. app.k6n.net
DocumentURL string `json:"documentUrl"`
Referrer string `json:"referrer,omitempty"`
BlockedURL string `json:"blockedUrl,omitempty"`
ViolatedDirective string `json:"violatedDirective,omitempty"`
EffectiveDirective string `json:"effectiveDirective,omitempty"`
OriginalPolicy string `json:"originalPolicy,omitempty"`
Disposition string `json:"disposition,omitempty"` // enforce | report
StatusCode int `json:"statusCode,omitempty"`
Sample string `json:"sample,omitempty"`
// ReportType is the Reporting API type ("csp-violation"); empty for
// legacy report-uri payloads.
ReportType string `json:"reportType,omitempty"`
UserAgent string `json:"userAgent,omitempty"`
SourceIP string `json:"sourceIP,omitempty"`
}
// CSPReportsQuery filters the stored reports for the UI list.
type CSPReportsQuery struct {
// Site filters on the document host ("app.k6n.net"); empty = all.
Site string
// Directive filters on the effective directive ("script-src"); empty = all.
Directive string
// Search is a substring match across urls, referrer and sample.
Search string
// Since (zero = all time) drops reports older than the given instant.
Since time.Time
Limit int
}
// CSPCount is one named bucket of the summary (directive, host, site).
type CSPCount struct {
Name string `json:"name"`
Count uint64 `json:"count"`
}
// CSPSummary aggregates reports over one window for the view's overview cards.
type CSPSummary struct {
Total uint64 `json:"total"`
Sites uint64 `json:"sites"`
ByDirective []CSPCount `json:"byDirective"`
ByBlockedURL []CSPCount `json:"byBlockedHost"`
BySite []CSPCount `json:"bySite"`
}
// CSPStore persists CSP violation reports and answers the view's queries.
// ClickHouseStore implements it; CSPFileStore is the fallback when no
// ClickHouse is configured.
type CSPStore interface {
AppendCSPReport(ctx context.Context, r CSPReport) error
CSPReports(ctx context.Context, q CSPReportsQuery) ([]CSPReport, error)
CSPSummary(ctx context.Context, hours int) (CSPSummary, error)
}
// EnsureCSPSchema creates the dedicated csp_reports table: one row per
// browser report, typed columns instead of a JSON blob so the view can
// aggregate server-side. Idempotent.
func EnsureCSPSchema(ctx context.Context, conn driver.Conn, db string) error {
return conn.Exec(ctx, fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.csp_reports (
id String,
timestamp DateTime64(9, 'UTC'),
site LowCardinality(String),
document_url String,
referrer String,
blocked_url String,
violated_directive LowCardinality(String),
effective_directive LowCardinality(String),
original_policy String,
disposition LowCardinality(String),
status_code UInt16,
sample String,
report_type LowCardinality(String),
user_agent String,
source_ip String
) ENGINE = MergeTree
PARTITION BY toYYYYMM(timestamp)
ORDER BY (site, effective_directive, timestamp)
TTL toDateTime(timestamp) + INTERVAL 90 DAY`, db))
}
func (s *ClickHouseStore) AppendCSPReport(ctx context.Context, r CSPReport) error {
return s.conn.Exec(ctx, fmt.Sprintf(`INSERT INTO %s.csp_reports
(id, timestamp, site, document_url, referrer, blocked_url, violated_directive,
effective_directive, original_policy, disposition, status_code, sample,
report_type, user_agent, source_ip) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, s.db),
r.ID, r.Timestamp.UTC(), r.Site, r.DocumentURL, r.Referrer, r.BlockedURL,
r.ViolatedDirective, r.EffectiveDirective, r.OriginalPolicy, r.Disposition,
r.StatusCode, r.Sample, r.ReportType, r.UserAgent, r.SourceIP)
}
func (s *ClickHouseStore) CSPReports(ctx context.Context, q CSPReportsQuery) ([]CSPReport, error) {
if q.Limit <= 0 || q.Limit > 1000 {
q.Limit = 200
}
where, args := "1", []any{}
if q.Site != "" {
where += " AND site = ?"
args = append(args, q.Site)
}
if q.Directive != "" {
where += " AND effective_directive = ?"
args = append(args, q.Directive)
}
if q.Search != "" {
where += " AND (position(document_url, ?) > 0 OR position(blocked_url, ?) > 0 OR position(referrer, ?) > 0 OR position(sample, ?) > 0)"
for i := 0; i < 4; i++ {
args = append(args, q.Search)
}
}
if !q.Since.IsZero() {
where += " AND timestamp >= ?"
args = append(args, q.Since.UTC())
}
args = append(args, q.Limit)
rows, err := s.conn.Query(ctx, fmt.Sprintf(`SELECT id, timestamp, site, document_url, referrer, blocked_url,
violated_directive, effective_directive, original_policy, disposition, status_code,
sample, report_type, user_agent, source_ip
FROM %s.csp_reports WHERE %s ORDER BY timestamp DESC LIMIT ?`, s.db, where), args...)
if err != nil {
return nil, fmt.Errorf("clickhouse csp reports: %w", err)
}
defer rows.Close()
out := []CSPReport{}
for rows.Next() {
var r CSPReport
var code uint16
if err := rows.Scan(&r.ID, &r.Timestamp, &r.Site, &r.DocumentURL, &r.Referrer,
&r.BlockedURL, &r.ViolatedDirective, &r.EffectiveDirective, &r.OriginalPolicy,
&r.Disposition, &code, &r.Sample, &r.ReportType, &r.UserAgent, &r.SourceIP); err != nil {
return nil, err
}
r.StatusCode = int(code)
out = append(out, r)
}
return out, rows.Err()
}
func (s *ClickHouseStore) CSPSummary(ctx context.Context, hours int) (CSPSummary, error) {
if hours <= 0 {
hours = 24
}
since := time.Now().UTC().Add(-time.Duration(hours) * time.Hour)
var sum CSPSummary
if err := s.conn.QueryRow(ctx, fmt.Sprintf(
`SELECT count(), uniqExact(site) FROM %s.csp_reports WHERE timestamp >= ?`, s.db),
since).Scan(&sum.Total, &sum.Sites); err != nil {
return sum, fmt.Errorf("clickhouse csp summary: %w", err)
}
counts := func(expr string) ([]CSPCount, error) {
rows, err := s.conn.Query(ctx, fmt.Sprintf(
`SELECT %s AS name, count() AS c FROM %s.csp_reports
WHERE timestamp >= ? GROUP BY name ORDER BY c DESC LIMIT 20`, expr, s.db), since)
if err != nil {
return nil, err
}
defer rows.Close()
var out []CSPCount
for rows.Next() {
var v CSPCount
if err := rows.Scan(&v.Name, &v.Count); err != nil {
return nil, err
}
out = append(out, v)
}
return out, rows.Err()
}
var err error
if sum.ByDirective, err = counts("effective_directive"); err != nil {
return sum, err
}
if sum.ByBlockedURL, err = counts("domain(blocked_url)"); err != nil {
return sum, err
}
if sum.BySite, err = counts("site"); err != nil {
return sum, err
}
return sum, nil
}
// URLLowerHost extracts the lowercase host (port stripped) from a URL-ish
// string without failing on scheme-relative or malformed input; shared by
// ingest normalization and the file-store summary.
func URLLowerHost(raw string) string {
raw = strings.TrimSpace(raw)
if raw == "" {
return ""
}
if i := strings.IndexAny(raw, "?#"); i != -1 {
raw = raw[:i]
}
raw = strings.TrimPrefix(strings.TrimPrefix(raw, "https://"), "http://")
raw = strings.TrimPrefix(raw, "//")
if i := strings.IndexByte(raw, '/'); i != -1 {
raw = raw[:i]
}
if i := strings.LastIndexByte(raw, '@'); i != -1 {
raw = raw[i+1:]
}
if i := strings.LastIndexByte(raw, ':'); i != -1 && strings.LastIndexByte(raw, ']') < i {
raw = raw[:i]
}
return strings.ToLower(raw)
}
// CSPFileStore is the no-ClickHouse fallback: newline-delimited JSON next to
// the telemetry log, read back with in-memory filtering.
type CSPFileStore struct {
filename string
mu sync.Mutex
}
func NewCSPFileStore(filename string) (*CSPFileStore, error) {
if dir := filepath.Dir(filename); dir != "" && dir != "." {
if err := os.MkdirAll(dir, 0o755); err != nil {
return nil, err
}
}
return &CSPFileStore{filename: filename}, nil
}
func (fs *CSPFileStore) AppendCSPReport(_ context.Context, r CSPReport) error {
fs.mu.Lock()
defer fs.mu.Unlock()
f, err := os.OpenFile(fs.filename, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
if err != nil {
return err
}
defer f.Close()
return json.NewEncoder(f).Encode(r)
}
// readAll loads every stored report, oldest first.
func (fs *CSPFileStore) readAll() ([]CSPReport, error) {
fs.mu.Lock()
defer fs.mu.Unlock()
f, err := os.Open(fs.filename)
if err != nil {
if os.IsNotExist(err) {
return nil, nil
}
return nil, err
}
defer f.Close()
var out []CSPReport
sc := bufio.NewScanner(f)
for sc.Scan() {
line := strings.TrimSpace(sc.Text())
if line == "" {
continue
}
var r CSPReport
if err := json.Unmarshal([]byte(line), &r); err == nil {
out = append(out, r)
}
}
return out, sc.Err()
}
func (fs *CSPFileStore) CSPReports(_ context.Context, q CSPReportsQuery) ([]CSPReport, error) {
if q.Limit <= 0 || q.Limit > 1000 {
q.Limit = 200
}
all, err := fs.readAll()
if err != nil {
return nil, err
}
match := func(hay, needle string) bool { return needle == "" || strings.Contains(hay, needle) }
out := []CSPReport{}
for _, r := range all {
if q.Site != "" && r.Site != q.Site {
continue
}
if q.Directive != "" && r.EffectiveDirective != q.Directive {
continue
}
if !q.Since.IsZero() && r.Timestamp.Before(q.Since) {
continue
}
if q.Search != "" &&
!match(r.DocumentURL, q.Search) && !match(r.BlockedURL, q.Search) &&
!match(r.Referrer, q.Search) && !match(r.Sample, q.Search) {
continue
}
out = append(out, r)
}
// Newest first, bounded.
sort.Slice(out, func(i, j int) bool { return out[i].Timestamp.After(out[j].Timestamp) })
if len(out) > q.Limit {
out = out[:q.Limit]
}
return out, nil
}
func (fs *CSPFileStore) CSPSummary(_ context.Context, hours int) (CSPSummary, error) {
if hours <= 0 {
hours = 24
}
since := time.Now().Add(-time.Duration(hours) * time.Hour)
all, err := fs.readAll()
if err != nil {
return CSPSummary{}, err
}
var sum CSPSummary
directive, blocked, site := map[string]uint64{}, map[string]uint64{}, map[string]uint64{}
sites := map[string]bool{}
add := func(m map[string]uint64, name string) {
if name != "" {
m[name]++
}
}
for _, r := range all {
if r.Timestamp.Before(since) {
continue
}
sum.Total++
sites[r.Site] = true
add(directive, r.EffectiveDirective)
add(blocked, URLLowerHost(r.BlockedURL))
add(site, r.Site)
}
sum.Sites = uint64(len(sites))
sum.ByDirective = topCounts(directive)
sum.ByBlockedURL = topCounts(blocked)
sum.BySite = topCounts(site)
return sum, nil
}
// topCounts sorts a counter map, heaviest first, capped like the ClickHouse
// summary so both stores return comparable payloads.
func topCounts(m map[string]uint64) []CSPCount {
out := make([]CSPCount, 0, len(m))
for name, count := range m {
out = append(out, CSPCount{Name: name, Count: count})
}
for i := 1; i < len(out); i++ {
for j := i; j > 0 && out[j].Count > out[j-1].Count; j-- {
out[j], out[j-1] = out[j-1], out[j]
}
}
if len(out) > 20 {
out = out[:20]
}
return out
}
+77
View File
@@ -0,0 +1,77 @@
package storage
import (
"context"
"os"
"testing"
"time"
)
// TestCSPReportsIntegration is skipped unless OTEL_CLICKHOUSE_TEST_ADDR
// points at a disposable ClickHouse (native port 9000). It exercises the
// real SQL: schema creation, typed insert and the view's queries.
func TestCSPReportsIntegration(t *testing.T) {
addr := os.Getenv("OTEL_CLICKHOUSE_TEST_ADDR")
if addr == "" {
t.Skip("set OTEL_CLICKHOUSE_TEST_ADDR=host:9000 (with a disposable DB) to run")
}
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
db := "otel_csp_test_" + time.Now().UTC().Format("150405")
s, err := NewClickHouseStore(ctx, ClickHouseConfig{Addr: addr, Database: db, Username: os.Getenv("OTEL_CLICKHOUSE_TEST_USER"), Password: os.Getenv("OTEL_CLICKHOUSE_TEST_PASSWORD")})
if err != nil {
t.Skipf("ClickHouse not reachable at %s: %v", addr, err)
}
defer func() {
_ = s.conn.Exec(ctx, "DROP DATABASE IF EXISTS "+db)
_ = s.Close()
}()
now := time.Now().UTC().Truncate(time.Millisecond)
reports := []CSPReport{
{ID: "r1", Timestamp: now, Site: "app.k6n.net", DocumentURL: "https://app.k6n.net/p", BlockedURL: "https://ads.example/x.js", EffectiveDirective: "script-src", Disposition: "enforce", StatusCode: 200, Sample: "evil()", ReportType: "csp-report", UserAgent: "UA/1", SourceIP: "203.0.113.1"},
{ID: "r2", Timestamp: now.Add(-time.Minute), Site: "docs.tornberg.me", DocumentURL: "https://docs.tornberg.me/", BlockedURL: "inline", EffectiveDirective: "style-src", Disposition: "report", ReportType: "csp-violation"},
}
for _, r := range reports {
if err := s.AppendCSPReport(ctx, r); err != nil {
t.Fatalf("append: %v", err)
}
}
got, err := s.CSPReports(ctx, CSPReportsQuery{Limit: 10})
if err != nil {
t.Fatalf("list: %v", err)
}
if len(got) != 2 || got[0].ID != "r1" {
t.Fatalf("list: got %+v", got)
}
if got[0].Site != "app.k6n.net" || got[0].StatusCode != 200 || got[0].Sample != "evil()" || got[0].Timestamp != now {
t.Errorf("round-trip mismatch: %+v", got[0])
}
if got, err = s.CSPReports(ctx, CSPReportsQuery{Site: "docs.tornberg.me"}); err != nil || len(got) != 1 {
t.Errorf("site filter: %v %+v", err, got)
}
if got, err = s.CSPReports(ctx, CSPReportsQuery{Directive: "style-src"}); err != nil || len(got) != 1 || got[0].ID != "r2" {
t.Errorf("directive filter: %v %+v", err, got)
}
if got, err = s.CSPReports(ctx, CSPReportsQuery{Search: "ads.example"}); err != nil || len(got) != 1 {
t.Errorf("search filter: %v %+v", err, got)
}
sum, err := s.CSPSummary(ctx, 24)
if err != nil {
t.Fatalf("summary: %v", err)
}
if sum.Total != 2 || sum.Sites != 2 {
t.Errorf("summary: %+v", sum)
}
if len(sum.ByDirective) != 2 {
t.Errorf("summary directives: %+v", sum.ByDirective)
}
if len(sum.ByBlockedURL) != 2 || sum.ByBlockedURL[0].Name != "ads.example" {
t.Errorf("summary blocked hosts: %+v", sum.ByBlockedURL)
}
if len(sum.BySite) != 2 || sum.BySite[0].Count != 1 {
t.Errorf("summary sites: %+v", sum.BySite)
}
}
+91
View File
@@ -0,0 +1,91 @@
package storage
import (
"context"
"os"
"path/filepath"
"testing"
"time"
)
func TestURLLowerHost(t *testing.T) {
cases := map[string]string{
"https://App.K6N.net/page?x=1#f": "app.k6n.net",
"http://localhost:8080/a": "localhost",
"//cdn.knatofs.se/x.js": "cdn.knatofs.se",
"inline": "inline",
"": "",
"https://user:pw@host.tornberg.me/p": "host.tornberg.me",
}
for in, want := range cases {
if got := URLLowerHost(in); got != want {
t.Errorf("URLLowerHost(%q) = %q, want %q", in, got, want)
}
}
}
func TestCSPFileStoreRoundTrip(t *testing.T) {
path := filepath.Join(t.TempDir(), "csp", "reports.jsonl")
fs, err := NewCSPFileStore(path)
if err != nil {
t.Fatalf("open: %v", err)
}
now := time.Now().UTC()
old := now.Add(-48 * time.Hour)
reports := []CSPReport{
{ID: "1", Timestamp: now, Site: "app.k6n.net", DocumentURL: "https://app.k6n.net/", BlockedURL: "https://ads.example/x.js", EffectiveDirective: "script-src"},
{ID: "2", Timestamp: now, Site: "docs.tornberg.me", DocumentURL: "https://docs.tornberg.me/", BlockedURL: "inline", EffectiveDirective: "style-src", Sample: "color:red"},
{ID: "3", Timestamp: old, Site: "app.k6n.net", DocumentURL: "https://app.k6n.net/old", EffectiveDirective: "img-src"},
}
for _, r := range reports {
if err := fs.AppendCSPReport(context.Background(), r); err != nil {
t.Fatalf("append: %v", err)
}
}
// All reports, newest first.
got, err := fs.CSPReports(context.Background(), CSPReportsQuery{Limit: 10})
if err != nil {
t.Fatalf("list: %v", err)
}
if len(got) != 3 || got[0].ID != "1" {
t.Errorf("list order: %+v", got)
}
// Filters: directive and since.
if got, _ = fs.CSPReports(context.Background(), CSPReportsQuery{Directive: "script-src"}); len(got) != 1 || got[0].ID != "1" {
t.Errorf("directive filter: %+v", got)
}
if got, _ = fs.CSPReports(context.Background(), CSPReportsQuery{Since: now.Add(-time.Hour), Limit: 10}); len(got) != 2 {
t.Errorf("since filter: %+v", got)
}
if got, _ = fs.CSPReports(context.Background(), CSPReportsQuery{Search: "color:red"}); len(got) != 1 || got[0].ID != "2" {
t.Errorf("search filter: %+v", got)
}
sum, err := fs.CSPSummary(context.Background(), 24)
if err != nil {
t.Fatalf("summary: %v", err)
}
if sum.Total != 2 || sum.Sites != 2 {
t.Errorf("summary totals: %+v", sum)
}
if len(sum.ByDirective) != 2 || sum.ByDirective[0].Name != "script-src" {
t.Errorf("summary directives: %+v", sum.ByDirective)
}
if len(sum.ByBlockedURL) != 2 || sum.ByBlockedURL[0].Name != "ads.example" {
t.Errorf("summary blocked hosts: %+v", sum.ByBlockedURL)
}
// A fresh store over a missing file answers empty instead of failing.
empty, err := NewCSPFileStore(filepath.Join(t.TempDir(), "none.jsonl"))
if err != nil {
t.Fatalf("open empty: %v", err)
}
if got, _ = empty.CSPReports(context.Background(), CSPReportsQuery{}); len(got) != 0 {
t.Errorf("missing file: %+v", got)
}
if _, err := os.Stat(path); err != nil {
t.Errorf("store file missing after appends: %v", err)
}
}