amadan

carlos / platform Public

reqrec: bounded method vocabulary, separate block and watch caps, summaries through EdgeSummary, env and method counts, recorder-stamped time (final review C2, I3, I6, I9)

The e2e tests run on a fixed recorder clock, close their registry, and
send the review's hostile inputs (masking gaps, bare query, controls, a
100,000-character method) through the real edge, sink and chunk codec.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ed6815820a9077337ea7b74d93667d37fe2a5db4 parent b7fd458
9 files changed, +543 −85
  • internal/edge/reqrecords.go +5 −3
  • internal/edge/reqrecords_test.go +19 −0
  • internal/host/agent.go +1 −1
  • internal/host/reqrec_e2e_test.go +214 −39
  • internal/host/reqrec_loop.go +6 −4
  • internal/host/reqrec_loop_test.go +4 −4
  • internal/reqrec/recorder.go +116 −25
  • internal/reqrec/recorder_test.go +173 −4
  • internal/reqrec/redact_test.go +5 −5
diff --git a/internal/edge/reqrecords.go b/internal/edge/reqrecords.go
index bb9d4387d..15c6121c7 100644
--- a/internal/edge/reqrecords.go
+++ b/internal/edge/reqrecords.go
@@ -33,14 +33,16 @@ func (e *Edge) EnableRequestRecords(rec RequestRecorder) {
// requestResponse is the Response for r on rt, shared by the observer defer
// and recordMatch so a match and a plain response describe a request the
-// same way. Woke is left for the caller: only the defer knows it.
+// same way. Woke is left for the caller: only the defer knows it. When is
+// left zero, so the recorder stamps it from its own clock; the environment
+// is the route's as it was when this request was served.
func requestResponse(rt registry.Route, r *http.Request, status int) reqrec.Response {
geoAns, _ := geoAnswerFrom(r.Context())
return reqrec.Response{
- Acct: release.RouteAccount(rt), App: rt.App, Host: rt.Host,
+ Acct: release.RouteAccount(rt), App: rt.App, Host: rt.Host, Environment: rt.Environment,
Method: r.Method, Path: r.URL.Path, RawQuery: r.URL.RawQuery,
UA: r.Header.Get("User-Agent"), ClientIP: clientIP(r), Country: geoAns.Country,
- Status: status, When: time.Now(),
+ Status: status,
}
}
diff --git a/internal/edge/reqrecords_test.go b/internal/edge/reqrecords_test.go
index ffe4745a5..fe73b612c 100644
--- a/internal/edge/reqrecords_test.go
+++ b/internal/edge/reqrecords_test.go
@@ -31,6 +31,20 @@ func TestRecordsSeeAppNotFound(t *testing.T) {
}
}
+// The edge leaves When unset: the recorder stamps it with its own clock, so
+// one clock decides which minute a request lands in.
+func TestRecordsLeaveWhenToTheRecorder(t *testing.T) {
+ e := edgeWithBackend(t, http.NotFoundHandler())
+ rec := &memRecorder{}
+ e.EnableRequestRecords(rec)
+ e.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", "http://app.test/x", nil))
+ rt := registry.Route{Host: "app.test", App: "shop"}
+ e.recordMatch(rt, httptest.NewRequest("GET", "http://app.test/.env", nil), reqrec.KindBlock, reqrec.TierHard, "dotenv", "", 404)
+ if len(rec.resp) != 1 || !rec.resp[0].When.IsZero() || len(rec.match) != 1 || !rec.match[0].When.IsZero() {
+ t.Fatalf("resp %+v match %+v", rec.resp, rec.match)
+ }
+}
+
// The hub that forwarded a request over the tunnel records it; the box it
// was delivered to must not record it a second time.
func TestRecordsSkipTunnelDelivered(t *testing.T) {
@@ -51,11 +65,16 @@ func TestRecordMatchReturnsSealedCode(t *testing.T) {
rec := &memRecorder{}
e.EnableRequestRecords(rec)
rt := registry.Route{Host: "app.test", App: "shop"}
+ rt.Environment = "staging"
+ rt.Environment = "staging"
code := e.recordMatch(rt, httptest.NewRequest("GET", "http://app.test/.env", nil), reqrec.KindBlock, reqrec.TierHard, "dotenv", "", 404)
r, err := reqrec.ParseRef(code, key)
if err != nil || !r.Verified || len(rec.match) != 1 || rec.match[0].Ref != code {
t.Fatalf("code %q: %+v %v; matches %d", code, r, err, len(rec.match))
}
+ if rec.match[0].Environment != "staging" {
+ t.Fatalf("environment %q, want the route's", rec.match[0].Environment)
+ }
}
// With recording off, recordMatch still hands back a code: the block page
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 84b4f7ba3..1d1f36c6a 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -961,7 +961,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// the analytics meter, so the record salt is the analytics day salt.
// The nil check is on the concrete recorder: a nil *Recorder passed
// as the edge's interface would not be nil to it.
- if rec := newRequestRecorder(logSink, dataDir(), os.Getenv); rec != nil {
+ if rec := newRequestRecorder(logSink, dataDir(), os.Getenv, time.Now); rec != nil {
e.EnableRequestRecords(rec)
go reqrecLoop(rec, time.NewTicker(time.Minute).C)
}
diff --git a/internal/host/reqrec_e2e_test.go b/internal/host/reqrec_e2e_test.go
index 22a015bc5..e083f9989 100644
--- a/internal/host/reqrec_e2e_test.go
+++ b/internal/host/reqrec_e2e_test.go
@@ -12,6 +12,7 @@ import (
"github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/reqrec"
)
// TestRequestRecordsEndToEnd drives the production wiring end to end: a real
@@ -29,41 +30,9 @@ func TestRequestRecordsEndToEnd(t *testing.T) {
clientAddr = "203.0.113.7"
)
- backend := httptest.NewServer(http.NotFoundHandler())
- t.Cleanup(backend.Close)
-
- reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
- if err != nil {
- t.Fatalf("open registry: %v", err)
- }
- if err := reg.Upsert(registry.Route{
- Host: "shop.example.com", Kind: "service", Account: "bab", App: "shop",
- Addr: backend.Listener.Addr().String(), Status: "ready", Version: "v1",
- }); err != nil {
- t.Fatalf("upsert: %v", err)
- }
- e, err := edge.New(reg, nil)
- if err != nil {
- t.Fatalf("edge.New: %v", err)
- }
-
- sink := applog.NewSink("box1", time.Now)
- rec := newRequestRecorder(sink, t.TempDir(), func(string) string { return "" })
- if rec == nil {
- t.Fatal("newRequestRecorder returned nil with a sink and no kill switch")
- }
- e.EnableRequestRecords(rec)
-
- get := func(path string) {
- t.Helper()
- w := httptest.NewRecorder()
- r := httptest.NewRequest("GET", "http://shop.example.com"+path, nil)
- r.RemoteAddr = clientAddr + ":4321"
- e.ServeHTTP(w, r)
- if w.Code != http.StatusNotFound {
- t.Fatalf("GET %s = %d, want 404 from the backend", path, w.Code)
- }
- }
+ x := newRecordsE2E(t)
+ sink := x.sink
+ get := func(path string) { t.Helper(); x.serve("GET", path, clientAddr) }
// The reset link goes first so it is inside the 60-record sample, which
// is the case where the token would leak if masking were not applied.
get("/reset/" + token)
@@ -71,10 +40,7 @@ func TestRequestRecordsEndToEnd(t *testing.T) {
get("/wp-login.php")
}
- tick := make(chan time.Time, 1)
- tick <- time.Now().Add(time.Minute)
- close(tick)
- reqrecLoop(rec, tick)
+ x.flush()
recs := sink.Drain()[applog.AppKey{Acct: "bab", App: "shop"}]
var notFound, minutes []applog.Record
@@ -101,11 +67,20 @@ func TestRequestRecordsEndToEnd(t *testing.T) {
if got := minutes[0].Fields["dropped_notfound"]; got != "11" {
t.Errorf("minute dropped_notfound = %q, want 11", got)
}
+ if got, want := minutes[0].Fields["minute"], x.at.Truncate(time.Minute).Format(time.RFC3339); got != want {
+ t.Errorf("minute = %q, want the recorder clock's %q", got, want)
+ }
+ if got := minutes[0].Fields["m_get"]; got != "71" {
+ t.Errorf("minute m_get = %q, want 71", got)
+ }
// Every not-found record carries the salted client hash (12 hex chars,
// which a nil salt would leave out) and the client as a /24, never the
// address itself.
for _, r := range notFound {
+ if r.Fields["env"] != "staging" {
+ t.Errorf("env = %q, want the route's environment: %v", r.Fields["env"], r.Fields)
+ }
if !isHex12(r.Fields["cid"]) {
t.Errorf("cid = %q, want 12 hex characters: %v", r.Fields["cid"], r.Fields)
}
@@ -164,6 +139,206 @@ func TestRequestRecordsEndToEnd(t *testing.T) {
leaks(back, "decoded")
}
+// recordsE2E is the production wiring the end-to-end tests share: a real
+// registry with one service route (environment "staging") to a backend that
+// 404s everything, a real edge, a real applog.Sink, and the recorder
+// newRequestRecorder builds against it. The recorder's clock is fixed, so a
+// run never straddles a minute; the sink's is the wall clock, as in
+// production, since only its rate buckets read it.
+type recordsE2E struct {
+ t *testing.T
+ at time.Time
+ e *edge.Edge
+ sink *applog.Sink
+ rec *reqrec.Recorder
+}
+
+func newRecordsE2E(t *testing.T) *recordsE2E {
+ t.Helper()
+ backend := httptest.NewServer(http.NotFoundHandler())
+ t.Cleanup(backend.Close)
+
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ if err := reg.Upsert(registry.Route{
+ Host: "shop.example.com", Kind: "service", Account: "bab", App: "shop",
+ Addr: backend.Listener.Addr().String(), Status: "ready", Version: "v1",
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := reg.SetEnvironment("shop.example.com", "staging"); err != nil {
+ t.Fatalf("set environment: %v", err)
+ }
+ e, err := edge.New(reg, nil)
+ if err != nil {
+ t.Fatalf("edge.New: %v", err)
+ }
+
+ at := time.Date(2026, 10, 24, 21, 7, 30, 0, time.UTC)
+ sink := applog.NewSink("box1", time.Now)
+ rec := newRequestRecorder(sink, t.TempDir(), func(string) string { return "" }, func() time.Time { return at })
+ if rec == nil {
+ t.Fatal("newRequestRecorder returned nil with a sink and no kill switch")
+ }
+ e.EnableRequestRecords(rec)
+ return &recordsE2E{t: t, at: at, e: e, sink: sink, rec: rec}
+}
+
+// serve sends one request through the edge from client and requires the
+// backend's 404.
+func (x *recordsE2E) serve(method, target, client string) {
+ x.t.Helper()
+ w := httptest.NewRecorder()
+ r := httptest.NewRequest(method, "http://shop.example.com"+target, nil)
+ r.RemoteAddr = client + ":4321"
+ x.e.ServeHTTP(w, r)
+ if w.Code != http.StatusNotFound {
+ x.t.Fatalf("%.20s %s = %d, want 404 from the backend", method, target, w.Code)
+ }
+}
+
+// flush closes the recorder's minute through reqrecLoop, as the agent does.
+func (x *recordsE2E) flush() {
+ tick := make(chan time.Time, 1)
+ tick <- x.at.Add(time.Minute)
+ close(tick)
+ reqrecLoop(x.rec, tick)
+}
+
+// TestRequestRecordsHostileInputEndToEnd sends what the final review found
+// getting through, over the real wiring and the chunk codec: tokens the old
+// masking kept, a bare query component, control characters, and an
+// extension method of 100,000 characters. The chunk must still decode (an
+// oversized line would make the app's whole log window unreadable), and none
+// of it may come back out.
+func TestRequestRecordsHostileInputEndToEnd(t *testing.T) {
+ const client = "198.51.100.23"
+ x := newRecordsE2E(t)
+ secrets := []string{"rKpWmTzQvLsNxYhJbCfDaGeS", "Xq3vB9kLmN2pQr7s", "T1uVwYz4aBcDeFg", "abcdefabcdefabcdefabcdefabcdefab", "hunter2"}
+ paths := map[string]string{
+ "/reset/rKpWmTzQvLsNxYhJbCfDaGeS": "/reset/[masked]",
+ "/reset/Xq3vB9kLmN2pQr7s/T1uVwYz4aBcDeFg": "/reset/[masked]",
+ "/zabcdefabcdefabcdefabcdefabcdefab": "/[masked]",
+ }
+ for p := range paths {
+ x.serve("GET", p, client)
+ }
+ x.serve("GET", "/login?hunter2", client)
+ x.serve("GET", "/x%1b%5b2J%0aforged?a%0d=1", client)
+ huge := strings.Repeat("&", 100000)
+ x.serve(huge, "/m", client)
+ x.flush()
+
+ recs := x.sink.Drain()[applog.AppKey{Acct: "bab", App: "shop"}]
+ b, err := applog.EncodeChunk(recs)
+ if err != nil {
+ t.Fatalf("EncodeChunk: %v", err)
+ }
+ if len(b) > 64<<10 {
+ t.Errorf("chunk of %d records is %d bytes", len(recs), len(b))
+ }
+ back, err := applog.DecodeChunk(b)
+ if err != nil {
+ t.Fatalf("DecodeChunk: %v (an unreadable chunk hides the app's own logs)", err)
+ }
+ if !reflect.DeepEqual(back, recs) {
+ t.Errorf("records changed across EncodeChunk/DecodeChunk")
+ }
+ for _, s := range secrets {
+ if strings.Contains(string(b), s) {
+ t.Errorf("%q in the encoded chunk", s)
+ }
+ }
+
+ seen := map[string]applog.Record{}
+ for _, r := range back {
+ if r.Fields["kind"] == "notfound" {
+ seen[r.Fields["path"]] = r
+ }
+ for _, v := range append([]string{r.Msg}, mapValues(r.Fields)...) {
+ for _, c := range v {
+ if c < 0x20 || c == 0x7f || c >= 0x80 && c <= 0x9f {
+ t.Errorf("control %U survived decoding in %q", c, v)
+ }
+ }
+ }
+ }
+ for in, want := range paths {
+ if _, ok := seen[want]; !ok {
+ t.Errorf("%s: no record with path %q (have %v)", in, want, mapKeys(seen))
+ }
+ }
+ if r, ok := seen["/login"]; !ok || r.Fields["qnames"] != "" {
+ t.Errorf("bare query component kept: %+v", r.Fields)
+ }
+ if r, ok := seen[`/x\x1b[2J\x0aforged`]; !ok || r.Fields["qnames"] != `a\x0d` {
+ t.Errorf("control path not escaped: have %v", mapKeys(seen))
+ }
+ if r, ok := seen["/m"]; !ok || r.Fields["method"] != "OTHER" || r.Msg != "404 OTHER /m" {
+ t.Errorf("extension method record: %+v", r)
+ }
+}
+
+// TestRequestRecordsKeepAppLinesEndToEnd: an app line written before a flood
+// of edge records, through the real wiring, is still there after it, and so
+// is the minute record that counts the flood, though the ring cannot hold
+// every request record.
+func TestRequestRecordsKeepAppLinesEndToEnd(t *testing.T) {
+ x := newRecordsE2E(t)
+ x.sink.MaxBytesPerApp = 6 << 10
+ x.sink.App("bab", "shop", "shop.example.com", "app line before the flood")
+ long := "/wp-login.php/" + strings.Repeat("a/", 150)
+ for i := 0; i < 70; i++ {
+ x.serve("GET", long, "203.0.113.7")
+ }
+ x.flush()
+ var app, notFound, minutes int
+ for _, r := range x.sink.Drain()[applog.AppKey{Acct: "bab", App: "shop"}] {
+ switch {
+ case r.Stream == applog.StreamApp && r.Msg == "app line before the flood":
+ app++
+ case r.Fields["kind"] == "notfound":
+ notFound++
+ case r.Fields["kind"] == "minute":
+ minutes++
+ if r.Fields["s404"] != "70" {
+ t.Errorf("minute s404 = %q, want 70", r.Fields["s404"])
+ }
+ }
+ }
+ if app != 1 || minutes != 1 || notFound == 0 || notFound >= 60 {
+ t.Fatalf("app line %d, minute records %d, notfound %d (of 60 sampled)", app, minutes, notFound)
+ }
+}
+
+// TestEdgeKindBlockAgrees: applog evicts block records after other request
+// records by reading the kind the recorder writes, and cannot import reqrec
+// to name it, so the two spellings are pinned together here.
+func TestEdgeKindBlockAgrees(t *testing.T) {
+ if string(reqrec.KindBlock) != applog.EdgeKindBlock {
+ t.Fatalf("reqrec.KindBlock %q, applog.EdgeKindBlock %q", reqrec.KindBlock, applog.EdgeKindBlock)
+ }
+}
+
+func mapValues(m map[string]string) []string {
+ out := make([]string, 0, len(m))
+ for _, v := range m {
+ out = append(out, v)
+ }
+ return out
+}
+
+func mapKeys(m map[string]applog.Record) []string {
+ out := make([]string, 0, len(m))
+ for k := range m {
+ out = append(out, k)
+ }
+ return out
+}
+
// isHex12 reports whether s is exactly 12 lowercase hex characters.
func isHex12(s string) bool {
if len(s) != 12 {
diff --git a/internal/host/reqrec_loop.go b/internal/host/reqrec_loop.go
index 2e518221a..0a143895a 100644
--- a/internal/host/reqrec_loop.go
+++ b/internal/host/reqrec_loop.go
@@ -14,7 +14,7 @@ import (
)
// The log sink is the recorder's destination; this fails the build if the
-// sink's Edge method ever drifts from what the recorder calls.
+// sink's Edge or EdgeSummary method ever drifts from what the recorder calls.
var _ reqrec.Sink = (*applog.Sink)(nil)
// newRequestRecorder returns the recorder for the edge, or nil when records
@@ -22,8 +22,10 @@ var _ reqrec.Sink = (*applog.Sink)(nil)
// CARLOS_EDGE_RECORDS=off kill switch. The salt is the analytics day salt in
// saltDir, so a record's client hash and the analytics visitor hash rest on
// the same per-day secret, which never leaves the box. A salt that cannot be
-// read gives nil, which leaves the client hash out of records.
-func newRequestRecorder(sink *applog.Sink, saltDir string, getenv func(string) string) *reqrec.Recorder {
+// read gives nil, which leaves the client hash out of records. now is the
+// recorder's clock, which stamps every record's minute (the edge leaves the
+// time to it); production passes time.Now.
+func newRequestRecorder(sink *applog.Sink, saltDir string, getenv func(string) string, now func() time.Time) *reqrec.Recorder {
if sink == nil || getenv("CARLOS_EDGE_RECORDS") == "off" {
return nil
}
@@ -34,7 +36,7 @@ func newRequestRecorder(sink *applog.Sink, saltDir string, getenv func(string) s
}
return b
}
- return reqrec.NewRecorder(sink, salt, time.Now)
+ return reqrec.NewRecorder(sink, salt, now)
}
// reqrecLoop closes finished minutes: each tick it asks the recorder to
diff --git a/internal/host/reqrec_loop_test.go b/internal/host/reqrec_loop_test.go
index 2fed33505..32583d885 100644
--- a/internal/host/reqrec_loop_test.go
+++ b/internal/host/reqrec_loop_test.go
@@ -18,20 +18,20 @@ func TestNewRequestRecorderGates(t *testing.T) {
}
return ""
}
- if newRequestRecorder(nil, t.TempDir(), on) != nil {
+ if newRequestRecorder(nil, t.TempDir(), on, time.Now) != nil {
t.Error("nil sink must give nil recorder")
}
- if newRequestRecorder(sink, t.TempDir(), off) != nil {
+ if newRequestRecorder(sink, t.TempDir(), off, time.Now) != nil {
t.Error("kill switch ignored")
}
- if newRequestRecorder(sink, t.TempDir(), on) == nil {
+ if newRequestRecorder(sink, t.TempDir(), on, time.Now) == nil {
t.Error("recorder missing")
}
}
func TestReqrecLoopFlushesToSink(t *testing.T) {
sink := applog.NewSink("box1", time.Now)
- rec := newRequestRecorder(sink, t.TempDir(), func(string) string { return "" })
+ rec := newRequestRecorder(sink, t.TempDir(), func(string) string { return "" }, time.Now)
at := time.Now().UTC().Add(-2 * time.Minute)
rec.Response(reqrec.Response{Acct: "a1", App: "shop", Status: 404, Method: "GET", Path: "/x", ClientIP: "203.0.113.7", When: at})
tick := make(chan time.Time, 1)
diff --git a/internal/reqrec/recorder.go b/internal/reqrec/recorder.go
index 1e54b6654..e12bc910d 100644
--- a/internal/reqrec/recorder.go
+++ b/internal/reqrec/recorder.go
@@ -3,14 +3,22 @@ package reqrec
import (
"sort"
"strconv"
+ "strings"
"sync"
"time"
)
// Sink receives finished records. The agent's log sink implements it; the
// recorder never holds its own lock while calling it.
+//
+// Edge takes request records (not-found, block, watch), which the sink may
+// refuse under its edge rate budget. EdgeSummary takes the per-minute counts,
+// which no rate budget may refuse: they are what says how much was sampled
+// away, so losing one loses the explanation for everything it counted. The
+// sink still bounds them by its buffer, evicting them last.
type Sink interface {
Edge(acct, app, host, level, msg string, fields map[string]string)
+ EdgeSummary(acct, app, host, level, msg string, fields map[string]string)
}
// Kind says what a record is about.
@@ -26,20 +34,73 @@ const (
// Per app and minute, at most this many records of each class are written. The
// minute counters still count everything, and say how many records were left
// out, so a flood costs a bounded number of log lines and nothing is hidden.
+// Blocks and watches have separate allowances: a watched rule matching every
+// request must not take the records that explain a block.
const (
NotFoundPerMin = 60
- MatchPerMin = 300
+ BlockPerMin = 300
+ WatchPerMin = 300
)
+// methods are the request methods a record names. Anything else is stored as
+// "OTHER": a request line may carry any extension method of any length, and
+// a record must cost a bounded number of bytes whatever the request said.
+var methods = [...]string{"GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "CONNECT", "TRACE"}
+
+// MethodOther is the stored method for anything not in methods.
+const MethodOther = "OTHER"
+
+// Method returns m if it is one of the methods a record names, else
+// MethodOther. Methods are case-sensitive, so "get" is OTHER.
+func Method(m string) string {
+ for _, k := range methods {
+ if m == k {
+ return k
+ }
+ }
+ return MethodOther
+}
+
+// methodCounters are the minute record's method counts, in field order.
+// CONNECT, TRACE and anything else count as m_other.
+var methodCounters = [...]string{"m_get", "m_head", "m_post", "m_put", "m_patch", "m_delete", "m_options", "m_other"}
+
+func methodIndex(m string) int {
+ switch m {
+ case "GET":
+ return 0
+ case "HEAD":
+ return 1
+ case "POST":
+ return 2
+ case "PUT":
+ return 3
+ case "PATCH":
+ return 4
+ case "DELETE":
+ return 5
+ case "OPTIONS":
+ return 6
+ }
+ return 7
+}
+
// Response is one response the edge served. It carries raw request data; the
// recorder redacts it before anything leaves.
type Response struct {
Acct, App, Host, Method, Path, RawQuery, UA, ClientIP, Country string
- Status int
+ // Environment is the route's environment when the request was served
+ // (registry.Route.Environment), "" for none. A route can change later,
+ // so the record keeps it.
+ Environment string
+ Status int
// Woke is whether the request woke the app, and only means something when
// WokeKnown is set.
Woke, WokeKnown bool
- When time.Time
+ // When is when the response was served. Zero means now by the recorder's
+ // clock, which is what the edge passes: one clock decides which minute a
+ // request lands in.
+ When time.Time
}
// Match is a Response that a protection rule matched, either blocking it or
@@ -60,11 +121,12 @@ type minuteKey struct {
}
type minuteCounts struct {
- s2xx, s3xx, s4xx, s404, s5xx int
- wakes, wakes404 int
- block, watch int
- notFoundRecs, matchRecs int
- droppedNotFound, droppedMatch int
+ s2xx, s3xx, s4xx, s404, s5xx int
+ wakes, wakes404 int
+ block, watch int
+ methods [len(methodCounters)]int
+ notFoundRecs, blockRecs, watchRecs int
+ droppedNotFound, droppedBlock, droppedWatch int
}
// Recorder turns the edge's responses into sampled records and per-minute
@@ -176,6 +238,14 @@ func (c *minuteCounts) count(x Response) {
c.wakes404++
}
}
+ c.methods[methodIndex(x.Method)]++
+}
+
+// stamp gives a zero When the recorder's own now.
+func (r *Recorder) stamp(x *Response) {
+ if x.When.IsZero() {
+ x.When = r.now()
+ }
}
func (r *Recorder) counts(x Response) *minuteCounts {
@@ -194,6 +264,7 @@ func (r *Recorder) Response(x Response) {
if r == nil {
return
}
+ r.stamp(&x)
r.mu.Lock()
c := r.counts(x)
c.count(x)
@@ -211,31 +282,40 @@ func (r *Recorder) Response(x Response) {
return
}
f := r.fields(x, KindNotFound)
- r.sink.Edge(x.Acct, x.App, x.Host, "info", "404 "+x.Method+" "+f["path"], f)
+ r.sink.Edge(x.Acct, x.App, x.Host, "info", "404 "+f["method"]+" "+f["path"], f)
}
// Match counts a response a rule matched, as a block or a watch and by status
-// class like any other response, and writes a match record under the
-// per-minute cap. It makes no not-found record, so a matched 404 is one record.
+// class like any other response, and writes a match record under its kind's
+// per-minute cap (BlockPerMin, WatchPerMin). It makes no not-found record, so a
+// matched 404 is one record. A kind other than block or watch is counted by
+// status and method only and writes nothing.
func (r *Recorder) Match(m Match) {
if r == nil {
return
}
+ r.stamp(&m.Response)
r.mu.Lock()
c := r.counts(m.Response)
c.count(m.Response)
+ emit := false
switch m.Kind {
case KindBlock:
c.block++
+ if c.blockRecs < BlockPerMin {
+ c.blockRecs++
+ emit = true
+ } else {
+ c.droppedBlock++
+ }
case KindWatch:
c.watch++
- }
- emit := false
- if c.matchRecs < MatchPerMin {
- c.matchRecs++
- emit = true
- } else {
- c.droppedMatch++
+ if c.watchRecs < WatchPerMin {
+ c.watchRecs++
+ emit = true
+ } else {
+ c.droppedWatch++
+ }
}
r.mu.Unlock()
if !emit {
@@ -245,26 +325,32 @@ func (r *Recorder) Match(m Match) {
f["tier"] = m.Tier.String()
set(f, "ref", m.Ref)
set(f, "rule", m.RuleID)
- set(f, "frag", m.Fragment)
+ // The caller passed the fragment through Fragment, which escapes
+ // controls; escaping again here keeps that true at the record boundary
+ // whatever a caller did, and costs nothing on text that has none.
+ set(f, "frag", truncate(escapeControls(strings.ToValidUTF8(m.Fragment, "\ufffd")), MaxFragment))
level := "info"
if m.Kind == KindBlock {
level = "warn"
}
r.sink.Edge(m.Acct, m.App, m.Host, level,
- string(m.Kind)+" "+m.Tier.String()+" "+m.RuleID+" "+m.Method+" "+f["path"], f)
+ string(m.Kind)+" "+m.Tier.String()+" "+m.RuleID+" "+f["method"]+" "+f["path"], f)
}
// fields builds the redacted record body shared by not-found and match records.
-// Empty values are left out.
+// Empty values are left out. Every request-derived string is bounded and has
+// its controls escaped: the method is one of a few words, the path and query
+// names go through MaskPath and QueryNames.
func (r *Recorder) fields(x Response, kind Kind) map[string]string {
f := map[string]string{
"kind": string(kind),
- "method": x.Method,
+ "method": Method(x.Method),
"path": MaskPath(x.Path),
"ua": UAFamily(x.UA),
"status": strconv.Itoa(x.Status),
}
set(f, "qnames", QueryNames(x.RawQuery))
+ set(f, "env", x.Environment)
set(f, "country", x.Country)
set(f, "client", ClientPrefix(x.ClientIP))
if key := ClientKey(x.ClientIP); key != "" {
@@ -319,7 +405,7 @@ func (r *Recorder) Flush(now time.Time) {
for _, d := range out {
minute := d.k.minute.Format(time.RFC3339)
n := strconv.Itoa
- r.sink.Edge(d.k.acct, d.k.app, "", "info", "minute "+minute, map[string]string{
+ f := map[string]string{
"kind": string(KindMinute),
"minute": minute,
"s2xx": n(d.c.s2xx),
@@ -332,7 +418,12 @@ func (r *Recorder) Flush(now time.Time) {
"block": n(d.c.block),
"watch": n(d.c.watch),
"dropped_notfound": n(d.c.droppedNotFound),
- "dropped_match": n(d.c.droppedMatch),
- })
+ "dropped_block": n(d.c.droppedBlock),
+ "dropped_watch": n(d.c.droppedWatch),
+ }
+ for i, name := range methodCounters {
+ f[name] = n(d.c.methods[i])
+ }
+ r.sink.EdgeSummary(d.k.acct, d.k.app, "", "info", "minute "+minute, f)
}
}
diff --git a/internal/reqrec/recorder_test.go b/internal/reqrec/recorder_test.go
index fc248618e..f5d66ac79 100644
--- a/internal/reqrec/recorder_test.go
+++ b/internal/reqrec/recorder_test.go
@@ -3,6 +3,7 @@ package reqrec
import (
"bytes"
"fmt"
+ "strings"
"sync"
"testing"
"time"
@@ -11,12 +12,18 @@ import (
type memSink struct {
recs []map[string]string
msgs, levels []string
+ summaries int // how many of recs came through EdgeSummary
}
func (m *memSink) Edge(acct, app, host, level, msg string, f map[string]string) {
m.recs, m.msgs, m.levels = append(m.recs, f), append(m.msgs, msg), append(m.levels, level)
}
+func (m *memSink) EdgeSummary(acct, app, host, level, msg string, f map[string]string) {
+ m.summaries++
+ m.Edge(acct, app, host, level, msg, f)
+}
+
func TestRecorderNotFoundSampledAndCounted(t *testing.T) {
now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
s := &memSink{}
@@ -53,7 +60,7 @@ func TestRecorderMatchRecordsAndCap(t *testing.T) {
r := NewRecorder(s, nil, func() time.Time { return now })
m := Match{Response: Response{Acct: "a1", App: "shop", Method: "GET", Path: "/.env", Status: 404, When: now},
Kind: KindBlock, Tier: TierHard, RuleID: "dotenv", Ref: "ref1"}
- for i := 0; i < MatchPerMin+5; i++ {
+ for i := 0; i < BlockPerMin+5; i++ {
r.Match(m)
}
r.Flush(now.Add(time.Minute))
@@ -68,7 +75,7 @@ func TestRecorderMatchRecordsAndCap(t *testing.T) {
t.Error("cid written without a salt")
}
last := s.recs[len(s.recs)-1]
- if last["kind"] != "minute" || last["block"] != fmt.Sprint(MatchPerMin+5) || last["dropped_match"] != "5" {
+ if last["kind"] != "minute" || last["block"] != fmt.Sprint(BlockPerMin+5) || last["dropped_block"] != "5" || last["dropped_watch"] != "0" {
t.Fatalf("bad minute %v", last)
}
}
@@ -109,7 +116,8 @@ func TestRecorderStatusClassesAndMatchCounts(t *testing.T) {
}
}
want := map[string]string{"s2xx": "2", "s3xx": "1", "s4xx": "2", "s404": "2", "s5xx": "2",
- "wakes": "1", "wakes404": "1", "block": "0", "watch": "1", "dropped_notfound": "0", "dropped_match": "0"}
+ "wakes": "1", "wakes404": "1", "block": "0", "watch": "1", "dropped_notfound": "0", "dropped_block": "0", "dropped_watch": "0",
+ "m_get": "9", "m_head": "0", "m_post": "0", "m_put": "0", "m_patch": "0", "m_delete": "0", "m_options": "0", "m_other": "0"}
for k, v := range want {
if min[k] != v {
t.Errorf("%s = %q, want %q (%v)", k, min[k], v, min)
@@ -143,7 +151,7 @@ func TestRecorderNotFoundMessageAndOmittedFields(t *testing.T) {
if s.msgs[0] != "404 GET /a" || s.levels[0] != "info" {
t.Errorf("msg %q level %q", s.msgs[0], s.levels[0])
}
- for _, k := range []string{"country", "cid", "qnames", "woke", "client"} {
+ for _, k := range []string{"country", "cid", "qnames", "woke", "client", "env"} {
if v, ok := f[k]; ok {
t.Errorf("%s present as %q, want omitted", k, v)
}
@@ -240,6 +248,10 @@ func (l *lockedSink) Edge(acct, app, host, level, msg string, f map[string]strin
l.mu.Unlock()
}
+func (l *lockedSink) EdgeSummary(acct, app, host, level, msg string, f map[string]string) {
+ l.Edge(acct, app, host, level, msg, f)
+}
+
func TestRecorderSlowSaltDoesNotStallCounting(t *testing.T) {
now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
release := make(chan struct{})
@@ -313,3 +325,160 @@ func TestRecorderEmptySaltRetriedAfterAMinute(t *testing.T) {
t.Fatalf("salt asked again: %d calls", calls)
}
}
+
+// TestRecorderMethodBounded: the stored method is one of a fixed few words, so
+// a request line's extension method (any length, any token characters) costs
+// at most "OTHER" in a record's field and message.
+func TestRecorderMethodBounded(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ huge := strings.Repeat("&", 100000)
+ in := []string{"GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "CONNECT", "TRACE", "get", "PROPFIND", huge, ""}
+ want := []string{"GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "CONNECT", "TRACE", "OTHER", "OTHER", "OTHER", "OTHER"}
+ for i, m := range in {
+ r.Response(Response{Acct: "a1", App: "shop", Method: m, Path: "/x", Status: 404, When: now})
+ if got := s.recs[i]["method"]; got != want[i] {
+ t.Errorf("method %.20q stored as %q, want %q", m, got, want[i])
+ }
+ if got := s.msgs[i]; got != "404 "+want[i]+" /x" {
+ t.Errorf("message %.40q", got)
+ }
+ }
+ r.Match(Match{Response: Response{Acct: "a1", App: "shop", Method: huge, Path: "/y", Status: 403, When: now},
+ Kind: KindBlock, Tier: TierHard, RuleID: "r"})
+ if got := s.msgs[len(s.msgs)-1]; got != "block hard r OTHER /y" || s.recs[len(s.recs)-1]["method"] != "OTHER" {
+ t.Errorf("match message %.60q", got)
+ }
+ r.Flush(now.Add(time.Minute))
+ min := s.recs[len(s.recs)-1]
+ wantCounts := map[string]string{"m_get": "1", "m_head": "1", "m_post": "1", "m_put": "1", "m_patch": "1",
+ "m_delete": "1", "m_options": "1", "m_other": "7"}
+ for k, v := range wantCounts {
+ if min[k] != v {
+ t.Errorf("%s = %q, want %q", k, min[k], v)
+ }
+ }
+}
+
+// TestRecorderEnvironment: a record names the route's environment, which a
+// route can later change, so it cannot be worked out afterwards.
+func TestRecorderEnvironment(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ r.Response(Response{Acct: "a1", App: "shop", Environment: "staging", Method: "GET", Path: "/x", Status: 404, When: now})
+ r.Match(Match{Response: Response{Acct: "a1", App: "shop", Environment: "staging", Method: "GET", Path: "/y", Status: 403, When: now},
+ Kind: KindWatch, Tier: TierAttack, RuleID: "r"})
+ for i, f := range s.recs {
+ if f["env"] != "staging" {
+ t.Errorf("record %d env = %q: %v", i, f["env"], f)
+ }
+ }
+}
+
+// TestRecorderBlocksNotCrowdedOutByWatches: blocks and watches have their own
+// allowances, so a minute of watch matches cannot take a block's record.
+func TestRecorderBlocksNotCrowdedOutByWatches(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ watch := Match{Response: Response{Acct: "a1", App: "shop", Method: "GET", Path: "/w", Status: 200, When: now},
+ Kind: KindWatch, Tier: TierAttack, RuleID: "w"}
+ for i := 0; i < WatchPerMin+10; i++ {
+ r.Match(watch)
+ }
+ block := watch
+ block.Kind, block.RuleID, block.Path = KindBlock, "b", "/b"
+ r.Match(block)
+ r.Flush(now.Add(time.Minute))
+ var watches, blocks int
+ var min map[string]string
+ for _, f := range s.recs {
+ switch f["kind"] {
+ case "watch":
+ watches++
+ case "block":
+ blocks++
+ case "minute":
+ min = f
+ }
+ }
+ if watches != WatchPerMin || blocks != 1 {
+ t.Fatalf("watch records %d (cap %d), block records %d", watches, WatchPerMin, blocks)
+ }
+ if min["dropped_watch"] != "10" || min["dropped_block"] != "0" || min["watch"] != fmt.Sprint(WatchPerMin+10) || min["block"] != "1" {
+ t.Fatalf("minute %v", min)
+ }
+}
+
+// TestRecorderSummariesGoThroughEdgeSummary: minute records take the sink's
+// summary door, which no rate budget can refuse, and request records the
+// ordinary one.
+func TestRecorderSummariesGoThroughEdgeSummary(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ r.Response(Response{Acct: "a1", App: "shop", Method: "GET", Path: "/x", Status: 404, When: now})
+ r.Response(Response{Acct: "a1", App: "blog", Method: "GET", Path: "/x", Status: 200, When: now})
+ if s.summaries != 0 {
+ t.Fatalf("a request record went through EdgeSummary")
+ }
+ r.Flush(now.Add(time.Minute))
+ if s.summaries != 2 || len(s.recs) != 3 {
+ t.Fatalf("summaries %d of %d records", s.summaries, len(s.recs))
+ }
+}
+
+// TestRecorderStampsZeroWhen: the edge leaves When unset and the recorder
+// stamps it with its own clock, so the minute a request lands in is the
+// recorder's minute.
+func TestRecorderStampsZeroWhen(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 59, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, func(day string) []byte { return []byte("salt-" + day) }, func() time.Time { return now })
+ for i := 0; i < NotFoundPerMin+5; i++ {
+ r.Response(Response{Acct: "a1", App: "shop", Method: "GET", Path: "/x", ClientIP: "203.0.113.7", Status: 404})
+ }
+ // The clock crosses the minute: a new minute, with its own allowance.
+ now = now.Add(time.Second)
+ for i := 0; i < 3; i++ {
+ r.Response(Response{Acct: "a1", App: "shop", Method: "GET", Path: "/x", ClientIP: "203.0.113.7", Status: 404})
+ }
+ r.Flush(now.Add(time.Minute))
+ var nf int
+ var mins []map[string]string
+ for _, f := range s.recs {
+ switch f["kind"] {
+ case "notfound":
+ nf++
+ if len(f["cid"]) != 12 {
+ t.Fatalf("no client hash: the stamped time found no salt day: %v", f)
+ }
+ case "minute":
+ mins = append(mins, f)
+ }
+ }
+ if nf != NotFoundPerMin+3 || len(mins) != 2 {
+ t.Fatalf("notfound %d, minutes %d", nf, len(mins))
+ }
+ if mins[0]["minute"] != "2026-10-24T21:07:00Z" || mins[0]["s404"] != fmt.Sprint(NotFoundPerMin+5) || mins[0]["dropped_notfound"] != "5" ||
+ mins[1]["minute"] != "2026-10-24T21:08:00Z" || mins[1]["s404"] != "3" || mins[1]["dropped_notfound"] != "0" {
+ t.Fatalf("minutes %v", mins)
+ }
+}
+
+// TestRecorderFragmentControlsEscaped: the fragment a caller hands in is
+// request text too, so the record boundary escapes its controls whatever the
+// caller did.
+func TestRecorderFragmentControlsEscaped(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ r.Match(Match{Response: Response{Acct: "a1", App: "shop", Method: "GET", Path: "/p\x1b[2J", Status: 403, When: now},
+ Kind: KindBlock, Tier: TierAttack, RuleID: "r", Fragment: "x\ny\x1b"})
+ f := s.recs[0]
+ if f["frag"] != `x\x0ay\x1b` || f["path"] != `/p\x1b[2J` || strings.ContainsAny(s.msgs[0], "\x1b\n") {
+ t.Fatalf("frag %q path %q msg %q", f["frag"], f["path"], s.msgs[0])
+ }
+}
diff --git a/internal/reqrec/redact_test.go b/internal/reqrec/redact_test.go
index ea3d12b45..ea6bc90f2 100644
--- a/internal/reqrec/redact_test.go
+++ b/internal/reqrec/redact_test.go
@@ -23,13 +23,13 @@ func TestMaskPath(t *testing.T) {
"/blog/Export-Attendees-From-All-Events": "/blog/Export-Attendees-From-All-Events",
// Final review C1: a long run with no digit, standard base64 split
// into short pieces at '/', and long hex beside a non-hex letter.
- "/reset/rKpWmTzQvLsNxYhJbCfDaGeS": "/reset/[masked]",
- "/reset/Xq3vB9kLmN2pQr7s/T1uVwYz4aBcDeFg": "/reset/[masked]",
- "/zabcdefabcdefabcdefabcdefabcdefab": "/[masked]",
+ "/reset/rKpWmTzQvLsNxYhJbCfDaGeS": "/reset/[masked]",
+ "/reset/Xq3vB9kLmN2pQr7s/T1uVwYz4aBcDeFg": "/reset/[masked]",
+ "/zabcdefabcdefabcdefabcdefabcdefab": "/[masked]",
"/a/Xq3vB9kL+mN2pQr7s/T1uVwYz4aBcDeFg=/b.js": "/a/[masked]/b.js",
// Title-case segments are words, not a split token.
"/en/Products/iPhone/Accessories/Cases": "/en/Products/iPhone/Accessories/Cases",
- "/api/v2/users/12345/orders/67890": "/api/v2/users/12345/orders/67890",
+ "/api/v2/users/12345/orders/67890": "/api/v2/users/12345/orders/67890",
}
for in, want := range cases {
if got := MaskPath(in); got != want {
@@ -158,7 +158,7 @@ func TestControlsEscaped(t *testing.T) {
t.Errorf("QueryNames = %q, want %q", got, want)
}
for _, in := range []struct {
- src Source
+ src Source
name, val string
}{
{SourceQuery, "q", "x\ny\x1bz"},