From 07b0f6c2e073413f8e9ff65f686a6cbd6976b79b Mon Sep 17 00:00:00 2001 From: Sebastian Norling <1932208+Bazze@users.noreply.github.com> Date: Wed, 27 May 2026 18:07:57 +0200 Subject: [PATCH] fix(event/stream): address round-3 review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Critical - Two-step lock race in renewLoop: getPullPoint() + pullPointGen() were separate Lock/Unlock pairs, leaving a window in which a concurrent setPullPoint could advance gen between the two reads. The renew then ran against ref-N but believed its captured gen was N+1, and updateGrantedTerminationIfGen "succeeded" writing the old subscription's grant onto the new one — same class of bug the generation counter was meant to fix. New snapshotPullPoint accessor reads both under one lock. - wssePasswordRE was missing the close-tag-or-EOF alternative that wsseSecurityRE got last round; a element truncated at the 64 KiB error cap escaped redaction. Pattern is now symmetric. Important - Dropped unused (granted, now) params from nextRenewIntervalAfterError; only opts.RetryBackoff is consulted. Tests adjusted. - Moved gen field next to pullPointMu with explicit guard comment. - Inlined the single-use rootTag helper; nil case handled directly. - Loosened TestNextRenewIntervalAfterError_FallsBackToRetryBackoff so future jitter doesn't break it. Docs - Extended trust-boundary godoc on SendSoap* to name the full set of weaponisable WS-* headers (wsa:To/ReplyTo/FaultTo/MessageID, wsu:Timestamp). - Mirrored the trust-boundary warning on gosoap.AddStringHeaderContents so library consumers see it at the package entry point too. - Noted that updateGrantedTerminationIfGen intentionally doesn't bump gen (would defeat rotation detection). - Documented the wsseSecurityRE truncation-branch trade-off (max-redact > max-context for log lines). Tests - TestSnapshotPullPoint_AtomicReadOfRefAndGen pins the new accessor. - TestEnrichSOAPErr_RedactsTruncatedPasswordOutsideSecurity exercises the now-symmetric password redaction. - TestEnrichSOAPErr_RedactsMultipleSecurityBlocks pins existing multi-block behaviour. - TestBuildRefParamsHeader_RejectsNonReferenceParametersRoot_Table covers vendor-suffix, multi-root, and unrelated-element cases. - One-line comment on the &Stream{} tests explaining nil-now safety. --- Device.go | 8 +++++--- event/stream/renew.go | 13 ++++++------- event/stream/renew_test.go | 40 ++++++++++++++++++++++++++++---------- event/stream/soap.go | 25 ++++++++++++------------ event/stream/soap_test.go | 39 +++++++++++++++++++++++++++++++++++++ event/stream/stream.go | 18 +++++++++++++---- gosoap/soap-builder.go | 4 ++++ 7 files changed, 110 insertions(+), 37 deletions(-) diff --git a/Device.go b/Device.go index 628d242..9c89b14 100644 --- a/Device.go +++ b/Device.go @@ -349,9 +349,11 @@ func (dev Device) SendSoap(endpoint string, xmlRequestBody string) (*http.Respon // element-free content errors before any request is made. // // SECURITY: do not pass content sourced from untrusted clients. The -// API assumes the caller is authoritative for the envelope; injected -// or in the header forwards verbatim and -// may override envelope defaults under our credentials. +// API assumes the caller is authoritative for the envelope. Header +// content forwards verbatim — among others, overrides +// auth, overrides intent, // +// redirect responses, enables replay- +// token forgery, and bypasses freshness checks. func (dev Device) SendSoapWithHeader(endpoint, xmlRequestBody, xmlHeaderContent string) (*http.Response, error) { return dev.SendSoapWithOptions(endpoint, xmlRequestBody, WithSOAPHeader(xmlHeaderContent)) } diff --git a/event/stream/renew.go b/event/stream/renew.go index c7c6969..3e29edf 100644 --- a/event/stream/renew.go +++ b/event/stream/renew.go @@ -19,15 +19,14 @@ import ( // reliable recovery once a subscription is GC'd. func (s *Stream) renewLoop(ctx context.Context) { for { - ref := s.getPullPoint() - gen := s.pullPointGen() + ref, gen := s.snapshotPullPoint() if !sleepCtx(ctx, nextRenewInterval(ref.GrantedTermination, s.opts, s.now())) { return } granted, err := renewPullPoint(s.caller, s.getPullPoint(), s.opts) if err != nil { s.surfaceError(ErrRenewFailed{Err: err}) - if !sleepCtx(ctx, nextRenewIntervalAfterError(ref.GrantedTermination, s.opts, s.now())) { + if !sleepCtx(ctx, nextRenewIntervalAfterError(s.opts)) { return } continue @@ -58,12 +57,12 @@ func nextRenewInterval(granted time.Time, opts Options, now time.Time) time.Dura return d } -// nextRenewIntervalAfterError ignores GrantedTermination — by the time -// renew has failed once, the grant is typically already in the past -// and nextRenewInterval would floor to 1s, hammering the camera. +// nextRenewIntervalAfterError returns the post-failure sleep. The +// grant is typically already in the past by the time renew has failed +// once, so nextRenewInterval would floor to 1s and hammer the camera. // Recovery is the pull loop's reconnect path; we just need to not // accelerate retries past the configured RetryBackoff. -func nextRenewIntervalAfterError(_ time.Time, opts Options, _ time.Time) time.Duration { +func nextRenewIntervalAfterError(opts Options) time.Duration { if opts.RetryBackoff > 0 { return opts.RetryBackoff } diff --git a/event/stream/renew_test.go b/event/stream/renew_test.go index 0cb86ab..c9a29e6 100644 --- a/event/stream/renew_test.go +++ b/event/stream/renew_test.go @@ -209,20 +209,17 @@ func TestRenewPullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) { // ticker design retried at the configured cadence regardless. After // a failure the loop must use a backoff decoupled from the stale // grant. -func TestNextRenewIntervalAfterError_IgnoresStaleGrantedTermination(t *testing.T) { - now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC) - stale := now.Add(-500 * time.Millisecond) - opts := Options{InitialTermination: 60 * time.Second, RenewMargin: 10 * time.Second, RetryBackoff: time.Second} - got := nextRenewIntervalAfterError(stale, opts, now) +func TestNextRenewIntervalAfterError_BacksOffAtLeastRetryBackoff(t *testing.T) { + opts := Options{RetryBackoff: time.Second} + got := nextRenewIntervalAfterError(opts) assert.GreaterOrEqual(t, got, opts.RetryBackoff, "failure path must back off at least RetryBackoff, not 1s floor on stale grant") } -func TestNextRenewIntervalAfterError_FallsBackToRetryBackoff(t *testing.T) { - now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC) - opts := Options{InitialTermination: 60 * time.Second, RenewMargin: 10 * time.Second, RetryBackoff: 5 * time.Second} - got := nextRenewIntervalAfterError(time.Time{}, opts, now) - assert.Equal(t, opts.RetryBackoff, got) +func TestNextRenewIntervalAfterError_DefaultsToOneSecondWhenRetryBackoffZero(t *testing.T) { + got := nextRenewIntervalAfterError(Options{}) + assert.Equal(t, time.Second, got, + "zero RetryBackoff must yield the safety floor, not a tight 0-duration sleep") } // Lost-update race: renew snapshots the ref, the SOAP call returns, @@ -231,6 +228,8 @@ func TestNextRenewIntervalAfterError_FallsBackToRetryBackoff(t *testing.T) { // granted time onto the NEW subscription, the new schedule is wrong. // Update must be conditioned on "same subscription as when I read." func TestUpdateGrantedTermination_DropsWriteWhenSubscriptionRotated(t *testing.T) { + // s.now is intentionally nil — this test does not exercise any + // time-dependent path; only the gen-counter accessors. s := &Stream{} original := subscriptionRef{Address: "http://camera/sub-A"} s.setPullPoint(original) @@ -248,6 +247,7 @@ func TestUpdateGrantedTermination_DropsWriteWhenSubscriptionRotated(t *testing.T } func TestUpdateGrantedTermination_AppliesWhenGenMatches(t *testing.T) { + // s.now is intentionally nil — gen-counter path only. s := &Stream{} s.setPullPoint(subscriptionRef{Address: "http://camera/sub"}) gen := s.pullPointGen() @@ -255,3 +255,23 @@ func TestUpdateGrantedTermination_AppliesWhenGenMatches(t *testing.T) { s.updateGrantedTerminationIfGen(gen, t1) assert.Equal(t, t1, s.getPullPoint().GrantedTermination) } + +// renewLoop must capture (ref, gen) atomically. Two separate +// getPullPoint() + pullPointGen() reads leave a window in which a +// concurrent setPullPoint advances gen between the two reads — the +// renew then runs against ref-N but believes its captured gen is N+1, +// and updateGrantedTerminationIfGen("succeeds") writing the old +// subscription's grant onto the new one. Single snapshot closes it. +func TestSnapshotPullPoint_AtomicReadOfRefAndGen(t *testing.T) { + // s.now is intentionally nil — gen-counter path only. + s := &Stream{} + s.setPullPoint(subscriptionRef{Address: "A"}) // gen=1 + ref, gen := s.snapshotPullPoint() + assert.Equal(t, "A", ref.Address) + assert.Equal(t, uint64(1), gen) + + s.setPullPoint(subscriptionRef{Address: "B"}) // gen=2 + ref, gen = s.snapshotPullPoint() + assert.Equal(t, "B", ref.Address) + assert.Equal(t, uint64(2), gen) +} diff --git a/event/stream/soap.go b/event/stream/soap.go index 1cc7555..d8a1e7f 100644 --- a/event/stream/soap.go +++ b/event/stream/soap.go @@ -197,11 +197,14 @@ var ( // echoes the request in a fault; scrub before logging. The // alternation handles the truncated case where the read cap fell // between and : in that case nothing past - // the opening tag is safe to retain. wssePasswordRE is the - // belt-and-braces fallback for non-conformant cameras emitting - // Password / UsernameToken outside a Security wrapper. + // the opening tag is safe to retain — the replacement re-emits a + // synthetic close tag, dropping the remainder of the body excerpt + // (max-redact preferred to max-context for log lines). + // wssePasswordRE is the belt-and-braces fallback for + // non-conformant cameras emitting Password / UsernameToken outside + // a Security wrapper; same truncation handling. wsseSecurityRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Security\b[^>]*>(?:.*?\s]+:)?Security>|.*)`) - wssePasswordRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Password\b[^>]*>.*?\s]+:)?Password>`) + wssePasswordRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Password\b[^>]*>(?:.*?\s]+:)?Password>|.*)`) ) // extractSOAPFault returns the reason text from a SOAP fault, falling @@ -247,8 +250,11 @@ func buildRefParamsHeader(refParamsXML string) (string, error) { return "", fmt.Errorf("parse ref params: %w", err) } wrapper := doc.Root() - if wrapper == nil || wrapper.Tag != "ReferenceParameters" { - return "", fmt.Errorf("ref params root must be <*:ReferenceParameters>, got <%s>", rootTag(wrapper)) + if wrapper == nil { + return "", errors.New("ref params has no root element") + } + if wrapper.Tag != "ReferenceParameters" { + return "", fmt.Errorf("ref params root must be <*:ReferenceParameters>, got <%s>", wrapper.Tag) } var out strings.Builder for _, child := range wrapper.ChildElements() { @@ -266,13 +272,6 @@ func buildRefParamsHeader(refParamsXML string) (string, error) { return out.String(), nil } -func rootTag(e *etree.Element) string { - if e == nil { - return "" - } - return e.Tag -} - // inheritXmlns copies xmlns / xmlns:* declarations from src onto dst // when dst doesn't already declare them, so a child whose namespace // prefix was declared on an ancestor stays valid in isolation. diff --git a/event/stream/soap_test.go b/event/stream/soap_test.go index 8289c05..153e2c1 100644 --- a/event/stream/soap_test.go +++ b/event/stream/soap_test.go @@ -680,3 +680,42 @@ func TestExtractReferenceParameters_EmptySubscriptionReferenceReturnsEmpty(t *te ` assert.Empty(t, extractReferenceParameters(body)) } + +func TestEnrichSOAPErr_RedactsTruncatedPasswordOutsideSecurity(t *testing.T) { + body := `` + + `` + + `admin` + + `hunter2` // truncated — no , no , no + got := enrichSOAPErr(fakeResponse(body), errors.New("500")) + require.Error(t, got) + assert.NotContains(t, got.Error(), "hunter2", + "Password without a closing tag (truncated at cap) must still be redacted") +} + +func TestEnrichSOAPErr_RedactsMultipleSecurityBlocks(t *testing.T) { + body := `` + + `secret1` + + `secret2` + + `` + got := enrichSOAPErr(fakeResponse(body), errors.New("500")) + require.Error(t, got) + assert.NotContains(t, got.Error(), "secret1") + assert.NotContains(t, got.Error(), "secret2", + "ReplaceAllString must catch every Security block, not just the first") +} + +func TestBuildRefParamsHeader_RejectsNonReferenceParametersRoot_Table(t *testing.T) { + cases := []struct { + name, raw string + }{ + {"vendor suffix", `X`}, + {"multi-root", ``}, + {"unrelated element", `content`}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + _, err := buildRefParamsHeader(c.raw) + require.Error(t, err) + }) + } +} diff --git a/event/stream/stream.go b/event/stream/stream.go index 49d3aeb..9dac55b 100644 --- a/event/stream/stream.go +++ b/event/stream/stream.go @@ -165,9 +165,9 @@ type Stream struct { caller caller opts Options - pullPointMu sync.Mutex - pullPoint subscriptionRef - gen uint64 // bumped on every setPullPoint so renews can detect a recreate + pullPointMu sync.Mutex // guards pullPoint and gen + pullPoint subscriptionRef + gen uint64 // bumped on every setPullPoint so renews detect mid-flight recreate events chan Event errors chan error @@ -205,10 +205,20 @@ func (s *Stream) pullPointGen() uint64 { return s.gen } +// snapshotPullPoint reads ref + gen under one lock so a concurrent +// setPullPoint can't slip in between two separate accessor calls and +// leave the caller with mismatched halves. +func (s *Stream) snapshotPullPoint() (subscriptionRef, uint64) { + s.pullPointMu.Lock() + defer s.pullPointMu.Unlock() + return s.pullPoint, s.gen +} + // updateGrantedTerminationIfGen writes the granted time only when the // caller's snapshot is still current. Use setPullPoint to replace the // full ref; this updates GrantedTermination in place after a renew -// response. +// response. Intentionally does not bump gen — that would defeat the +// rotation-detection it implements. func (s *Stream) updateGrantedTerminationIfGen(gen uint64, t time.Time) { s.pullPointMu.Lock() defer s.pullPointMu.Unlock() diff --git a/gosoap/soap-builder.go b/gosoap/soap-builder.go index 4d053f6..d86fc7f 100644 --- a/gosoap/soap-builder.go +++ b/gosoap/soap-builder.go @@ -170,6 +170,10 @@ func (msg *SoapMessage) AddStringHeaderContent(data string) error { // reference parameter to be a separate Header block, but a Go XML // document only has one root. Comments and text outside elements are // silently dropped. +// +// SECURITY: data forwards verbatim into the outbound envelope. Do not +// pass content sourced from untrusted clients — see the same caveat +// on onvif.Device.SendSoapWithOptions / SendSoapWithHeader. func (msg *SoapMessage) AddStringHeaderContents(data string) error { in := etree.NewDocument() if err := in.ReadFromString("" + data + ""); err != nil {