From d5ae8adea5e52418f1c89e042ea700675c269252 Mon Sep 17 00:00:00 2001 From: Matthew DeVenny Date: Mon, 29 Jun 2026 10:01:39 -0700 Subject: [PATCH 1/7] #75 exclude pre-ready events Signed-off-by: Matthew DeVenny --- cmd/github_actions_handler.go | 5 +- cmd/github_actions_handler_test.go | 75 ++++++++ cmd/start.go | 14 +- pkg/events/events.go | 99 +++++++++- pkg/events/events_test.go | 287 +++++++++++++++++++++++++++-- 5 files changed, 452 insertions(+), 28 deletions(-) create mode 100644 cmd/github_actions_handler_test.go diff --git a/cmd/github_actions_handler.go b/cmd/github_actions_handler.go index f9c2d2d..7741db0 100644 --- a/cmd/github_actions_handler.go +++ b/cmd/github_actions_handler.go @@ -77,7 +77,10 @@ func (h *GitHubActionsHandler) Handle(_ context.Context, r slog.Record) error { return true }) - fmt.Fprintf(os.Stderr, "%s%s\n", prefix, sb.String()) + // Leading timestamp on every line. Placed after the prefix so GitHub still + // parses the ::error::/::warning:: workflow command at column 0. Millisecond + // precision because the startup race this surfaces is sub-second. + fmt.Fprintf(os.Stderr, "%s%s %s\n", prefix, r.Time.Format("2006-01-02 15:04:05.000"), sb.String()) return nil } diff --git a/cmd/github_actions_handler_test.go b/cmd/github_actions_handler_test.go new file mode 100644 index 0000000..7b602c3 --- /dev/null +++ b/cmd/github_actions_handler_test.go @@ -0,0 +1,75 @@ +// Copyright 2026 BoxBuild Inc DBA CodeCargo +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build linux + +package cmd + +import ( + "context" + "io" + "log/slog" + "os" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// captureStderr redirects os.Stderr around fn and returns what was written. The +// GitHubActionsHandler writes to os.Stderr directly, so the test swaps the global +// for the duration of the call. +func captureStderr(t *testing.T, fn func()) string { + t.Helper() + old := os.Stderr + r, w, err := os.Pipe() + require.NoError(t, err) + os.Stderr = w + defer func() { os.Stderr = old }() + + fn() + require.NoError(t, w.Close()) + data, err := io.ReadAll(r) + require.NoError(t, err) + return string(data) +} + +// TestGitHubActionsHandler_TimestampPrefix verifies every emitted line carries a +// leading millisecond timestamp, and that for annotated levels the ::error:: / +// ::warning:: workflow command stays at column 0 (timestamp after the prefix). +func TestGitHubActionsHandler_TimestampPrefix(t *testing.T) { + h := NewGitHubActionsHandler(false) + ts := time.Date(2026, 6, 29, 15, 24, 8, 456_000_000, time.UTC) + + out := captureStderr(t, func() { + info := slog.NewRecord(ts, slog.LevelInfo, "Connection blocked", 0) + info.AddAttrs(slog.Bool("pre_ready", true)) + require.NoError(t, h.Handle(context.Background(), info)) + + errRec := slog.NewRecord(ts, slog.LevelError, "boom", 0) + require.NoError(t, h.Handle(context.Background(), errRec)) + }) + + lines := strings.Split(strings.TrimRight(out, "\n"), "\n") + require.Len(t, lines, 2) + + // Info line: " ", no prefix. + assert.Regexp(t, `^2026-06-29 15:24:08\.456 Connection blocked\b`, lines[0]) + assert.Contains(t, lines[0], "pre_ready=true") + + // Error line: prefix stays at column 0, timestamp follows it. + assert.Regexp(t, `^::error::2026-06-29 15:24:08\.456 boom\b`, lines[1]) +} diff --git a/cmd/start.go b/cmd/start.go index dab54a3..669052c 100644 --- a/cmd/start.go +++ b/cmd/start.go @@ -309,8 +309,14 @@ func StartCargoWall(cmd *StartCmd, hooks *StartHooks) error { } defer rd.Close() + // Readiness boundary for startup-race suppression: blocked events stamped + // (CLOCK_MONOTONIC) before MarkReady are drops from the attach-before-program + // window and are excluded from the audit log / SaaS / summary / notifications. + // MarkReady runs once, just before the ready signal is published. + readiness := events.NewReadiness() + // Start processors before attaching programs - go events.ProcessBlockedEvents(rd, configMgr, notificationTracker, auditLogger, fw, logger) + go events.ProcessBlockedEvents(rd, configMgr, notificationTracker, auditLogger, fw, readiness, logger) // Attach TC programs egressLink, err := tc.AttachEgress(ifname, objs.TcEgress, logger) @@ -455,6 +461,12 @@ func StartCargoWall(cmd *StartCmd, hooks *StartHooks) error { }() } + // Record the readiness boundary BEFORE publishing the ready signal, so the + // happens-before is: boundary recorded → sentinel/hook fires → CI proceeds → + // user traffic. A user-triggered block is therefore stamped after the + // boundary and correctly reported; it can never be misclassified as pre-ready. + readiness.MarkReady() + if hooks != nil && hooks.Ready != nil { if err := hooks.Ready(); err != nil { return fmt.Errorf("ready hook failed: %w", err) diff --git a/pkg/events/events.go b/pkg/events/events.go index 51b74e7..697309a 100644 --- a/pkg/events/events.go +++ b/pkg/events/events.go @@ -25,6 +25,7 @@ import ( "os" "strings" "sync" + "sync/atomic" "time" "unsafe" @@ -40,6 +41,55 @@ type FirewallUpdater interface { AddIP(ip net.IP, action config.Action, ports []config.Port) (bool, error) } +// Readiness records the CLOCK_MONOTONIC instant at which CargoWall finished +// programming its allow rules and published the ready signal. Blocked events +// whose BPF timestamp (bpf_ktime_get_ns, also CLOCK_MONOTONIC) predates that +// instant are startup-race drops — the TC program attaches default-deny before +// the auto-allow infra rules land, so a connection in that window is dropped +// once and self-heals on TCP retransmit. They are not policy decisions, so they +// are excluded from the audit log (hence the SaaS push + summary) and from block +// notifications. The zero value is "not ready yet": every event before MarkReady +// is treated as pre-ready. +type Readiness struct { + readyMono atomic.Uint64 // CLOCK_MONOTONIC ns at ready; 0 = not yet ready +} + +// NewReadiness returns a Readiness in the not-yet-ready state. +func NewReadiness() *Readiness { return &Readiness{} } + +// MarkReady records the current CLOCK_MONOTONIC instant as the readiness +// boundary. Call exactly once, immediately before publishing the ready signal, +// so it happens-before any externally triggered (e.g. user) traffic. +func (r *Readiness) MarkReady() { + var ts unix.Timespec + if err := unix.ClockGettime(unix.CLOCK_MONOTONIC, &ts); err != nil { + // Fail open for reporting: never leave readyMono at 0 (which would + // suppress every block forever). A 1ns boundary suppresses nothing + // real — every genuine bpf_ktime_get_ns stamp is far larger. + r.readyMono.Store(1) + return + } + r.readyMono.Store(uint64(ts.Nano())) +} + +// isPreReady reports whether a blocked event with the given CLOCK_MONOTONIC BPF +// timestamp occurred before the readiness boundary. It is nil-safe, and a zero +// event timestamp never suppresses, so a missing/garbage field can't mask a real +// block. +func (r *Readiness) isPreReady(eventMono uint64) bool { + if r == nil { + return false + } + readyMono := r.readyMono.Load() + if readyMono == 0 { // MarkReady not run yet → everything so far is pre-ready + return true + } + if eventMono == 0 { // defensive: never suppress on a zero/garbage stamp + return false + } + return eventMono < readyMono +} + // lookupProcessName reads the process name from /proc//comm. // Returns empty string if the process no longer exists or can't be read. func lookupProcessName(pid uint32) string { @@ -244,7 +294,7 @@ func dstPortAllowedByRule(dstPort uint16, proto uint8, ports []config.Port) bool // processEvent handles a single blocked/allowed event from the ring buffer. func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *NotificationTracker, - auditLogger *AuditLogger, fw FirewallUpdater, logger *slog.Logger, + auditLogger *AuditLogger, fw FirewallUpdater, readiness *Readiness, logger *slog.Logger, ) { if len(raw) < int(unsafe.Sizeof(BpfBlockedEvent{})) { return @@ -452,6 +502,24 @@ func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *No } else if event.SrcPort == 0 && event.DstPort < 256 { // Non-TCP/UDP protocol block — dst_port contains the protocol number protocolName := getProtocolName(uint8(event.DstPort)) + + if readiness.isPreReady(event.Timestamp) { + // Startup-race drop (see Readiness): keep it in the process log for + // forensics, but exclude it from the audit log (→ SaaS push + + // summary) and the notification — it is not a policy decision. + logConnEvent(logger, "Protocol blocked", cnameChain, + "src", srcIP, + "dst", displayHostname, + "dst_ip", dstIP, + "protocol", protocolName, + "protocol_num", event.DstPort, + "process", comm, + "pid", pid, + "pre_ready", true, + "reported", false) + return + } + logConnEvent(logger, "Protocol blocked", cnameChain, "src", srcIP, "dst", displayHostname, @@ -459,8 +527,7 @@ func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *No "protocol", protocolName, "protocol_num", event.DstPort, "process", comm, - "pid", pid, - "timestamp", time.Now().Format("2006-01-02 15:04:05")) + "pid", pid) // Log to audit file if configured if auditLogger != nil { @@ -496,14 +563,32 @@ func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *No } } else { // Blocked TCP SYN or UDP connection + if readiness.isPreReady(event.Timestamp) { + // Startup-race drop (see Readiness): the TC program attached + // default-deny before the auto-allow infra rules were programmed, so + // this was dropped once and self-heals on retransmit. Keep it in the + // process log for forensics, but exclude it from the audit log (→ + // SaaS push + summary) and the notification — it is not a policy + // decision. + logConnEvent(logger, "Connection blocked", cnameChain, + "src", fmt.Sprintf("%s:%d", srcIP, event.SrcPort), + "dst", displayHostname, + "dst_ip", dstIP, + "dst_port", event.DstPort, + "process", comm, + "pid", pid, + "pre_ready", true, + "reported", false) + return + } + logConnEvent(logger, "Connection blocked", cnameChain, "src", fmt.Sprintf("%s:%d", srcIP, event.SrcPort), "dst", displayHostname, "dst_ip", dstIP, "dst_port", event.DstPort, "process", comm, - "pid", pid, - "timestamp", time.Now().Format("2006-01-02 15:04:05")) + "pid", pid) // Log to audit file if configured if auditLogger != nil { @@ -539,7 +624,7 @@ func logConnEvent(logger *slog.Logger, msg string, cnameChain []string, attrs .. } // ProcessBlockedEvents processes blocked connection events -func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notificationTracker *NotificationTracker, auditLogger *AuditLogger, fw FirewallUpdater, logger *slog.Logger) { +func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notificationTracker *NotificationTracker, auditLogger *AuditLogger, fw FirewallUpdater, readiness *Readiness, logger *slog.Logger) { for { record, err := rd.Read() if err != nil { @@ -550,6 +635,6 @@ func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notific continue } - processEvent(record.RawSample, configMgr, notificationTracker, auditLogger, fw, logger) + processEvent(record.RawSample, configMgr, notificationTracker, auditLogger, fw, readiness, logger) } } diff --git a/pkg/events/events_test.go b/pkg/events/events_test.go index 83e21d6..a843f5f 100644 --- a/pkg/events/events_test.go +++ b/pkg/events/events_test.go @@ -510,7 +510,7 @@ func TestProcessEvent_IPv4BlockedTCP(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, nil, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -546,7 +546,7 @@ func TestProcessEvent_IPv4AllowedTCP(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -579,7 +579,7 @@ func TestProcessEvent_ProtocolBlocked(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, nil, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -587,6 +587,255 @@ func TestProcessEvent_ProtocolBlocked(t *testing.T) { assert.Equal(t, "ICMP", events[0].Protocol) } +// --- Readiness gate: pre-ready startup-race drops (issue: suppress) --- + +// blockTestDeps bundles the fixtures the readiness-gate tests share: a +// temp-file audit logger, a state-machine mock + tracker, and a default-deny +// config manager. +type blockTestDeps struct { + auditLogger *AuditLogger + smClient *mockStateMachineClient + tracker *NotificationTracker + cm *config.Manager + auditPath string +} + +func newBlockTestDeps(t *testing.T) blockTestDeps { + t.Helper() + auditPath := filepath.Join(t.TempDir(), "audit.jsonl") + auditLogger, err := NewAuditLogger(auditPath, false) + require.NoError(t, err) + t.Cleanup(func() { auditLogger.Close() }) + + smClient := &mockStateMachineClient{} + cm := config.NewConfigManager() + require.NoError(t, cm.LoadConfigFromRules(nil, config.ActionDeny)) + + return blockTestDeps{ + auditLogger: auditLogger, + smClient: smClient, + tracker: NewNotificationTracker(smClient, newTestLogger()), + cm: cm, + auditPath: auditPath, + } +} + +// markReverseDNSAttempted resets the reverse-DNS cache (avoiding cross-test +// pollution) and pre-seeds the given IPs as already-attempted, so processEvent +// skips the real PTR lookup — keeping these tests hermetic and fast. +func markReverseDNSAttempted(ips ...string) { + reverseDNSMu.Lock() + reverseDNSCache = make(map[string]time.Time) + for _, ip := range ips { + reverseDNSCache[ip] = time.Now() + } + reverseDNSMu.Unlock() +} + +// makeBlockedTCPv4Event builds a blocked IPv4 TCP SYN to 93.184.216.34:443 with +// the given BPF (CLOCK_MONOTONIC) timestamp. +func makeBlockedTCPv4Event(ts uint64) []byte { + return makeBpfEvent(BpfBlockedEvent{ + IpVersion: 4, + Allowed: 0, + IpProto: unix.IPPROTO_TCP, + SrcIp: ipv4ToUint32("10.0.0.1"), + DstIp: ipv4ToUint32("93.184.216.34"), + SrcPort: 54321, + DstPort: 443, + Timestamp: ts, + }) +} + +func TestProcessEvent_PreReadyBlockedTCPSuppressed(t *testing.T) { + d := newBlockTestDeps(t) + r := NewReadiness() + r.readyMono.Store(1_000_000) // 1ms boundary + + markReverseDNSAttempted("93.184.216.34") + processEvent(makeBlockedTCPv4Event(500_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) + + assert.Empty(t, readAuditEvents(t, d.auditPath), "pre-ready block must not be written to the audit log") + assert.Empty(t, d.smClient.calls, "pre-ready block must not notify") +} + +func TestProcessEvent_PostReadyBlockedTCPReported(t *testing.T) { + d := newBlockTestDeps(t) + r := NewReadiness() + r.readyMono.Store(1_000_000) + + markReverseDNSAttempted("93.184.216.34") + processEvent(makeBlockedTCPv4Event(2_000_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) + + events := readAuditEvents(t, d.auditPath) + require.Len(t, events, 1) + assert.Equal(t, EventConnectionBlocked, events[0].EventType) + assert.Len(t, d.smClient.calls, 1, "post-ready block must notify") +} + +func TestProcessEvent_NotYetReadyBlockedSuppressed(t *testing.T) { + d := newBlockTestDeps(t) + r := NewReadiness() // readyMono == 0: not ready yet + + markReverseDNSAttempted("93.184.216.34") + processEvent(makeBlockedTCPv4Event(2_000_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) + + assert.Empty(t, readAuditEvents(t, d.auditPath), "blocks before MarkReady must be suppressed") + assert.Empty(t, d.smClient.calls) +} + +func TestProcessEvent_ZeroEventTimestampNotSuppressed(t *testing.T) { + d := newBlockTestDeps(t) + r := NewReadiness() + r.readyMono.Store(1_000_000) + + markReverseDNSAttempted("93.184.216.34") + // Defensive: a zero/garbage BPF stamp must never mask a real block. + processEvent(makeBlockedTCPv4Event(0), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) + + events := readAuditEvents(t, d.auditPath) + require.Len(t, events, 1) + assert.Equal(t, EventConnectionBlocked, events[0].EventType) +} + +func TestProcessEvent_NilReadinessNeverSuppresses(t *testing.T) { + d := newBlockTestDeps(t) + + markReverseDNSAttempted("93.184.216.34") + processEvent(makeBlockedTCPv4Event(500_000), d.cm, d.tracker, d.auditLogger, nil, nil, newTestLogger()) + + events := readAuditEvents(t, d.auditPath) + require.Len(t, events, 1) + assert.Equal(t, EventConnectionBlocked, events[0].EventType) + assert.Len(t, d.smClient.calls, 1) +} + +func makeBlockedICMPv4Event(ts uint64) []byte { + return makeBpfEvent(BpfBlockedEvent{ + IpVersion: 4, + Allowed: 0, + IpProto: unix.IPPROTO_ICMP, + SrcIp: ipv4ToUint32("10.0.0.1"), + DstIp: ipv4ToUint32("10.0.0.2"), + SrcPort: 0, + DstPort: 1, // ICMP + Timestamp: ts, + }) +} + +func TestProcessEvent_PreReadyProtocolBlockedSuppressed(t *testing.T) { + d := newBlockTestDeps(t) + r := NewReadiness() + r.readyMono.Store(1_000_000) + + markReverseDNSAttempted("10.0.0.2") + processEvent(makeBlockedICMPv4Event(500_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) + + assert.Empty(t, readAuditEvents(t, d.auditPath), "pre-ready protocol block must not be written to the audit log") + assert.Empty(t, d.smClient.calls) +} + +func TestProcessEvent_PostReadyProtocolBlockedReported(t *testing.T) { + d := newBlockTestDeps(t) + r := NewReadiness() + r.readyMono.Store(1_000_000) + + markReverseDNSAttempted("10.0.0.2") + processEvent(makeBlockedICMPv4Event(2_000_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) + + events := readAuditEvents(t, d.auditPath) + require.Len(t, events, 1) + assert.Equal(t, EventProtocolBlocked, events[0].EventType) +} + +func TestProcessEvent_PreReadyDoesNotSuppressAllowed(t *testing.T) { + d := newBlockTestDeps(t) + r := NewReadiness() + r.readyMono.Store(1_000_000) // boundary above the event's timestamp + + raw := makeBpfEvent(BpfBlockedEvent{ + IpVersion: 4, + Allowed: 1, // allow outcome — must never be gated + IpProto: unix.IPPROTO_TCP, + SrcIp: ipv4ToUint32("10.0.0.1"), + DstIp: ipv4ToUint32("93.184.216.34"), + SrcPort: 54321, + DstPort: 443, + Timestamp: 500_000, + }) + + markReverseDNSAttempted("93.184.216.34") + processEvent(raw, d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) + + events := readAuditEvents(t, d.auditPath) + require.Len(t, events, 1) + assert.Equal(t, EventConnectionAllowed, events[0].EventType, "the readiness gate must only apply to blocked branches") +} + +func TestProcessEvent_PreReadyBlockedIPv6Suppressed(t *testing.T) { + d := newBlockTestDeps(t) + r := NewReadiness() + r.readyMono.Store(1_000_000) + + var dstIp6 [16]byte + copy(dstIp6[:], net.ParseIP("2001:db8::dead").To16()) + raw := makeBpfEvent(BpfBlockedEvent{ + IpVersion: 6, + Allowed: 0, + IpProto: unix.IPPROTO_TCP, + DstIp6: dstIp6, + SrcPort: 54321, + DstPort: 443, + Timestamp: 500_000, + }) + + markReverseDNSAttempted("2001:db8::dead") + processEvent(raw, d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) + + assert.Empty(t, readAuditEvents(t, d.auditPath), "pre-ready IPv6 block must be suppressed (v6 parity)") + assert.Empty(t, d.smClient.calls) +} + +func TestReadiness_isPreReady(t *testing.T) { + tests := []struct { + name string + readyMono uint64 + eventMono uint64 + want bool + }{ + {"not ready yet suppresses everything", 0, 123, true}, + {"event before boundary", 100, 50, true}, + {"event at boundary is reported (exclusive)", 100, 100, false}, + {"event after boundary", 100, 150, false}, + {"zero event timestamp never suppresses", 100, 0, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := NewReadiness() + r.readyMono.Store(tt.readyMono) + assert.Equal(t, tt.want, r.isPreReady(tt.eventMono)) + }) + } + + t.Run("nil receiver never suppresses", func(t *testing.T) { + var r *Readiness + assert.False(t, r.isPreReady(50)) + }) +} + +func TestReadiness_MarkReadySetsMonotonic(t *testing.T) { + r := NewReadiness() + require.Equal(t, uint64(0), r.readyMono.Load()) + + r.MarkReady() + stored := r.readyMono.Load() + assert.NotZero(t, stored, "MarkReady must record a non-zero boundary") + + var ts unix.Timespec + require.NoError(t, unix.ClockGettime(unix.CLOCK_MONOTONIC, &ts)) + assert.GreaterOrEqual(t, uint64(ts.Nano()), stored, "a later monotonic read must be >= the boundary") +} + func TestProcessEvent_LateResolvedIPAddedToFirewall(t *testing.T) { cm := config.NewConfigManager() require.NoError(t, cm.LoadConfigFromRules([]config.Rule{ @@ -612,7 +861,7 @@ func TestProcessEvent_LateResolvedIPAddedToFirewall(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, nil, fw, newTestLogger()) + processEvent(raw, cm, nil, nil, fw, nil, newTestLogger()) require.Len(t, fw.addedIPs, 1) assert.Equal(t, config.ActionAllow, fw.addedIPs[0].action) @@ -648,7 +897,7 @@ func TestProcessEvent_LateResolvedIPInheritsRulePorts(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, nil, fw, newTestLogger()) + processEvent(raw, cm, nil, nil, fw, nil, newTestLogger()) require.Len(t, fw.addedIPs, 1) assert.Equal(t, config.ActionAllow, fw.addedIPs[0].action) @@ -687,7 +936,7 @@ func TestProcessEvent_LateResolvedIPv6InheritsRulePorts(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, nil, fw, newTestLogger()) + processEvent(raw, cm, nil, nil, fw, nil, newTestLogger()) require.Len(t, fw.addedIPs, 1) assert.Equal(t, config.ActionAllow, fw.addedIPs[0].action) @@ -730,7 +979,7 @@ func TestProcessEvent_LateResolvedEmitsLateAllowedAudit(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -786,7 +1035,7 @@ func TestProcessEvent_LateResolvedMatchedRuleIsRulePatternNotHostname(t *testing reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, fw, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, fw, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -831,7 +1080,7 @@ func TestProcessEvent_BlockedNoMatchStillLogsBlocked(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) require.Empty(t, fw.addedIPs, "late-add must not fire when no allow rule matches") @@ -878,7 +1127,7 @@ func TestProcessEvent_CNAMEDerivedAttributesToOrigin(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -930,7 +1179,7 @@ func TestProcessEvent_CNAMEDerivedBlockedAttributesToOrigin(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, fw, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, fw, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -982,7 +1231,7 @@ func TestProcessEvent_LateResolvedDstPortNotInRulePortsStaysBlocked(t *testing.T reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) // AddIP still fires (so future port-443 retries succeed), but this // connection's audit/notification must reflect the genuine block. @@ -1031,7 +1280,7 @@ func TestProcessEvent_LateResolvedUDPEventNotAllowedByTCPRule(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1077,7 +1326,7 @@ func TestProcessEvent_LateResolvedTCPEventAllowedByTCPRule(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1121,7 +1370,7 @@ func TestProcessEvent_LateResolvedTCPEventAllowedByProtocolAllRule(t *testing.T) reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1159,7 +1408,7 @@ func TestProcessEvent_IPv6Event(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1172,7 +1421,7 @@ func TestProcessEvent_TooShortBuffer(t *testing.T) { require.NoError(t, cm.LoadConfigFromRules(nil, config.ActionDeny)) // Should not panic — processEvent should return early for short buffers - processEvent([]byte{0x04}, cm, nil, nil, nil, newTestLogger()) + processEvent([]byte{0x04}, cm, nil, nil, nil, nil, newTestLogger()) } func TestProcessEvent_AutoAllowedType(t *testing.T) { @@ -1200,7 +1449,7 @@ func TestProcessEvent_AutoAllowedType(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1236,7 +1485,7 @@ func TestProcessEvent_NoAutoAllowedTypeForUserRule(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) From 6c0276fea7d9a81800e1baa35f1b63e4dc2f93c3 Mon Sep 17 00:00:00 2001 From: Matthew DeVenny Date: Mon, 29 Jun 2026 10:12:10 -0700 Subject: [PATCH 2/7] #75 clarify isPreReady zero-stamp semantics The "a zero event timestamp never suppresses" guarantee only holds after MarkReady. Before readiness (readyMono == 0) the whole window is pre-ready, so a zero/garbage-stamped event is suppressed too. Fix the doc comment to state the precedence accurately and add a (readyMono=0, eventMono=0) case. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Matthew DeVenny --- pkg/events/events.go | 13 ++++++++++--- pkg/events/events_test.go | 1 + 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/pkg/events/events.go b/pkg/events/events.go index 697309a..ee0dccb 100644 --- a/pkg/events/events.go +++ b/pkg/events/events.go @@ -73,9 +73,16 @@ func (r *Readiness) MarkReady() { } // isPreReady reports whether a blocked event with the given CLOCK_MONOTONIC BPF -// timestamp occurred before the readiness boundary. It is nil-safe, and a zero -// event timestamp never suppresses, so a missing/garbage field can't mask a real -// block. +// timestamp occurred before the readiness boundary (and so should be excluded +// from reporting). Precedence: +// - a nil Readiness never suppresses (reporting is unaffected when the gate +// isn't wired up); +// - before MarkReady (readyMono == 0) the firewall is still being programmed, +// so the whole window is pre-ready and every event is suppressed — including +// one with a zero/garbage stamp; +// - after MarkReady, a zero event timestamp never suppresses, so a +// missing/garbage field can't mask a real (post-ready) block; otherwise the +// event is pre-ready iff its stamp precedes the boundary. func (r *Readiness) isPreReady(eventMono uint64) bool { if r == nil { return false diff --git a/pkg/events/events_test.go b/pkg/events/events_test.go index a843f5f..8c2fdef 100644 --- a/pkg/events/events_test.go +++ b/pkg/events/events_test.go @@ -804,6 +804,7 @@ func TestReadiness_isPreReady(t *testing.T) { want bool }{ {"not ready yet suppresses everything", 0, 123, true}, + {"not ready yet suppresses even a zero stamp", 0, 0, true}, {"event before boundary", 100, 50, true}, {"event at boundary is reported (exclusive)", 100, 100, false}, {"event after boundary", 100, 150, false}, From 1920771ae7596ce4e5fcba51c9fb12c069b25fad Mon Sep 17 00:00:00 2001 From: Matthew DeVenny Date: Mon, 29 Jun 2026 10:55:42 -0700 Subject: [PATCH 3/7] #75 program BPF maps before attach; drop the pre-ready reporting gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the reporting-layer suppression with a root-cause fix. Attach the TC egress program only AFTER the default action, static allowlist, auto-allow infrastructure rules, tracked hostnames, and existing connections are programmed into the maps (the maps exist independently of the link). The firewall is therefore correct from its first enforced packet, so the attach-before-program race that produced spurious "Connection blocked" drops (e.g. the Azure wireserver on :32526) no longer happens — there is nothing to suppress, and real startup-window blocks stay honest, reported policy decisions. Removes the Readiness type/gate/MarkReady and its processEvent threading (now unnecessary) and the readiness tests. Keeps the per-line process-log timestamps (the redundant per-event timestamp= field stays removed). Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Matthew DeVenny --- cmd/github_actions_handler_test.go | 4 +- cmd/start.go | 37 ++-- pkg/events/events.go | 100 +--------- pkg/events/events_test.go | 288 ++--------------------------- 4 files changed, 40 insertions(+), 389 deletions(-) diff --git a/cmd/github_actions_handler_test.go b/cmd/github_actions_handler_test.go index 7b602c3..effd8b9 100644 --- a/cmd/github_actions_handler_test.go +++ b/cmd/github_actions_handler_test.go @@ -56,7 +56,7 @@ func TestGitHubActionsHandler_TimestampPrefix(t *testing.T) { out := captureStderr(t, func() { info := slog.NewRecord(ts, slog.LevelInfo, "Connection blocked", 0) - info.AddAttrs(slog.Bool("pre_ready", true)) + info.AddAttrs(slog.Int("dst_port", 443)) require.NoError(t, h.Handle(context.Background(), info)) errRec := slog.NewRecord(ts, slog.LevelError, "boom", 0) @@ -68,7 +68,7 @@ func TestGitHubActionsHandler_TimestampPrefix(t *testing.T) { // Info line: " ", no prefix. assert.Regexp(t, `^2026-06-29 15:24:08\.456 Connection blocked\b`, lines[0]) - assert.Contains(t, lines[0], "pre_ready=true") + assert.Contains(t, lines[0], "dst_port=443") // Error line: prefix stays at column 0, timestamp follows it. assert.Regexp(t, `^::error::2026-06-29 15:24:08\.456 boom\b`, lines[1]) diff --git a/cmd/start.go b/cmd/start.go index 669052c..4d48da6 100644 --- a/cmd/start.go +++ b/cmd/start.go @@ -309,21 +309,8 @@ func StartCargoWall(cmd *StartCmd, hooks *StartHooks) error { } defer rd.Close() - // Readiness boundary for startup-race suppression: blocked events stamped - // (CLOCK_MONOTONIC) before MarkReady are drops from the attach-before-program - // window and are excluded from the audit log / SaaS / summary / notifications. - // MarkReady runs once, just before the ready signal is published. - readiness := events.NewReadiness() - - // Start processors before attaching programs - go events.ProcessBlockedEvents(rd, configMgr, notificationTracker, auditLogger, fw, readiness, logger) - - // Attach TC programs - egressLink, err := tc.AttachEgress(ifname, objs.TcEgress, logger) - if err != nil { - return fmt.Errorf("failed to attach TC egress: %w", err) - } - defer egressLink.Close() + // Start the event reader before the firewall is enforcing so no events are missed. + go events.ProcessBlockedEvents(rd, configMgr, notificationTracker, auditLogger, fw, logger) // Set default action through firewall defaultAction := configMgr.GetDefaultAction() @@ -392,6 +379,20 @@ func StartCargoWall(cmd *StartCmd, hooks *StartHooks) error { } } + // Attach the TC egress program only AFTER the maps are fully programmed + // (default action + static allowlist + auto-allow infra + tracked hostnames + + // existing connections). The BPF maps exist independently of the link, so + // attaching earlier would run the program live in default-deny with an empty + // allowlist, dropping legitimate startup traffic until the rules land (the + // attach-before-program race). Programming first makes the firewall correct + // from its first enforced packet; dynamic DNS-resolved hosts are still handled + // in-band by the late-add path in processEvent. + egressLink, err := tc.AttachEgress(ifname, objs.TcEgress, logger) + if err != nil { + return fmt.Errorf("failed to attach TC egress: %w", err) + } + defer egressLink.Close() + // Log appropriate config source switch { case apiPolicyLoaded: @@ -461,12 +462,6 @@ func StartCargoWall(cmd *StartCmd, hooks *StartHooks) error { }() } - // Record the readiness boundary BEFORE publishing the ready signal, so the - // happens-before is: boundary recorded → sentinel/hook fires → CI proceeds → - // user traffic. A user-triggered block is therefore stamped after the - // boundary and correctly reported; it can never be misclassified as pre-ready. - readiness.MarkReady() - if hooks != nil && hooks.Ready != nil { if err := hooks.Ready(); err != nil { return fmt.Errorf("ready hook failed: %w", err) diff --git a/pkg/events/events.go b/pkg/events/events.go index ee0dccb..74f7765 100644 --- a/pkg/events/events.go +++ b/pkg/events/events.go @@ -25,7 +25,6 @@ import ( "os" "strings" "sync" - "sync/atomic" "time" "unsafe" @@ -41,62 +40,6 @@ type FirewallUpdater interface { AddIP(ip net.IP, action config.Action, ports []config.Port) (bool, error) } -// Readiness records the CLOCK_MONOTONIC instant at which CargoWall finished -// programming its allow rules and published the ready signal. Blocked events -// whose BPF timestamp (bpf_ktime_get_ns, also CLOCK_MONOTONIC) predates that -// instant are startup-race drops — the TC program attaches default-deny before -// the auto-allow infra rules land, so a connection in that window is dropped -// once and self-heals on TCP retransmit. They are not policy decisions, so they -// are excluded from the audit log (hence the SaaS push + summary) and from block -// notifications. The zero value is "not ready yet": every event before MarkReady -// is treated as pre-ready. -type Readiness struct { - readyMono atomic.Uint64 // CLOCK_MONOTONIC ns at ready; 0 = not yet ready -} - -// NewReadiness returns a Readiness in the not-yet-ready state. -func NewReadiness() *Readiness { return &Readiness{} } - -// MarkReady records the current CLOCK_MONOTONIC instant as the readiness -// boundary. Call exactly once, immediately before publishing the ready signal, -// so it happens-before any externally triggered (e.g. user) traffic. -func (r *Readiness) MarkReady() { - var ts unix.Timespec - if err := unix.ClockGettime(unix.CLOCK_MONOTONIC, &ts); err != nil { - // Fail open for reporting: never leave readyMono at 0 (which would - // suppress every block forever). A 1ns boundary suppresses nothing - // real — every genuine bpf_ktime_get_ns stamp is far larger. - r.readyMono.Store(1) - return - } - r.readyMono.Store(uint64(ts.Nano())) -} - -// isPreReady reports whether a blocked event with the given CLOCK_MONOTONIC BPF -// timestamp occurred before the readiness boundary (and so should be excluded -// from reporting). Precedence: -// - a nil Readiness never suppresses (reporting is unaffected when the gate -// isn't wired up); -// - before MarkReady (readyMono == 0) the firewall is still being programmed, -// so the whole window is pre-ready and every event is suppressed — including -// one with a zero/garbage stamp; -// - after MarkReady, a zero event timestamp never suppresses, so a -// missing/garbage field can't mask a real (post-ready) block; otherwise the -// event is pre-ready iff its stamp precedes the boundary. -func (r *Readiness) isPreReady(eventMono uint64) bool { - if r == nil { - return false - } - readyMono := r.readyMono.Load() - if readyMono == 0 { // MarkReady not run yet → everything so far is pre-ready - return true - } - if eventMono == 0 { // defensive: never suppress on a zero/garbage stamp - return false - } - return eventMono < readyMono -} - // lookupProcessName reads the process name from /proc//comm. // Returns empty string if the process no longer exists or can't be read. func lookupProcessName(pid uint32) string { @@ -301,7 +244,7 @@ func dstPortAllowedByRule(dstPort uint16, proto uint8, ports []config.Port) bool // processEvent handles a single blocked/allowed event from the ring buffer. func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *NotificationTracker, - auditLogger *AuditLogger, fw FirewallUpdater, readiness *Readiness, logger *slog.Logger, + auditLogger *AuditLogger, fw FirewallUpdater, logger *slog.Logger, ) { if len(raw) < int(unsafe.Sizeof(BpfBlockedEvent{})) { return @@ -509,24 +452,6 @@ func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *No } else if event.SrcPort == 0 && event.DstPort < 256 { // Non-TCP/UDP protocol block — dst_port contains the protocol number protocolName := getProtocolName(uint8(event.DstPort)) - - if readiness.isPreReady(event.Timestamp) { - // Startup-race drop (see Readiness): keep it in the process log for - // forensics, but exclude it from the audit log (→ SaaS push + - // summary) and the notification — it is not a policy decision. - logConnEvent(logger, "Protocol blocked", cnameChain, - "src", srcIP, - "dst", displayHostname, - "dst_ip", dstIP, - "protocol", protocolName, - "protocol_num", event.DstPort, - "process", comm, - "pid", pid, - "pre_ready", true, - "reported", false) - return - } - logConnEvent(logger, "Protocol blocked", cnameChain, "src", srcIP, "dst", displayHostname, @@ -570,25 +495,6 @@ func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *No } } else { // Blocked TCP SYN or UDP connection - if readiness.isPreReady(event.Timestamp) { - // Startup-race drop (see Readiness): the TC program attached - // default-deny before the auto-allow infra rules were programmed, so - // this was dropped once and self-heals on retransmit. Keep it in the - // process log for forensics, but exclude it from the audit log (→ - // SaaS push + summary) and the notification — it is not a policy - // decision. - logConnEvent(logger, "Connection blocked", cnameChain, - "src", fmt.Sprintf("%s:%d", srcIP, event.SrcPort), - "dst", displayHostname, - "dst_ip", dstIP, - "dst_port", event.DstPort, - "process", comm, - "pid", pid, - "pre_ready", true, - "reported", false) - return - } - logConnEvent(logger, "Connection blocked", cnameChain, "src", fmt.Sprintf("%s:%d", srcIP, event.SrcPort), "dst", displayHostname, @@ -631,7 +537,7 @@ func logConnEvent(logger *slog.Logger, msg string, cnameChain []string, attrs .. } // ProcessBlockedEvents processes blocked connection events -func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notificationTracker *NotificationTracker, auditLogger *AuditLogger, fw FirewallUpdater, readiness *Readiness, logger *slog.Logger) { +func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notificationTracker *NotificationTracker, auditLogger *AuditLogger, fw FirewallUpdater, logger *slog.Logger) { for { record, err := rd.Read() if err != nil { @@ -642,6 +548,6 @@ func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notific continue } - processEvent(record.RawSample, configMgr, notificationTracker, auditLogger, fw, readiness, logger) + processEvent(record.RawSample, configMgr, notificationTracker, auditLogger, fw, logger) } } diff --git a/pkg/events/events_test.go b/pkg/events/events_test.go index 8c2fdef..83e21d6 100644 --- a/pkg/events/events_test.go +++ b/pkg/events/events_test.go @@ -510,7 +510,7 @@ func TestProcessEvent_IPv4BlockedTCP(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, nil, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -546,7 +546,7 @@ func TestProcessEvent_IPv4AllowedTCP(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -579,7 +579,7 @@ func TestProcessEvent_ProtocolBlocked(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, nil, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -587,256 +587,6 @@ func TestProcessEvent_ProtocolBlocked(t *testing.T) { assert.Equal(t, "ICMP", events[0].Protocol) } -// --- Readiness gate: pre-ready startup-race drops (issue: suppress) --- - -// blockTestDeps bundles the fixtures the readiness-gate tests share: a -// temp-file audit logger, a state-machine mock + tracker, and a default-deny -// config manager. -type blockTestDeps struct { - auditLogger *AuditLogger - smClient *mockStateMachineClient - tracker *NotificationTracker - cm *config.Manager - auditPath string -} - -func newBlockTestDeps(t *testing.T) blockTestDeps { - t.Helper() - auditPath := filepath.Join(t.TempDir(), "audit.jsonl") - auditLogger, err := NewAuditLogger(auditPath, false) - require.NoError(t, err) - t.Cleanup(func() { auditLogger.Close() }) - - smClient := &mockStateMachineClient{} - cm := config.NewConfigManager() - require.NoError(t, cm.LoadConfigFromRules(nil, config.ActionDeny)) - - return blockTestDeps{ - auditLogger: auditLogger, - smClient: smClient, - tracker: NewNotificationTracker(smClient, newTestLogger()), - cm: cm, - auditPath: auditPath, - } -} - -// markReverseDNSAttempted resets the reverse-DNS cache (avoiding cross-test -// pollution) and pre-seeds the given IPs as already-attempted, so processEvent -// skips the real PTR lookup — keeping these tests hermetic and fast. -func markReverseDNSAttempted(ips ...string) { - reverseDNSMu.Lock() - reverseDNSCache = make(map[string]time.Time) - for _, ip := range ips { - reverseDNSCache[ip] = time.Now() - } - reverseDNSMu.Unlock() -} - -// makeBlockedTCPv4Event builds a blocked IPv4 TCP SYN to 93.184.216.34:443 with -// the given BPF (CLOCK_MONOTONIC) timestamp. -func makeBlockedTCPv4Event(ts uint64) []byte { - return makeBpfEvent(BpfBlockedEvent{ - IpVersion: 4, - Allowed: 0, - IpProto: unix.IPPROTO_TCP, - SrcIp: ipv4ToUint32("10.0.0.1"), - DstIp: ipv4ToUint32("93.184.216.34"), - SrcPort: 54321, - DstPort: 443, - Timestamp: ts, - }) -} - -func TestProcessEvent_PreReadyBlockedTCPSuppressed(t *testing.T) { - d := newBlockTestDeps(t) - r := NewReadiness() - r.readyMono.Store(1_000_000) // 1ms boundary - - markReverseDNSAttempted("93.184.216.34") - processEvent(makeBlockedTCPv4Event(500_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) - - assert.Empty(t, readAuditEvents(t, d.auditPath), "pre-ready block must not be written to the audit log") - assert.Empty(t, d.smClient.calls, "pre-ready block must not notify") -} - -func TestProcessEvent_PostReadyBlockedTCPReported(t *testing.T) { - d := newBlockTestDeps(t) - r := NewReadiness() - r.readyMono.Store(1_000_000) - - markReverseDNSAttempted("93.184.216.34") - processEvent(makeBlockedTCPv4Event(2_000_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) - - events := readAuditEvents(t, d.auditPath) - require.Len(t, events, 1) - assert.Equal(t, EventConnectionBlocked, events[0].EventType) - assert.Len(t, d.smClient.calls, 1, "post-ready block must notify") -} - -func TestProcessEvent_NotYetReadyBlockedSuppressed(t *testing.T) { - d := newBlockTestDeps(t) - r := NewReadiness() // readyMono == 0: not ready yet - - markReverseDNSAttempted("93.184.216.34") - processEvent(makeBlockedTCPv4Event(2_000_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) - - assert.Empty(t, readAuditEvents(t, d.auditPath), "blocks before MarkReady must be suppressed") - assert.Empty(t, d.smClient.calls) -} - -func TestProcessEvent_ZeroEventTimestampNotSuppressed(t *testing.T) { - d := newBlockTestDeps(t) - r := NewReadiness() - r.readyMono.Store(1_000_000) - - markReverseDNSAttempted("93.184.216.34") - // Defensive: a zero/garbage BPF stamp must never mask a real block. - processEvent(makeBlockedTCPv4Event(0), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) - - events := readAuditEvents(t, d.auditPath) - require.Len(t, events, 1) - assert.Equal(t, EventConnectionBlocked, events[0].EventType) -} - -func TestProcessEvent_NilReadinessNeverSuppresses(t *testing.T) { - d := newBlockTestDeps(t) - - markReverseDNSAttempted("93.184.216.34") - processEvent(makeBlockedTCPv4Event(500_000), d.cm, d.tracker, d.auditLogger, nil, nil, newTestLogger()) - - events := readAuditEvents(t, d.auditPath) - require.Len(t, events, 1) - assert.Equal(t, EventConnectionBlocked, events[0].EventType) - assert.Len(t, d.smClient.calls, 1) -} - -func makeBlockedICMPv4Event(ts uint64) []byte { - return makeBpfEvent(BpfBlockedEvent{ - IpVersion: 4, - Allowed: 0, - IpProto: unix.IPPROTO_ICMP, - SrcIp: ipv4ToUint32("10.0.0.1"), - DstIp: ipv4ToUint32("10.0.0.2"), - SrcPort: 0, - DstPort: 1, // ICMP - Timestamp: ts, - }) -} - -func TestProcessEvent_PreReadyProtocolBlockedSuppressed(t *testing.T) { - d := newBlockTestDeps(t) - r := NewReadiness() - r.readyMono.Store(1_000_000) - - markReverseDNSAttempted("10.0.0.2") - processEvent(makeBlockedICMPv4Event(500_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) - - assert.Empty(t, readAuditEvents(t, d.auditPath), "pre-ready protocol block must not be written to the audit log") - assert.Empty(t, d.smClient.calls) -} - -func TestProcessEvent_PostReadyProtocolBlockedReported(t *testing.T) { - d := newBlockTestDeps(t) - r := NewReadiness() - r.readyMono.Store(1_000_000) - - markReverseDNSAttempted("10.0.0.2") - processEvent(makeBlockedICMPv4Event(2_000_000), d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) - - events := readAuditEvents(t, d.auditPath) - require.Len(t, events, 1) - assert.Equal(t, EventProtocolBlocked, events[0].EventType) -} - -func TestProcessEvent_PreReadyDoesNotSuppressAllowed(t *testing.T) { - d := newBlockTestDeps(t) - r := NewReadiness() - r.readyMono.Store(1_000_000) // boundary above the event's timestamp - - raw := makeBpfEvent(BpfBlockedEvent{ - IpVersion: 4, - Allowed: 1, // allow outcome — must never be gated - IpProto: unix.IPPROTO_TCP, - SrcIp: ipv4ToUint32("10.0.0.1"), - DstIp: ipv4ToUint32("93.184.216.34"), - SrcPort: 54321, - DstPort: 443, - Timestamp: 500_000, - }) - - markReverseDNSAttempted("93.184.216.34") - processEvent(raw, d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) - - events := readAuditEvents(t, d.auditPath) - require.Len(t, events, 1) - assert.Equal(t, EventConnectionAllowed, events[0].EventType, "the readiness gate must only apply to blocked branches") -} - -func TestProcessEvent_PreReadyBlockedIPv6Suppressed(t *testing.T) { - d := newBlockTestDeps(t) - r := NewReadiness() - r.readyMono.Store(1_000_000) - - var dstIp6 [16]byte - copy(dstIp6[:], net.ParseIP("2001:db8::dead").To16()) - raw := makeBpfEvent(BpfBlockedEvent{ - IpVersion: 6, - Allowed: 0, - IpProto: unix.IPPROTO_TCP, - DstIp6: dstIp6, - SrcPort: 54321, - DstPort: 443, - Timestamp: 500_000, - }) - - markReverseDNSAttempted("2001:db8::dead") - processEvent(raw, d.cm, d.tracker, d.auditLogger, nil, r, newTestLogger()) - - assert.Empty(t, readAuditEvents(t, d.auditPath), "pre-ready IPv6 block must be suppressed (v6 parity)") - assert.Empty(t, d.smClient.calls) -} - -func TestReadiness_isPreReady(t *testing.T) { - tests := []struct { - name string - readyMono uint64 - eventMono uint64 - want bool - }{ - {"not ready yet suppresses everything", 0, 123, true}, - {"not ready yet suppresses even a zero stamp", 0, 0, true}, - {"event before boundary", 100, 50, true}, - {"event at boundary is reported (exclusive)", 100, 100, false}, - {"event after boundary", 100, 150, false}, - {"zero event timestamp never suppresses", 100, 0, false}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - r := NewReadiness() - r.readyMono.Store(tt.readyMono) - assert.Equal(t, tt.want, r.isPreReady(tt.eventMono)) - }) - } - - t.Run("nil receiver never suppresses", func(t *testing.T) { - var r *Readiness - assert.False(t, r.isPreReady(50)) - }) -} - -func TestReadiness_MarkReadySetsMonotonic(t *testing.T) { - r := NewReadiness() - require.Equal(t, uint64(0), r.readyMono.Load()) - - r.MarkReady() - stored := r.readyMono.Load() - assert.NotZero(t, stored, "MarkReady must record a non-zero boundary") - - var ts unix.Timespec - require.NoError(t, unix.ClockGettime(unix.CLOCK_MONOTONIC, &ts)) - assert.GreaterOrEqual(t, uint64(ts.Nano()), stored, "a later monotonic read must be >= the boundary") -} - func TestProcessEvent_LateResolvedIPAddedToFirewall(t *testing.T) { cm := config.NewConfigManager() require.NoError(t, cm.LoadConfigFromRules([]config.Rule{ @@ -862,7 +612,7 @@ func TestProcessEvent_LateResolvedIPAddedToFirewall(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, nil, fw, nil, newTestLogger()) + processEvent(raw, cm, nil, nil, fw, newTestLogger()) require.Len(t, fw.addedIPs, 1) assert.Equal(t, config.ActionAllow, fw.addedIPs[0].action) @@ -898,7 +648,7 @@ func TestProcessEvent_LateResolvedIPInheritsRulePorts(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, nil, fw, nil, newTestLogger()) + processEvent(raw, cm, nil, nil, fw, newTestLogger()) require.Len(t, fw.addedIPs, 1) assert.Equal(t, config.ActionAllow, fw.addedIPs[0].action) @@ -937,7 +687,7 @@ func TestProcessEvent_LateResolvedIPv6InheritsRulePorts(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, nil, fw, nil, newTestLogger()) + processEvent(raw, cm, nil, nil, fw, newTestLogger()) require.Len(t, fw.addedIPs, 1) assert.Equal(t, config.ActionAllow, fw.addedIPs[0].action) @@ -980,7 +730,7 @@ func TestProcessEvent_LateResolvedEmitsLateAllowedAudit(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1036,7 +786,7 @@ func TestProcessEvent_LateResolvedMatchedRuleIsRulePatternNotHostname(t *testing reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, fw, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, fw, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1081,7 +831,7 @@ func TestProcessEvent_BlockedNoMatchStillLogsBlocked(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) require.Empty(t, fw.addedIPs, "late-add must not fire when no allow rule matches") @@ -1128,7 +878,7 @@ func TestProcessEvent_CNAMEDerivedAttributesToOrigin(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1180,7 +930,7 @@ func TestProcessEvent_CNAMEDerivedBlockedAttributesToOrigin(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, fw, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, fw, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1232,7 +982,7 @@ func TestProcessEvent_LateResolvedDstPortNotInRulePortsStaysBlocked(t *testing.T reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) // AddIP still fires (so future port-443 retries succeed), but this // connection's audit/notification must reflect the genuine block. @@ -1281,7 +1031,7 @@ func TestProcessEvent_LateResolvedUDPEventNotAllowedByTCPRule(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1327,7 +1077,7 @@ func TestProcessEvent_LateResolvedTCPEventAllowedByTCPRule(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1371,7 +1121,7 @@ func TestProcessEvent_LateResolvedTCPEventAllowedByProtocolAllRule(t *testing.T) reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, tracker, auditLogger, fw, nil, newTestLogger()) + processEvent(raw, cm, tracker, auditLogger, fw, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1409,7 +1159,7 @@ func TestProcessEvent_IPv6Event(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1422,7 +1172,7 @@ func TestProcessEvent_TooShortBuffer(t *testing.T) { require.NoError(t, cm.LoadConfigFromRules(nil, config.ActionDeny)) // Should not panic — processEvent should return early for short buffers - processEvent([]byte{0x04}, cm, nil, nil, nil, nil, newTestLogger()) + processEvent([]byte{0x04}, cm, nil, nil, nil, newTestLogger()) } func TestProcessEvent_AutoAllowedType(t *testing.T) { @@ -1450,7 +1200,7 @@ func TestProcessEvent_AutoAllowedType(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) @@ -1486,7 +1236,7 @@ func TestProcessEvent_NoAutoAllowedTypeForUserRule(t *testing.T) { reverseDNSCache = make(map[string]time.Time) reverseDNSMu.Unlock() - processEvent(raw, cm, nil, auditLogger, nil, nil, newTestLogger()) + processEvent(raw, cm, nil, auditLogger, nil, newTestLogger()) events := readAuditEvents(t, auditPath) require.Len(t, events, 1) From dbf733daa900399803cf897485c3d693252bbff8 Mon Sep 17 00:00:00 2001 From: Matthew DeVenny Date: Mon, 29 Jun 2026 10:57:38 -0700 Subject: [PATCH 4/7] #75 make captureStderr test helper leak-free and panic-safe Drain the pipe's read end in a goroutine (so r is always consumed and closed), and restore os.Stderr + close the write end via defer even if fn panics. The concurrent drain also removes the >64KB pipe-buffer deadlock risk. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Matthew DeVenny --- cmd/github_actions_handler_test.go | 27 ++++++++++++++++++++------- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/cmd/github_actions_handler_test.go b/cmd/github_actions_handler_test.go index effd8b9..c00cfa5 100644 --- a/cmd/github_actions_handler_test.go +++ b/cmd/github_actions_handler_test.go @@ -31,20 +31,33 @@ import ( // captureStderr redirects os.Stderr around fn and returns what was written. The // GitHubActionsHandler writes to os.Stderr directly, so the test swaps the global -// for the duration of the call. +// for the duration of the call. A goroutine drains the read end concurrently, so +// a write larger than the OS pipe buffer can't deadlock and the read end is +// always consumed and closed; os.Stderr is restored and the write end closed even +// if fn panics. func captureStderr(t *testing.T, fn func()) string { t.Helper() old := os.Stderr r, w, err := os.Pipe() require.NoError(t, err) os.Stderr = w - defer func() { os.Stderr = old }() - fn() - require.NoError(t, w.Close()) - data, err := io.ReadAll(r) - require.NoError(t, err) - return string(data) + done := make(chan string, 1) + go func() { + data, _ := io.ReadAll(r) + _ = r.Close() + done <- string(data) + }() + + func() { + defer func() { + os.Stderr = old + _ = w.Close() + }() + fn() + }() + + return <-done } // TestGitHubActionsHandler_TimestampPrefix verifies every emitted line carries a From 3753731de5856bad0664af729d8118cdeb9336f0 Mon Sep 17 00:00:00 2001 From: Matthew DeVenny Date: Mon, 29 Jun 2026 11:01:55 -0700 Subject: [PATCH 5/7] #75 scope GH log timestamps to plain lines; UTC; multi-line safe Address the review nits on the process-log timestamp: - Render in UTC so it aligns with GitHub's own gutter timestamps and is unambiguous across runner timezones. - Only plain (Info) lines get the leading timestamp; ::error::/::warning::/ ::debug:: keep a bare workflow command with no injected timestamp, so GitHub still de-duplicates identical annotations instead of treating each as unique. - Timestamp every physical line of a multi-line plain record, and escape embedded newlines (%0A, with % and CR) in workflow-command messages so a multi-line message stays a single annotation rather than leaking un-prefixed continuation lines. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Matthew DeVenny --- cmd/github_actions_handler.go | 37 +++++++++++++++++--- cmd/github_actions_handler_test.go | 56 +++++++++++++++++++++++++++--- 2 files changed, 84 insertions(+), 9 deletions(-) diff --git a/cmd/github_actions_handler.go b/cmd/github_actions_handler.go index 7741db0..4319a7e 100644 --- a/cmd/github_actions_handler.go +++ b/cmd/github_actions_handler.go @@ -76,14 +76,43 @@ func (h *GitHubActionsHandler) Handle(_ context.Context, r slog.Record) error { sb.WriteString(fmt.Sprintf(" %s=%v", a.Key, a.Value)) return true }) + body := sb.String() - // Leading timestamp on every line. Placed after the prefix so GitHub still - // parses the ::error::/::warning:: workflow command at column 0. Millisecond - // precision because the startup race this surfaces is sub-second. - fmt.Fprintf(os.Stderr, "%s%s %s\n", prefix, r.Time.Format("2006-01-02 15:04:05.000"), sb.String()) + // UTC so the timestamp aligns with GitHub's own per-line log timestamps and + // is unambiguous across runner timezones. Millisecond precision because the + // startup race this surfaces is sub-second. + ts := r.Time.UTC().Format("2006-01-02 15:04:05.000") + + if prefix == "" { + // Plain log line. Timestamp every physical line so a multi-line attribute + // value (e.g. a wrapped, multi-line error) stays timestamped instead of + // only its first line. + for line := range strings.SplitSeq(body, "\n") { + fmt.Fprintf(os.Stderr, "%s %s\n", ts, line) + } + return nil + } + + // Workflow command (::error::/::warning::/::debug::). Emit a single command and + // do NOT inject the timestamp: a unique value per line would defeat GitHub's + // de-duplication of identical annotations and inflate the per-run annotation + // count. GitHub already timestamps the raw log line. Escape so an embedded + // newline stays one annotation rather than leaking un-prefixed continuation + // lines. + fmt.Fprintf(os.Stderr, "%s%s\n", prefix, escapeWorkflowMessage(body)) return nil } +// escapeWorkflowMessage encodes a string for use as the message of a GitHub +// Actions workflow command, per GitHub's data-escaping rules (% first, then CR +// and LF), so multi-line or percent-bearing messages parse as a single command. +func escapeWorkflowMessage(s string) string { + s = strings.ReplaceAll(s, "%", "%25") + s = strings.ReplaceAll(s, "\r", "%0D") + s = strings.ReplaceAll(s, "\n", "%0A") + return s +} + func (h *GitHubActionsHandler) WithAttrs(attrs []slog.Attr) slog.Handler { newAttrs := make([]slog.Attr, len(h.attrs)+len(attrs)) copy(newAttrs, h.attrs) diff --git a/cmd/github_actions_handler_test.go b/cmd/github_actions_handler_test.go index c00cfa5..89d24f2 100644 --- a/cmd/github_actions_handler_test.go +++ b/cmd/github_actions_handler_test.go @@ -60,9 +60,10 @@ func captureStderr(t *testing.T, fn func()) string { return <-done } -// TestGitHubActionsHandler_TimestampPrefix verifies every emitted line carries a -// leading millisecond timestamp, and that for annotated levels the ::error:: / -// ::warning:: workflow command stays at column 0 (timestamp after the prefix). +// TestGitHubActionsHandler_TimestampPrefix verifies plain (Info) lines get a +// leading UTC millisecond timestamp, while workflow-command levels (::error::, +// ::warning::) keep the command at column 0 with NO injected timestamp (so +// GitHub still de-duplicates identical annotations). func TestGitHubActionsHandler_TimestampPrefix(t *testing.T) { h := NewGitHubActionsHandler(false) ts := time.Date(2026, 6, 29, 15, 24, 8, 456_000_000, time.UTC) @@ -83,6 +84,51 @@ func TestGitHubActionsHandler_TimestampPrefix(t *testing.T) { assert.Regexp(t, `^2026-06-29 15:24:08\.456 Connection blocked\b`, lines[0]) assert.Contains(t, lines[0], "dst_port=443") - // Error line: prefix stays at column 0, timestamp follows it. - assert.Regexp(t, `^::error::2026-06-29 15:24:08\.456 boom\b`, lines[1]) + // Error line: bare workflow command, no timestamp injected (preserves dedup). + assert.Equal(t, "::error::boom", lines[1]) +} + +// TestGitHubActionsHandler_TimestampUTC verifies the timestamp is rendered in +// UTC regardless of the record's location, so it aligns with GitHub's UTC gutter. +func TestGitHubActionsHandler_TimestampUTC(t *testing.T) { + h := NewGitHubActionsHandler(false) + // A fixed instant expressed in a non-UTC zone; the handler must print its UTC + // wall-clock (17:00 in -05:00 == 22:00 UTC), not the local 17:00. + loc := time.FixedZone("X", -5*60*60) + ts := time.Date(2026, 6, 29, 17, 0, 0, 250_000_000, loc) + + out := captureStderr(t, func() { + rec := slog.NewRecord(ts, slog.LevelInfo, "hello", 0) + require.NoError(t, h.Handle(context.Background(), rec)) + }) + + assert.Equal(t, "2026-06-29 22:00:00.250 hello", strings.TrimRight(out, "\n")) +} + +// TestGitHubActionsHandler_MultiLine verifies a multi-line attribute value +// timestamps every physical line on a plain record, and stays a single escaped +// annotation on a workflow-command record. +func TestGitHubActionsHandler_MultiLine(t *testing.T) { + h := NewGitHubActionsHandler(false) + ts := time.Date(2026, 6, 29, 15, 24, 8, 456_000_000, time.UTC) + + out := captureStderr(t, func() { + info := slog.NewRecord(ts, slog.LevelInfo, "validation failed", 0) + info.AddAttrs(slog.String("error", "line1\nline2")) + require.NoError(t, h.Handle(context.Background(), info)) + + errRec := slog.NewRecord(ts, slog.LevelError, "boom", 0) + errRec.AddAttrs(slog.String("detail", "a\nb")) + require.NoError(t, h.Handle(context.Background(), errRec)) + }) + + lines := strings.Split(strings.TrimRight(out, "\n"), "\n") + require.Len(t, lines, 3) + + // Each physical line of the plain record is independently timestamped. + assert.Equal(t, "2026-06-29 15:24:08.456 validation failed error=line1", lines[0]) + assert.Equal(t, "2026-06-29 15:24:08.456 line2", lines[1]) + + // The workflow command stays one line with newlines escaped as %0A. + assert.Equal(t, "::error::boom detail=a%0Ab", lines[2]) } From cb005a837576e15306eadef3c4c65e46ed8f9254 Mon Sep 17 00:00:00 2001 From: Matthew DeVenny Date: Mon, 29 Jun 2026 11:15:57 -0700 Subject: [PATCH 6/7] #75 write each GH log record in a single stderr write The multi-line plain-line path emitted one Fprintf per physical line, so a concurrent log from another goroutine could interleave between a record's lines and scramble multi-line output. Buffer the whole formatted record into a strings.Builder and write it once, restoring the original one-write-per-record behavior. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Matthew DeVenny --- cmd/github_actions_handler.go | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/cmd/github_actions_handler.go b/cmd/github_actions_handler.go index 4319a7e..b2e7b4f 100644 --- a/cmd/github_actions_handler.go +++ b/cmd/github_actions_handler.go @@ -83,23 +83,27 @@ func (h *GitHubActionsHandler) Handle(_ context.Context, r slog.Record) error { // startup race this surfaces is sub-second. ts := r.Time.UTC().Format("2006-01-02 15:04:05.000") + var out strings.Builder if prefix == "" { // Plain log line. Timestamp every physical line so a multi-line attribute // value (e.g. a wrapped, multi-line error) stays timestamped instead of // only its first line. for line := range strings.SplitSeq(body, "\n") { - fmt.Fprintf(os.Stderr, "%s %s\n", ts, line) + fmt.Fprintf(&out, "%s %s\n", ts, line) } - return nil + } else { + // Workflow command (::error::/::warning::/::debug::). Emit a single command + // and do NOT inject the timestamp: a unique value per line would defeat + // GitHub's de-duplication of identical annotations and inflate the per-run + // annotation count. GitHub already timestamps the raw log line. Escape so an + // embedded newline stays one annotation rather than leaking un-prefixed + // continuation lines. + fmt.Fprintf(&out, "%s%s\n", prefix, escapeWorkflowMessage(body)) } - // Workflow command (::error::/::warning::/::debug::). Emit a single command and - // do NOT inject the timestamp: a unique value per line would defeat GitHub's - // de-duplication of identical annotations and inflate the per-run annotation - // count. GitHub already timestamps the raw log line. Escape so an embedded - // newline stays one annotation rather than leaking un-prefixed continuation - // lines. - fmt.Fprintf(os.Stderr, "%s%s\n", prefix, escapeWorkflowMessage(body)) + // Write the whole record in one call so a concurrent log from another + // goroutine can't interleave between a multi-line record's lines. + _, _ = os.Stderr.WriteString(out.String()) return nil } From b88171a279d2e2cfea05d35c0f486fc65d6321f1 Mon Sep 17 00:00:00 2001 From: Matthew DeVenny Date: Mon, 29 Jun 2026 11:46:54 -0700 Subject: [PATCH 7/7] #75 trim trailing newline before per-line GH log timestamping A plain Info record whose assembled body ended in "\n" produced a stray timestamp-only line, because strings.SplitSeq yields a trailing empty element. Trim a single trailing newline before splitting, and add a regression test for an attribute value ending in "\n". Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Matthew DeVenny --- cmd/github_actions_handler.go | 5 +++-- cmd/github_actions_handler_test.go | 17 +++++++++++++++++ 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/cmd/github_actions_handler.go b/cmd/github_actions_handler.go index b2e7b4f..bd10994 100644 --- a/cmd/github_actions_handler.go +++ b/cmd/github_actions_handler.go @@ -87,8 +87,9 @@ func (h *GitHubActionsHandler) Handle(_ context.Context, r slog.Record) error { if prefix == "" { // Plain log line. Timestamp every physical line so a multi-line attribute // value (e.g. a wrapped, multi-line error) stays timestamped instead of - // only its first line. - for line := range strings.SplitSeq(body, "\n") { + // only its first line. Trim a single trailing newline first so a value + // ending in "\n" doesn't produce a stray timestamp-only line. + for line := range strings.SplitSeq(strings.TrimSuffix(body, "\n"), "\n") { fmt.Fprintf(&out, "%s %s\n", ts, line) } } else { diff --git a/cmd/github_actions_handler_test.go b/cmd/github_actions_handler_test.go index 89d24f2..a0054ec 100644 --- a/cmd/github_actions_handler_test.go +++ b/cmd/github_actions_handler_test.go @@ -132,3 +132,20 @@ func TestGitHubActionsHandler_MultiLine(t *testing.T) { // The workflow command stays one line with newlines escaped as %0A. assert.Equal(t, "::error::boom detail=a%0Ab", lines[2]) } + +// TestGitHubActionsHandler_TrailingNewline verifies a plain record whose body +// ends in a newline does not emit a stray timestamp-only line. +func TestGitHubActionsHandler_TrailingNewline(t *testing.T) { + h := NewGitHubActionsHandler(false) + ts := time.Date(2026, 6, 29, 15, 24, 8, 456_000_000, time.UTC) + + out := captureStderr(t, func() { + rec := slog.NewRecord(ts, slog.LevelInfo, "ran", 0) + rec.AddAttrs(slog.String("output", "done\n")) + require.NoError(t, h.Handle(context.Background(), rec)) + }) + + lines := strings.Split(strings.TrimRight(out, "\n"), "\n") + require.Len(t, lines, 1) + assert.Equal(t, "2026-06-29 15:24:08.456 ran output=done", lines[0]) +}