diff --git a/diam/internal/base/error_answer.go b/diam/internal/base/error_answer.go index ce6717d..0a3a002 100644 --- a/diam/internal/base/error_answer.go +++ b/diam/internal/base/error_answer.go @@ -10,7 +10,7 @@ import ( ) // BuildErrorAnswer constructs RFC 6733 §§6.2 and 7 answers; Verified Errata -// 4808 and 4887 govern Failed-AVP and local Origin-Realm respectively. +// 4615 and 4887 govern Failed-AVP and local Origin-Realm respectively. func BuildErrorAnswer(request *diam.Message, cfg Settings, resultCode uint32, failedAVPs []*diam.AVP, protocolError bool) (*diam.Message, error) { answer := request.Answer(0) dictionary := answer.Dictionary() @@ -54,7 +54,7 @@ func BuildErrorAnswer(request *diam.Message, cfg Settings, resultCode uint32, fa } } if len(failedAVPs) != 0 { - // RFC 6733 Section 7.5 and Verified Errata 4808 require one + // RFC 6733 §7.1.5 and Verified Erratum 4615 require one // Failed-AVP container; it may retain the Grouped hierarchy. failed := &diam.GroupedAVP{AVP: failedAVPs} if _, err := answer.NewAVP(avp.FailedAVP, avp.Mbit, 0, failed); err != nil { diff --git a/diam/internal/protocol/closing.go b/diam/internal/protocol/closing.go index c28cb9b..5fc2f97 100644 --- a/diam/internal/protocol/closing.go +++ b/diam/internal/protocol/closing.go @@ -55,7 +55,7 @@ func (c *Core) disconnect(es []Effect, in Input) []Effect { es = c.arm(es, ReceiverDisconnectTimer, in.At, in.Timeout) return c.reserve(es, diam.DisconnectPeer) } - if c.phase != Open || in.Cause > 2 && !c.cfg.AllowAnyLocalDisconnectCause { + if c.phase != Open || in.Cause > 2 { return append(es, c.notify(Notification{Kind: DisconnectNotice, Outcome: NotOpen})) } return c.beginClosing(es, in, true) diff --git a/diam/internal/protocol/core.go b/diam/internal/protocol/core.go index 39df72e..bf66feb 100644 --- a/diam/internal/protocol/core.go +++ b/diam/internal/protocol/core.go @@ -465,7 +465,7 @@ func (c *Core) reject(es []Effect, in Input, kind diam.ProtocolErrorKind, me *di func (c *Core) failedRejection(es []Effect, in Input, kind diam.ProtocolErrorKind, code uint32, err error) []Effect { initialCER := c.phase == PreCER && in.Message != nil && in.Message.Header.CommandCode == diam.CapabilitiesExchange es = c.terminate(es, Closed) - n := Notification{Protocol: kind, ResultCode: code, Reply: diam.ProtocolReplyFailed, ReplyCode: code, Closing: true, Message: metadata(in.Message), ReplyError: copyHandshakeError(err, c.cfg.PreserveReplyErrorFormatWrapper)} + n := Notification{Protocol: kind, ResultCode: code, Reply: diam.ProtocolReplyFailed, ReplyCode: code, Closing: true, Message: metadata(in.Message), ReplyError: copyHandshakeError(err, true)} if initialCER { n.Diagnostic = "reject-cer" } diff --git a/diam/internal/protocol/diagnostic_policy_test.go b/diam/internal/protocol/diagnostic_policy_test.go index 012d035..243801f 100644 --- a/diam/internal/protocol/diagnostic_policy_test.go +++ b/diam/internal/protocol/diagnostic_policy_test.go @@ -12,7 +12,7 @@ import ( "github.com/gomaja/go-diameter/diam/internal/base" ) -func policyProtocolNotice(t testing.TB, es []Effect) Notification { +func protocolNotice(t testing.TB, es []Effect) Notification { t.Helper() for _, e := range es { if e.Kind == Notify && e.Notification.Protocol != "" { @@ -24,9 +24,13 @@ func policyProtocolNotice(t testing.TB, es []Effect) Notification { } func TestReaderOwnsMessageErrorClosure(t *testing.T) { - for _, reader := range []bool{false, true} { - t.Run(fmt.Sprintf("reader=%t", reader), func(t *testing.T) { - c := opened(t, func(cfg *Config) { cfg.ReaderOwnsMessageErrorClosure = reader }) + for _, owner := range []string{"core", "reader", "external-admission"} { + t.Run(owner, func(t *testing.T) { + reader := owner != "core" + c := opened(t, func(cfg *Config) { cfg.ReaderOwnsMessageErrorClosure = owner == "reader" }) + if owner == "external-admission" { + c = admitted(t, true) + } in := receive(request(t, diam.DeviceWatchdog)) in.ErrorInput = true in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: true} @@ -34,7 +38,7 @@ func TestReaderOwnsMessageErrorClosure(t *testing.T) { if !sent.Required || sent.CloseAfter == reader { t.Fatalf("fatal reply closure=%t, reader ownership=%t", sent.CloseAfter, reader) } - n := policyProtocolNotice(t, c.Handle(complete(sent))) + n := protocolNotice(t, c.Handle(complete(sent))) if !n.Closing || (c.Phase() == Closed) == reader { t.Fatalf("fatality/closure conflated: notice=%+v phase=%v", n, c.Phase()) } @@ -42,79 +46,61 @@ func TestReaderOwnsMessageErrorClosure(t *testing.T) { } } -func TestClosingMessageErrorFatalityPolicy(t *testing.T) { - for _, retain := range []bool{false, true} { - t.Run(fmt.Sprintf("retain=%t", retain), func(t *testing.T) { - c := opened(t, func(cfg *Config) { - cfg.ReaderOwnsMessageErrorClosure = true - cfg.RetainClosingErrorFatality = retain - }) +func TestClosingMessageErrorRetainsFatality(t *testing.T) { + for _, reader := range []bool{false, true} { + t.Run(fmt.Sprintf("reader=%t", reader), func(t *testing.T) { + c := opened(t, func(cfg *Config) { cfg.ReaderOwnsMessageErrorClosure = reader }) c.Handle(event(Disconnect)) before := c.Snapshot().Timer(ClosingTimer) in := receive(request(t, diam.DisconnectPeer)) in.ErrorInput = true in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: true} - n := policyProtocolNotice(t, c.Handle(in)) - if n.Closing != retain || n.Reply != diam.ProtocolReplyNone || c.Phase() != Closing || c.Snapshot().Timer(ClosingTimer) != before { - t.Fatalf("Closing report altered transport ownership/deadline: %+v", n) + n := protocolNotice(t, c.Handle(in)) + if !n.Closing || n.Reply != diam.ProtocolReplyNone || c.Phase() != Closing || c.Snapshot().Timer(ClosingTimer) != before { + t.Fatalf("Closing report altered fatality/transport ownership/deadline: %+v", n) } }) } } -func TestHeaderDiagnosisRetainsSelectedDecoderFatality(t *testing.T) { - for _, retain := range []bool{false, true} { - t.Run(fmt.Sprintf("retain=%t", retain), func(t *testing.T) { - c := opened(t, func(cfg *Config) { - cfg.ReaderOwnsMessageErrorClosure = true - cfg.PreserveDecoderFatalityOnHeaderError = retain - }) +func TestHeaderDiagnosisRetainsDecoderFatality(t *testing.T) { + for _, reader := range []bool{false, true} { + t.Run(fmt.Sprintf("reader=%t", reader), func(t *testing.T) { + c := opened(t, func(cfg *Config) { cfg.ReaderOwnsMessageErrorClosure = reader }) in := receive(request(t, diam.DeviceWatchdog)) in.Message.Header.CommandFlags |= diam.ProxiableFlag in.ErrorInput = true in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: true} sent := find(t, c.Handle(in), Send) - n := policyProtocolNotice(t, c.Handle(complete(sent))) + n := protocolNotice(t, c.Handle(complete(sent))) cause, ok := n.CauseError.(*diam.MessageError) - if !ok || cause.ResultCode != diam.InvalidHDRBits || cause.Fatal != retain || !n.Closing || sent.CloseAfter || c.Phase() != Open { - t.Fatalf("replacement header cause or original report changed: %+v cause=%#v", n, cause) + if !ok || cause.ResultCode != diam.InvalidHDRBits || !cause.Fatal || !n.Closing || sent.CloseAfter == reader || (c.Phase() == Closed) == reader { + t.Fatalf("replacement header cause or original report changed: %+v cause=%#v phase=%v", n, cause, c.Phase()) } }) } } -func TestHeaderErrorDiagnosticClassificationPolicy(t *testing.T) { - for _, separate := range []bool{false, true} { - for _, closing := range []bool{false, true} { - t.Run(fmt.Sprintf("separate=%t/closing=%t", separate, closing), func(t *testing.T) { - c := opened(t, func(cfg *Config) { - cfg.ReaderOwnsMessageErrorClosure = true - cfg.SeparateHeaderErrorDiagnostics = separate - }) - if closing { - c.Handle(event(Disconnect)) - } - in := receive(request(t, diam.DisconnectPeer)) - in.Message.Header.CommandFlags |= diam.ProxiableFlag - in.ErrorInput = true - in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength} - es := c.Handle(in) - if !closing { - es = c.Handle(complete(find(t, es, Send))) - } - n := policyProtocolNotice(t, es) - kind, diagnostic := diam.ProtocolInvalidHeader, "external-message-error" - if separate { - diagnostic = "header-error" - if closing { - kind = diam.ProtocolMalformedMessage - } - } - if n.Protocol != kind || n.Diagnostic != diagnostic || n.ResultCode != diam.InvalidHDRBits { - t.Fatalf("header diagnostic=%+v, want %s/%s", n, kind, diagnostic) - } - }) - } +func TestHeaderErrorRetainsInvalidHeaderClassification(t *testing.T) { + for _, closing := range []bool{false, true} { + t.Run(fmt.Sprintf("closing=%t", closing), func(t *testing.T) { + c := opened(t, func(cfg *Config) { cfg.ReaderOwnsMessageErrorClosure = true }) + if closing { + c.Handle(event(Disconnect)) + } + in := receive(request(t, diam.DisconnectPeer)) + in.Message.Header.CommandFlags |= diam.ProxiableFlag + in.ErrorInput = true + in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength} + es := c.Handle(in) + if !closing { + es = c.Handle(complete(find(t, es, Send))) + } + n := protocolNotice(t, es) + if n.Protocol != diam.ProtocolInvalidHeader || n.Diagnostic != "external-message-error" || n.ResultCode != diam.InvalidHDRBits { + t.Fatalf("header diagnostic=%+v", n) + } + }) } } @@ -123,7 +109,6 @@ func TestCERMessageErrorClosurePolicy(t *testing.T) { t.Run(fmt.Sprintf("only-invalid-header-bits=%t", onlyBits), func(t *testing.T) { c := opened(t, func(cfg *Config) { cfg.ReaderOwnsMessageErrorClosure = true - cfg.RejectFatalRepeatedCER = true cfg.CloseCEROnInvalidHeaderBitsOnly = onlyBits }) in := receive(cer(t)) @@ -142,10 +127,10 @@ func TestCERMessageErrorClosurePolicy(t *testing.T) { } } -func TestCompletedCEAValidationPolicy(t *testing.T) { - for _, validate := range []bool{false, true} { - t.Run(fmt.Sprintf("validate=%t", validate), func(t *testing.T) { - c := machine(t, func(cfg *Config) { cfg.Role, cfg.ValidateCompletedCEA = Initiator, validate }) +func TestCompletedCEAUsesCorrelationDiagnostic(t *testing.T) { + for _, duplicate := range []bool{false, true} { + t.Run(fmt.Sprintf("duplicate=%t", duplicate), func(t *testing.T) { + c := machine(t, func(cfg *Config) { cfg.Role = Initiator }) sent := grant(t, c, c.Handle(event(Start))) c.Handle(complete(sent)) answer, err := base.BuildCEA(sent.Message, remote(t).Local, diam.Success) @@ -157,80 +142,55 @@ func TestCompletedCEAValidationPolicy(t *testing.T) { release.Token = ready.Token c.Handle(release) answer.AVP = nil - n := policyProtocolNotice(t, c.Handle(receive(answer))) - kind, code := diam.ProtocolUnexpectedMessage, uint32(0) - if validate { - kind, code = diam.ProtocolCEARejected, diam.MissingAVP + diagnostic := "duplicate-cea" + if !duplicate { + answer.Header.HopByHopID++ + diagnostic = "cea-correlation" } - if n.Protocol != kind || n.ResultCode != code || n.Closing || c.Phase() != Open { + n := protocolNotice(t, c.Handle(receive(answer))) + if n.Protocol != diam.ProtocolUnexpectedMessage || n.ResultCode != 0 || n.Diagnostic != diagnostic || n.Closing || c.Phase() != Open { t.Fatalf("completed CEA diagnostic=%+v phase=%v", n, c.Phase()) } }) } } -func TestDWAOwnershipPrecedesSemanticDiagnosticsPolicy(t *testing.T) { - for _, ownership := range []bool{false, true} { - t.Run(fmt.Sprintf("ownership-first=%t", ownership), func(t *testing.T) { - c := opened(t, func(cfg *Config) { - cfg.Watchdog, cfg.DWAOwnershipBeforeSemantics = true, ownership - cfg.WatchdogInterval = 6 * time.Second - }) - sent := probe(t, c) - answer := dwa(t, sent, diam.UnableToComply) - answer.Header.HopByHopID++ - n := policyProtocolNotice(t, c.Handle(receive(answer))) - kind, code := diam.ProtocolRejectedDWA, uint32(diam.UnableToComply) - if ownership { - kind, code = diam.ProtocolUnexpectedMessage, 0 - } - if n.Protocol != kind || n.ResultCode != code || !c.Snapshot().Watchdog().Pending { - t.Fatalf("unowned DWA changed diagnosis/ownership: %+v", n) - } - }) +func TestDWAOwnershipPrecedesSemanticDiagnostics(t *testing.T) { + c := opened(t, func(cfg *Config) { cfg.Watchdog = true; cfg.WatchdogInterval = 6 * time.Second }) + sent := probe(t, c) + answer := dwa(t, sent, diam.UnableToComply) + answer.Header.HopByHopID++ + n := protocolNotice(t, c.Handle(receive(answer))) + if n.Protocol != diam.ProtocolUnexpectedMessage || n.ResultCode != 0 || n.Diagnostic != "dwa-correlation" || !c.Snapshot().Watchdog().Pending { + t.Fatalf("unowned DWA changed diagnosis/ownership: %+v", n) } } -func TestDWAIdentityPrecedesUnsuccessfulResultPolicy(t *testing.T) { - for _, identity := range []bool{false, true} { - t.Run(fmt.Sprintf("identity-first=%t", identity), func(t *testing.T) { - c := opened(t, func(cfg *Config) { - cfg.Watchdog, cfg.DWAIdentityBeforeResult = true, identity - cfg.WatchdogInterval = 6 * time.Second - }) - sent := probe(t, c) - answer := dwa(t, sent, diam.UnableToComply) - for _, field := range answer.AVP { - if field.Code == avp.OriginHost { - field.Data = datatype.DiameterIdentity("other.example") - } - } - n := policyProtocolNotice(t, c.Handle(receive(answer))) - kind, code := diam.ProtocolRejectedDWA, uint32(diam.UnableToComply) - if identity { - kind, code = diam.ProtocolInvalidDWA, 0 - } - if n.Protocol != kind || n.ResultCode != code || !c.Snapshot().Watchdog().Pending { - t.Fatalf("unbound DWA changed diagnosis/ownership: %+v", n) - } - }) +func TestDWAIdentityPrecedesUnsuccessfulResult(t *testing.T) { + c := opened(t, func(cfg *Config) { cfg.Watchdog = true; cfg.WatchdogInterval = 6 * time.Second }) + sent := probe(t, c) + answer := dwa(t, sent, diam.UnableToComply) + for _, field := range answer.AVP { + if field.Code == avp.OriginHost { + field.Data = datatype.DiameterIdentity("other.example") + } + } + n := protocolNotice(t, c.Handle(receive(answer))) + if n.Protocol != diam.ProtocolInvalidDWA || n.ResultCode != 0 || !c.Snapshot().Watchdog().Pending { + t.Fatalf("unbound DWA changed diagnosis/ownership: %+v", n) } } -func TestReplyBuilderErrorWrapperPolicy(t *testing.T) { - for _, preserve := range []bool{false, true} { - t.Run(fmt.Sprintf("preserve=%t", preserve), func(t *testing.T) { - c := opened(t, func(cfg *Config) { cfg.PreserveReplyErrorFormatWrapper = preserve }) - empty, err := dict.NewParser() - if err != nil { - t.Fatal(err) - } - in := receive(diam.NewMessage(diam.DeviceWatchdog, diam.RequestFlag, 0, 1, 2, empty)) - es := c.reject(nil, in, diam.ProtocolInvalidRequest, &diam.MessageError{ResultCode: diam.MissingAVP}) - n := policyProtocolNotice(t, es) - if got := fmt.Sprintf("%T", n.ReplyError); (got == "*fmt.wrapError") != preserve { - t.Fatalf("builder wrapper=%s, preserve=%t", got, preserve) - } - }) +func TestReplyBuilderErrorRetainsFormatWrapper(t *testing.T) { + c := opened(t) + empty, err := dict.NewParser() + if err != nil { + t.Fatal(err) + } + in := receive(diam.NewMessage(diam.DeviceWatchdog, diam.RequestFlag, 0, 1, 2, empty)) + es := c.reject(nil, in, diam.ProtocolInvalidRequest, &diam.MessageError{ResultCode: diam.MissingAVP}) + n := protocolNotice(t, es) + if got := fmt.Sprintf("%T", n.ReplyError); got != "*fmt.wrapError" { + t.Fatalf("builder wrapper=%s, want *fmt.wrapError", got) } } diff --git a/diam/internal/protocol/fatal_callback_activity_test.go b/diam/internal/protocol/fatal_callback_activity_test.go index 5e87de7..da07017 100644 --- a/diam/internal/protocol/fatal_callback_activity_test.go +++ b/diam/internal/protocol/fatal_callback_activity_test.go @@ -8,37 +8,54 @@ import ( "github.com/gomaja/go-diameter/diam" ) -func TestFatalBodyCallbackPreservesWatchdogDeadline(t *testing.T) { - for _, fatal := range []bool{false, true} { - for _, header := range []bool{false, true} { - t.Run(fmt.Sprintf("fatal=%t/invalid-header=%t", fatal, header), func(t *testing.T) { - c := opened(t, func(cfg *Config) { - cfg.ReaderOwnsMessageErrorClosure = true - cfg.FatalBodyErrorKeepsWatchdogDeadline = true - cfg.Watchdog, cfg.QueueWrites = true, true - cfg.WatchdogInterval = 6 * time.Second - }) - before := c.Snapshot().Timer(WatchdogTimer) - in := receive(request(t, diam.DeviceWatchdog)) - in.At = epoch.Add(time.Second) - in.ErrorInput = true - in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: fatal} - if header { - in.Message.Header.CommandFlags |= diam.ProxiableFlag - } - sent := find(t, c.Handle(in), Send) - if sent.CloseAfter || !sent.Required || c.Phase() != Open { - t.Fatal("callback transferred transport closure to the core") - } - after := c.Snapshot().Timer(WatchdogTimer) - if fatal && !header { - if after != before { - t.Fatalf("fatal body callback extended deadline: before=%+v after=%+v", before, after) +func TestBodyCallbackResetsWatchdogDeadlineWithoutHealthCredit(t *testing.T) { + for _, pending := range []bool{false, true} { + for _, fatal := range []bool{false, true} { + for _, header := range []bool{false, true} { + t.Run(fmt.Sprintf("pending=%t/fatal=%t/invalid-header=%t", pending, fatal, header), func(t *testing.T) { + c := opened(t, func(cfg *Config) { + cfg.ReaderOwnsMessageErrorClosure = true + cfg.Watchdog, cfg.QueueWrites = true, true + cfg.WatchdogInterval = 6 * time.Second + }) + at := epoch.Add(time.Second) + if pending { + probe(t, c) + tick(t, c) + at = epoch.Add(13 * time.Second) + } + before := c.Snapshot().Timer(WatchdogTimer) + health := c.Snapshot().Watchdog() + controls := c.Snapshot().Controls() + in := receive(request(t, diam.DeviceWatchdog)) + in.At = at + in.ErrorInput = true + in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: fatal} + if header { + in.Message.Header.CommandFlags |= diam.ProxiableFlag + } + sent := find(t, c.Handle(in), Send) + if sent.CloseAfter || !sent.Required || c.Phase() != Open { + t.Fatal("callback transferred transport closure to the core") + } + after := c.Snapshot().Timer(WatchdogTimer) + if after.Token == before.Token || !after.Deadline.Equal(in.At.Add(6*time.Second)) { + t.Fatalf("eligible decoded activity did not reset Tw: %+v", after) } - } else if after.Token == before.Token || !after.Deadline.Equal(in.At.Add(6*time.Second)) { - t.Fatalf("eligible decoded activity did not reset Tw: %+v", after) - } - }) + if c.Snapshot().Watchdog() != health { + t.Fatalf("decoder activity changed watchdog health: %+v", c.Snapshot().Watchdog()) + } + retained := c.Snapshot().Controls() + if len(retained) != len(controls) { + t.Fatal("decoder activity changed pending control ownership") + } + for i := range controls { + if retained[i] != controls[i] { + t.Fatal("decoder activity changed pending control") + } + } + }) + } } } } diff --git a/diam/internal/protocol/handshake.go b/diam/internal/protocol/handshake.go index 36f7ec6..4ca69af 100644 --- a/diam/internal/protocol/handshake.go +++ b/diam/internal/protocol/handshake.go @@ -71,9 +71,19 @@ func (c *Core) handshakeReady() Effect { } func (c *Core) cea(es []Effect, in Input) []Effect { r := c.controls[diam.CapabilitiesExchange] - fatal := base.FatalHeaderError(in.Error) || c.cfg.FatalCEAErrorBeforeCorrelation && in.Error != nil && in.Error.Fatal + // RFC 6733 §3: an uncorrelated body report cannot reject a pending CER. + // Framing failure still terminates independently of exchange ownership. + fatal := base.FatalHeaderError(in.Error) if !fatal && !c.matches(r, in.Message) { - return c.discard(es, in, diam.ProtocolUnexpectedMessage, 0) + // Discarding the answer does not make a fatal decoder report recoverable. + // The selected closure owner still ends the connection. + fatalInput := in.Error != nil && in.Error.Fatal + if fatalInput && !c.cfg.ReaderOwnsMessageErrorClosure { + return c.fail(es, in, diam.ProtocolUnexpectedMessage, 0) + } + es = c.discard(es, in, diam.ProtocolUnexpectedMessage, 0) + es[len(es)-1].Notification.Closing = fatalInput + return es } if in.Error != nil { c.handshake = newHandshakeResult(in.Error, nil) diff --git a/diam/internal/protocol/header_fatal_policy_test.go b/diam/internal/protocol/header_fatal_policy_test.go index 7dd9f82..459c98b 100644 --- a/diam/internal/protocol/header_fatal_policy_test.go +++ b/diam/internal/protocol/header_fatal_policy_test.go @@ -10,10 +10,9 @@ import ( ) // RFC 6733 §§3 and 7.1.3 distinguish framing loss from an invalid base -// command header. Header precedence retains each frontend's established -// observer and transport contract: Manager preserves the original fatal flag; -// StateMachine's replacement header diagnosis is recoverable. -func TestHeaderValidationFatalDispositionPolicy(t *testing.T) { +// command header. Replacing a diagnosis preserves the decoder's fatality; +// reader ownership independently determines transport closure. +func TestHeaderValidationPreservesFatalityAndClosureOwnership(t *testing.T) { descriptions := []struct { name string code uint32 @@ -28,17 +27,15 @@ func TestHeaderValidationFatalDispositionPolicy(t *testing.T) { {"fatal-version", diam.UnsupportedVersion, true, true, false}, {"fatal-message-length", diam.InvalidMessageLength, true, true, false}, } - for _, manager := range []bool{false, true} { + for _, reader := range []bool{false, true} { for _, command := range []uint32{diam.CapabilitiesExchange, diam.DeviceWatchdog, diam.DisconnectPeer} { for _, req := range []bool{false, true} { for _, description := range descriptions { - name := fmt.Sprintf("manager=%t/command=%d/request=%t/%s", manager, command, req, description.name) + name := fmt.Sprintf("reader=%t/command=%d/request=%t/%s", reader, command, req, description.name) t.Run(name, func(t *testing.T) { c := opened(t, func(cfg *Config) { - cfg.ReaderOwnsMessageErrorClosure = manager - cfg.PreserveDecoderFatalityOnHeaderError = manager - cfg.SeparateHeaderErrorDiagnostics = manager - cfg.CloseCEROnInvalidHeaderBitsOnly = manager + cfg.ReaderOwnsMessageErrorClosure = reader + cfg.CloseCEROnInvalidHeaderBitsOnly = reader }) msg := request(t, command) if !req { @@ -71,7 +68,7 @@ func TestHeaderValidationFatalDispositionPolicy(t *testing.T) { if req && sends != 1 { t.Fatalf("request selected %d answers, want one", sends) } - wantClosing := description.framing || manager && description.fatal || req && command == diam.CapabilitiesExchange + wantClosing := description.fatal || req && command == diam.CapabilitiesExchange wantCode := uint32(diam.InvalidHDRBits) wantKind := diam.ProtocolInvalidHeader if description.framing { @@ -90,7 +87,7 @@ func TestHeaderValidationFatalDispositionPolicy(t *testing.T) { t.Fatalf("header/error precedence notification=%+v, want %s/%d closing=%t", notice, wantKind, wantCode, wantClosing) } wantClosed := wantClosing - if manager { + if reader { wantClosed = req && command == diam.CapabilitiesExchange && !description.framing } if got := c.Snapshot().Phase(); (got == Closed) != wantClosed { diff --git a/diam/internal/protocol/local_disconnect_cause_test.go b/diam/internal/protocol/local_disconnect_cause_test.go index 5a7228f..42d0775 100644 --- a/diam/internal/protocol/local_disconnect_cause_test.go +++ b/diam/internal/protocol/local_disconnect_cause_test.go @@ -9,39 +9,40 @@ import ( "github.com/gomaja/go-diameter/diam/datatype" ) -func TestLocalDisconnectAllowsOpaqueCause(t *testing.T) { - for _, enabled := range []bool{false, true} { - for _, cause := range []uint32{0, 1, 2, 3, ^uint32(0)} { - t.Run(fmt.Sprintf("allow-any=%t/cause=%d", enabled, cause), func(t *testing.T) { - c := opened(t, func(cfg *Config) { cfg.AllowAnyLocalDisconnectCause = enabled }) - in := event(Disconnect) - in.Cause = cause - if !enabled && cause > 2 { - wantKinds(t, c.Handle(in), Notify) - if c.Snapshot().Phase() != Open { - t.Fatal("default rejected cause changed connection phase") - } - return +func TestLocalDisconnectValidatesCause(t *testing.T) { + for _, cause := range []uint32{0, 1, 2, 3, ^uint32(0)} { + t.Run(fmt.Sprintf("cause=%d", cause), func(t *testing.T) { + c := opened(t) + in := event(Disconnect) + in.Cause = cause + if cause > 2 { + before := c.Snapshot() + effects := c.Handle(in) + wantKinds(t, effects, Notify) + n := find(t, effects, Notify).Notification + if n.Kind != DisconnectNotice || n.Outcome != NotOpen || c.Phase() != Open || c.Snapshot().Timer(WatchdogTimer) != before.Timer(WatchdogTimer) { + t.Fatal("rejected cause changed connection phase or timer") } - reservation := find(t, fence(t, c, in, false), ReserveIDs) - allocated := event(ReserveResult) - allocated.Token = reservation.Token - allocated.Reservation = Reservation{Token: reservation.Token, Hop: 71, End: 72} - sent := find(t, c.Handle(allocated), Send) - value, err := sent.Message.FindAVP(avp.DisconnectCause, 0) - if err != nil || value.Data != datatype.Enumerated(cause) { - t.Fatalf("Disconnect-Cause=%v, %v; want %d", value, err, cause) - } - if sent.Message.Header.CommandCode != diam.DisconnectPeer || c.Snapshot().Phase() != Closing { - t.Fatal("local disconnect did not send DPR in Closing") - } - }) - } + return + } + reservation := find(t, fence(t, c, in, false), ReserveIDs) + allocated := event(ReserveResult) + allocated.Token = reservation.Token + allocated.Reservation = Reservation{Token: reservation.Token, Hop: 71, End: 72} + sent := find(t, c.Handle(allocated), Send) + value, err := sent.Message.FindAVP(avp.DisconnectCause, 0) + if err != nil || value.Data != datatype.Enumerated(cause) { + t.Fatalf("Disconnect-Cause=%v, %v; want %d", value, err, cause) + } + if sent.Message.Header.CommandCode != diam.DisconnectPeer || c.Phase() != Closing { + t.Fatal("local disconnect did not send DPR in Closing") + } + }) } } -func TestOpaqueLocalCauseDoesNotRelaxReceivedDPR(t *testing.T) { - c := opened(t, func(cfg *Config) { cfg.AllowAnyLocalDisconnectCause = true }) +func TestReceivedDPRRejectsUndefinedCause(t *testing.T) { + c := opened(t) dpr := request(t, diam.DisconnectPeer) value, err := dpr.FindAVP(avp.DisconnectCause, 0) if err != nil { diff --git a/diam/internal/protocol/message_error_callback_test.go b/diam/internal/protocol/message_error_callback_test.go index c09f6f3..a87451f 100644 --- a/diam/internal/protocol/message_error_callback_test.go +++ b/diam/internal/protocol/message_error_callback_test.go @@ -7,30 +7,40 @@ import ( "github.com/gomaja/go-diameter/diam" ) -func TestBoundHeaderlessCallbackHasNoProtocolEffect(t *testing.T) { - for _, closing := range []bool{false, true} { - for _, empty := range []bool{false, true} { - t.Run(fmt.Sprintf("closing=%t/empty-message=%t", closing, empty), func(t *testing.T) { - c := opened(t, func(cfg *Config) { cfg.IgnoreBoundHeaderlessInput = true }) - if closing { - c.Handle(event(Disconnect)) - } - phase, timer := c.Phase(), c.Snapshot().Timer(ClosingTimer) - in := event(Receive) - in.ErrorInput = true - in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: true} - if empty { - in.Message = new(diam.Message) - } - for _, e := range c.Handle(in) { - if e.Kind != Discard { - t.Fatalf("headerless bound callback selected work: %+v", e) +func TestBoundHeaderlessInputClosureOwnership(t *testing.T) { + for _, ignore := range []bool{false, true} { + for _, closing := range []bool{false, true} { + for _, empty := range []bool{false, true} { + t.Run(fmt.Sprintf("ignore=%t/closing=%t/empty-message=%t", ignore, closing, empty), func(t *testing.T) { + c := opened(t, func(cfg *Config) { cfg.IgnoreBoundHeaderlessInput = ignore }) + if closing { + c.Handle(event(Disconnect)) } - } - if c.Phase() != phase || c.Snapshot().Timer(ClosingTimer) != timer { - t.Fatal("headerless callback changed lifecycle/deadline") - } - }) + phase, timer := c.Phase(), c.Snapshot().Timer(ClosingTimer) + in := event(Receive) + in.ErrorInput = true + in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: true} + if empty { + in.Message = new(diam.Message) + } + effects := c.Handle(in) + if ignore || closing { + for _, e := range effects { + if e.Kind != Discard { + t.Fatalf("headerless bound callback selected work: %+v", e) + } + } + if c.Phase() != phase || c.Snapshot().Timer(ClosingTimer) != timer { + t.Fatal("headerless callback changed lifecycle/deadline") + } + } else { + n := protocolNotice(t, effects) + if c.Phase() != Closed || !n.Closing || n.Protocol != diam.ProtocolMalformedMessage { + t.Fatalf("ordinary headerless input did not close: %+v phase=%v", n, c.Phase()) + } + } + }) + } } } } @@ -38,15 +48,15 @@ func TestBoundHeaderlessCallbackHasNoProtocolEffect(t *testing.T) { func TestInitialNonCERCallbackDiagnosticPrecedence(t *testing.T) { for _, errorInput := range []bool{false, true} { t.Run(fmt.Sprintf("callback=%t", errorInput), func(t *testing.T) { - c := machine(t, func(cfg *Config) { cfg.InitialNonCERHeaderOnError = true }) + c := machine(t) c.Handle(event(Start)) in := receive(request(t, diam.DeviceWatchdog)) in.Message.Header.CommandFlags |= diam.ProxiableFlag in.ErrorInput = errorInput - wantCode := uint32(0) + wantCode := uint32(diam.InvalidHDRBits) if errorInput { in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength} - wantCode = diam.InvalidHDRBits + wantCode = diam.InvalidAVPLength } n := find(t, c.Handle(in), Notify).Notification if n.Protocol != diam.ProtocolHandshakeState || n.ResultCode != wantCode || !n.Closing || n.Reply != diam.ProtocolReplyNone || c.Phase() != Closed { @@ -78,28 +88,75 @@ func TestInitialNonCERCallbackDiagnosticPrecedence(t *testing.T) { } } -func TestFatalCEACallbackPrecedesCorrelation(t *testing.T) { - for _, fatal := range []bool{false, true} { - t.Run(fmt.Sprintf("fatal=%t", fatal), func(t *testing.T) { - c := machine(t, func(cfg *Config) { cfg.Role = Initiator; cfg.FatalCEAErrorBeforeCorrelation = true }) - sent := grant(t, c, c.Handle(event(Start))) - in := receive(sent.Message.Answer(diam.Success)) - in.Message.Header.HopByHopID++ - in.ErrorInput = true - in.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: fatal} - var n Notification - for _, e := range c.Handle(in) { - if e.Kind == Notify && e.Notification.Protocol != "" { - n = e.Notification - } - } - if fatal { - if n.Protocol != diam.ProtocolCEARejected || n.ResultCode != diam.InvalidAVPLength || !n.Closing || c.Phase() != Closed { - t.Fatalf("fatal handshake callback was treated as stray: %+v phase=%v", n, c.Phase()) - } - } else if n.Protocol != diam.ProtocolUnexpectedMessage || n.ResultCode != 0 || n.Closing || c.Phase() != WaitCEA { - t.Fatalf("recoverable stray changed handshake ownership: %+v phase=%v", n, c.Phase()) +func TestUncorrelatedCEAErrorDisposition(t *testing.T) { + for _, readerOwns := range []bool{false, true} { + for _, coordinate := range []string{"hop", "end"} { + for _, decoder := range []struct { + name string + code uint32 + fatal, framing bool + }{ + {"body", diam.InvalidAVPLength, false, false}, + {"fatal-body", diam.InvalidAVPLength, true, false}, + {"version", diam.UnsupportedVersion, true, true}, + {"message-length", diam.InvalidMessageLength, true, true}, + } { + t.Run(fmt.Sprintf("reader=%t/%s/%s", readerOwns, coordinate, decoder.name), func(t *testing.T) { + c := machine(t, func(cfg *Config) { + cfg.Role, cfg.ReaderOwnsMessageErrorClosure = Initiator, readerOwns + }) + sent := grant(t, c, c.Handle(event(Start))) + c.Handle(complete(sent)) + before := c.Snapshot().Timer(HandshakeTimer) + if before.Token == 0 { + t.Fatal("CER response deadline was not armed") + } + controls := c.Snapshot().Controls() + in := receive(sent.Message.Answer(diam.Success)) + if coordinate == "hop" { + in.Message.Header.HopByHopID++ + } else { + in.Message.Header.EndToEndID++ + } + in.ErrorInput = true + in.Error = &diam.MessageError{ResultCode: decoder.code, Fatal: decoder.fatal} + effects := c.Handle(in) + n := protocolNotice(t, effects) + if decoder.framing { + if n.Protocol != diam.ProtocolCEARejected || n.ResultCode != decoder.code || !n.Closing || c.Phase() != Closed || c.Snapshot().Timer(HandshakeTimer).Token != 0 { + t.Fatalf("framing failure did not terminate handshake: %+v phase=%v", n, c.Phase()) + } + return + } + if n.Protocol != diam.ProtocolUnexpectedMessage || n.ResultCode != 0 || n.Closing != decoder.fatal { + t.Fatalf("stray body error lost its discard/fatal disposition: %+v", n) + } + closed := decoder.fatal && !readerOwns + if closed { + if c.Phase() != Closed || c.Snapshot().Timer(HandshakeTimer).Token != 0 || len(c.Snapshot().Controls()) != 0 { + t.Fatal("fatal native discard retained handshake work") + } + } else { + if c.Phase() != WaitCEA || c.Snapshot().Timer(HandshakeTimer) != before { + t.Fatal("reader-owned or recoverable discard changed the CER deadline") + } + retained := c.Snapshot().Controls() + if len(controls) != 1 || len(retained) != 1 || retained[0] != controls[0] { + t.Fatal("stray body error consumed CER ownership") + } + } + closeSelected := false + for _, effect := range effects { + if effect.Kind == Send { + t.Fatal("uncorrelated CEA selected a wire answer") + } + closeSelected = closeSelected || effect.Kind == Close + } + if closeSelected != closed { + t.Fatalf("close selected=%t, want %t", closeSelected, closed) + } + }) } - }) + } } } diff --git a/diam/internal/protocol/receive.go b/diam/internal/protocol/receive.go index c270db1..792630b 100644 --- a/diam/internal/protocol/receive.go +++ b/diam/internal/protocol/receive.go @@ -64,15 +64,9 @@ func (c *Core) receive(es []Effect, in Input) (out []Effect) { var code uint32 if in.Error != nil { code = in.Error.ResultCode - } else if headerErr := base.ValidateHeader(m); headerErr != nil && !c.cfg.InitialNonCERHeaderOnError { + } else if headerErr := base.ValidateHeader(m); headerErr != nil { code = headerErr.ResultCode } - if c.cfg.InitialNonCERHeaderOnError && in.ErrorInput && c.candidateToken == 0 && !base.FatalHeaderError(in.Error) { - if headerErr := base.ValidateHeader(m); headerErr != nil { - in.Error = headerErr - code = headerErr.ResultCode - } - } return c.fail(es, in, diam.ProtocolHandshakeState, code) } start := len(es) @@ -83,9 +77,9 @@ func (c *Core) receive(es []Effect, in Input) (out []Effect) { out[start].ReleaseDispatch = true } }() - // Repeated CER disposition precedes optional body revalidation; fatal - // decoder reports retain the frontend's rejection/close obligation. - fatalCER := in.Error != nil && in.Error.Fatal && (c.cfg.RejectFatalRepeatedCER || c.passiveResponder && in.ErrorInput) + // RFC 6733 §7, §7.1.5: a detected fatal request error retains its + // required answer. Decodable repeated CERs keep their ordinary disposition. + fatalCER := in.Error != nil && in.Error.Fatal if c.phase == Open && isCER && !base.FatalHeaderError(in.Error) && !fatalCER { if c.passiveResponder && c.passiveCER && in.Error == nil && !in.ErrorInput { return c.discard(es, in, "", 0) @@ -98,11 +92,10 @@ func (c *Core) receive(es []Effect, in Input) (out []Effect) { } return c.discard(es, in, diam.ProtocolUnexpectedMessage, 0) } - // Client's completed CER exchange consumes no further decoded base CEA, - // including one with invalid semantics. Keep its correlation diagnostic - // ahead of validation; receipt activity above still receives no health credit. - // Decoder failures retain the separate message-error path below. - if !c.cfg.ValidateCompletedCEA && !in.ErrorInput && c.phase == Open && c.cfg.Role == Initiator && m.Header.CommandCode == diam.CapabilitiesExchange && m.Header.ApplicationID == 0 && m.Header.CommandFlags&diam.RequestFlag == 0 && in.Error == nil { + // RFC 6733 §3: discard duplicate or unowned completed answers before body + // semantics. Receipt activity receives no health credit; decoder failures + // retain the separate message-error path below. + if !in.ErrorInput && c.phase == Open && c.cfg.Role == Initiator && m.Header.CommandCode == diam.CapabilitiesExchange && m.Header.ApplicationID == 0 && m.Header.CommandFlags&diam.RequestFlag == 0 && in.Error == nil { es = c.discard(es, in, diam.ProtocolUnexpectedMessage, 0) es[len(es)-1].Notification.Diagnostic = "cea-correlation" if h := c.completedCEA; h != nil && h.HopByHopID == m.Header.HopByHopID && h.EndToEndID == m.Header.EndToEndID { @@ -113,9 +106,7 @@ func (c *Core) receive(es []Effect, in Input) (out []Effect) { originalFatal := in.Error != nil && in.Error.Fatal if err := base.ValidateHeader(m); err != nil && !base.FatalHeaderError(in.Error) { // Fatality is independent of which diagnosis wins header precedence. - if c.cfg.PreserveDecoderFatalityOnHeaderError { - err.Fatal = originalFatal - } + err.Fatal = originalFatal in.Error = err } if in.Error != nil { @@ -250,7 +241,12 @@ func (c *Core) messageError(es []Effect, in Input, originalFatal bool) []Effect return c.externalMessageError(es, in, kind, originalFatal) } if c.phase == Closing { - return c.discard(es, in, kind, me.ResultCode) + // Closing selects no new reply or termination for message errors. + // Preserve fatality in the report; the reader and Closing deadline + // retain responsibility for closing the connection. + es = c.discard(es, in, kind, me.ResultCode) + es[len(es)-1].Notification.Closing = originalFatal + return es } if m.Header.CommandFlags&diam.RequestFlag == 0 { if me.Fatal { @@ -273,16 +269,6 @@ func (c *Core) externalMessageError(es []Effect, in Input, kind diam.ProtocolErr if me != nil { n.ResultCode, n.CauseError = me.ResultCode, copyHandshakeError(me) } - headerDiagnosis := c.cfg.SeparateHeaderErrorDiagnostics && kind == diam.ProtocolInvalidHeader - if headerDiagnosis { - n.Diagnostic = "header-error" - if c.phase == Closing && in.Message.Header.CommandFlags&diam.RequestFlag != 0 { - n.Protocol = diam.ProtocolMalformedMessage - } - } - if c.phase == Closing && !c.cfg.RetainClosingErrorFatality { - n.Closing = false - } if c.phase == Closing || in.Message.Header == nil || in.Message.Header.CommandFlags&diam.RequestFlag == 0 { return append(es, c.effect(Discard), c.notify(n)) } @@ -309,9 +295,6 @@ func (c *Core) externalMessageError(es []Effect, in Input, kind diam.ProtocolErr } if terminal { c.terminal, n.Closing, n.Diagnostic = true, true, "external-reject-cer" - if headerDiagnosis { - n.Diagnostic = "reject-cer-header" - } } return c.send(es, answer, nil, true, terminal, false, &n) } diff --git a/diam/internal/protocol/repeated_cer_fatal_test.go b/diam/internal/protocol/repeated_cer_fatal_test.go index ae1870f..9fadf56 100644 --- a/diam/internal/protocol/repeated_cer_fatal_test.go +++ b/diam/internal/protocol/repeated_cer_fatal_test.go @@ -7,26 +7,25 @@ import ( "github.com/gomaja/go-diameter/diam" ) -// Manager retains the required fatal-error answer while leaving transport -// closure to the decoder's caller. StateMachine has its own repeat policy. +// Fatal repeated CERs require an error answer; reader ownership determines +// whether the core also closes after the answer completes. func TestRepeatedCERFatalDecoderDisposition(t *testing.T) { - for _, manager := range []bool{false, true} { + for _, reader := range []bool{false, true} { for _, errorInput := range []bool{false, true} { for _, fatal := range []bool{false, true} { - t.Run(fmt.Sprintf("manager=%t/error-input=%t/fatal=%t", manager, errorInput, fatal), func(t *testing.T) { + t.Run(fmt.Sprintf("reader=%t/error-input=%t/fatal=%t", reader, errorInput, fatal), func(t *testing.T) { c := opened(t, func(cfg *Config) { - cfg.RejectFatalRepeatedCER = manager - cfg.ReaderOwnsMessageErrorClosure = manager + cfg.ReaderOwnsMessageErrorClosure = reader cfg.QueueWrites = true }) input := receive(cer(t)) input.ErrorInput = errorInput input.Error = &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: fatal} effects := c.Handle(input) - if !manager || !fatal { + if !fatal { notice := find(t, effects, Notify).Notification if notice.Protocol != diam.ProtocolUnexpectedMessage || notice.ResultCode != 0 || notice.Closing || notice.Reply != diam.ProtocolReplyNone || c.Snapshot().Phase() != Open { - t.Fatalf("recoverable/frontend repeated CER disposition changed: %+v phase=%v", notice, c.Snapshot().Phase()) + t.Fatalf("recoverable repeated CER disposition changed: %+v phase=%v", notice, c.Snapshot().Phase()) } for _, effect := range effects { if effect.Kind == Send || effect.Kind == Close { @@ -36,7 +35,7 @@ func TestRepeatedCERFatalDecoderDisposition(t *testing.T) { return } sent := find(t, effects, Send) - if !sent.Required || sent.CloseAfter || resultCode(t, sent.Message) != diam.InvalidAVPLength || sent.Message.Header.CommandCode != diam.CapabilitiesExchange || sent.Message.Header.HopByHopID != input.Message.Header.HopByHopID || sent.Message.Header.EndToEndID != input.Message.Header.EndToEndID { + if !sent.Required || sent.CloseAfter == reader || resultCode(t, sent.Message) != diam.InvalidAVPLength || sent.Message.Header.CommandCode != diam.CapabilitiesExchange || sent.Message.Header.HopByHopID != input.Message.Header.HopByHopID || sent.Message.Header.EndToEndID != input.Message.Header.EndToEndID { t.Fatalf("fatal repeated CER required answer changed: %+v", sent) } queued := event(WriteResult) @@ -53,11 +52,11 @@ func TestRepeatedCERFatalDecoderDisposition(t *testing.T) { written := complete(sent) written.ReplyCode = diam.InvalidAVPLength effects = c.Handle(written) - if c.Snapshot().Phase() != Open { - t.Fatal("fatal callback took transport closure from its caller") + if (c.Snapshot().Phase() == Closed) == reader { + t.Fatalf("fatal repeated CER closure ownership changed: reader=%t phase=%v", reader, c.Snapshot().Phase()) } for _, effect := range effects { - if effect.Kind == Notify || effect.Kind == Close { + if effect.Kind == Notify || reader && effect.Kind == Close { t.Fatalf("successful reply duplicated its report or closed the transport: %+v", effect) } } diff --git a/diam/internal/protocol/reply_build_cause_test.go b/diam/internal/protocol/reply_build_cause_test.go index 6e1a46c..29f1555 100644 --- a/diam/internal/protocol/reply_build_cause_test.go +++ b/diam/internal/protocol/reply_build_cause_test.go @@ -33,9 +33,7 @@ func TestRejectionBuildFailureRetainsTypedBuilderCause(t *testing.T) { for _, open := range []bool{false, true} { t.Run(fmt.Sprintf("open=%t", open), func(t *testing.T) { cfg := func(cfg *Config) { - cfg.PreserveReplyErrorFormatWrapper = true cfg.ReaderOwnsMessageErrorClosure = true - cfg.RejectFatalRepeatedCER = true } c := machine(t, cfg) if open { diff --git a/diam/internal/protocol/types.go b/diam/internal/protocol/types.go index eec02b8..9a9a370 100644 --- a/diam/internal/protocol/types.go +++ b/diam/internal/protocol/types.go @@ -55,13 +55,17 @@ const ( // - Manager uses QueueWrites, defers initiator CER until actor acceptance and // holds validated responder candidates for authorization/election. It // supplies Recovery from peer history and always enables watchdogs. -// - The diagnostic options from IgnoreBoundHeaderlessInput through -// PreserveReplyErrorFormatWrapper, plus AllowAnyLocalDisconnectCause, are -// enabled by Manager alone. Core selects their protocol effects; adapters -// retain their public error types, logging and notification boundaries. +// - Message-error closure normally remains with Core for natively +// admitted StateMachine/Client connections and with the reader for Manager +// and externally admitted StateMachine connections. Manager selects the +// three ownership options below; external admission selects reader ownership +// from its public peer metadata. During Closing, Core discards error inputs +// and leaves closure to the reader and the existing Closing deadline. // -// These choices preserve distinct frontend contracts. Changing a policy's -// default or its component selection requires a separate behavior decision. +// Closure ownership stays in Core because one serialized decision marks a +// connection terminal and selects whether its required reply closes it once +// written. That write's completion is a later core decision that terminates the +// connection and disposes its timers. type Config struct { ValidateRequests, RejectUnknownMandatoryAVPs bool Connection Connection @@ -84,59 +88,27 @@ type Config struct { // A peer actor owns the election deadline after a CER validates. The // connection's pre-CER deadline ends when it hands that candidate over. CandidateDeadlineOwnedByPeer bool - // IgnoreBoundHeaderlessInput discards nil/headerless input except when a - // responder awaits its first CER. That initial input is a handshake-state - // failure retaining the supplied decoder result. + // IgnoreBoundHeaderlessInput leaves bound nil/headerless message-error + // closure with Manager's reader. An initial responder input still fails + // admission with the supplied decoder result; native StateMachine/Client + // inputs retain core-owned failure and timer disposal. IgnoreBoundHeaderlessInput bool - // InitialNonCERHeaderOnError derives an initial non-CER header diagnosis - // only for non-framing decoder reports. A decoded non-CER reports only the - // handshake-state violation, without deriving a header result code. - InitialNonCERHeaderOnError bool - // RejectFatalRepeatedCER preserves the required error reply for a repeated - // CER carrying a fatal decoder error, ahead of ordinary repeat disposition. - RejectFatalRepeatedCER bool - // ValidateCompletedCEA retains grammar/header diagnostics for decoded CEAs - // after an initiator handshake, rather than classifying duplicates first. - ValidateCompletedCEA bool - // PreserveDecoderFatalityOnHeaderError keeps the input decoder's fatal flag - // when base-header validation replaces its error diagnosis. - PreserveDecoderFatalityOnHeaderError bool - // ReaderOwnsMessageErrorClosure reports message errors on admitted inputs - // without taking transport closure from the reader. Required error replies - // still run; the CER header-rejection rule can select close-after-reply. + // ReaderOwnsMessageErrorClosure leaves admitted Manager message-error + // closure with its reader. Externally admitted StateMachine connections + // select the same ownership from metadata; native StateMachine/Client + // connections normally close in Core. During Closing, error inputs are + // discarded; the reader and Closing deadline retain closure ownership. + // Required replies and failed-reply closure remain core decisions. ReaderOwnsMessageErrorClosure bool - // RetainClosingErrorFatality keeps original fatality in message-error - // reports during Closing, independently of transport closure ownership. - RetainClosingErrorFatality bool - // SeparateHeaderErrorDiagnostics selects distinct base-header diagnostics; - // a rejected request during Closing is reported as malformed input. - SeparateHeaderErrorDiagnostics bool - // CloseCEROnInvalidHeaderBitsOnly restricts message-error CER closure to - // InvalidHDRBits; other decoder errors leave closure with their caller. + // CloseCEROnInvalidHeaderBitsOnly commits Manager's CER close-after-reply + // only for InvalidHDRBits; framing failures remain reader-owned. Native + // and externally admitted StateMachine retain their CER header-rejection + // close-after-reply boundary. The core selects terminality and CloseAfter + // together; write completion later terminates and disposes the timers. CloseCEROnInvalidHeaderBitsOnly bool - // FatalBodyErrorKeepsWatchdogDeadline excludes fatal, valid-header decoder - // callbacks from Tw activity. Recoverable and invalid-header input retain - // their ordinary decoded-activity policy. - FatalBodyErrorKeepsWatchdogDeadline bool - // FatalCEAErrorBeforeCorrelation rejects a fatal CEA decoder report before - // checking whether its identifiers match the pending CER. - FatalCEAErrorBeforeCorrelation bool - // DWAOwnershipBeforeSemantics classifies an unowned DWA after grammar - // validation but before interpreting its identity or Result-Code. - DWAOwnershipBeforeSemantics bool - // DWAIdentityBeforeResult diagnoses an unbound unsuccessful DWA by identity - // before interpreting its Result-Code as a rejection of the local probe. - DWAIdentityBeforeResult bool - // PreserveReplyErrorFormatWrapper retains the concrete standard fmt wrapper - // type around a bounded, detached error-answer builder failure. - PreserveReplyErrorFormatWrapper bool // StateMachine's synchronous Disconnect may start after a peer DPR. Its // caller deadline is independent of the peer-initiated Closing deadline. LocalDisconnectInClosing bool - // AllowAnyLocalDisconnectCause preserves Manager.Close's existing opaque - // uint32 input. RFC 6733 section 5.4.3 defines only values 0-2; this local - // API compatibility option never relaxes validation of received DPRs. - AllowAnyLocalDisconnectCause bool // SkipDWRGrammarAfterHook applies the control parser's discard/result rules // to the hook output without repeating pre-hook wire-grammar validation. SkipDWRGrammarAfterHook bool diff --git a/diam/internal/protocol/unowned_control_test.go b/diam/internal/protocol/unowned_control_test.go index b0e1225..f52f3eb 100644 --- a/diam/internal/protocol/unowned_control_test.go +++ b/diam/internal/protocol/unowned_control_test.go @@ -14,7 +14,7 @@ import ( func TestInitialUnownedControlAnswerRetainsBodyValidation(t *testing.T) { for _, command := range []uint32{diam.DeviceWatchdog, diam.DisconnectPeer} { t.Run(controlTestName(command), func(t *testing.T) { - c := opened(t, func(cfg *Config) { cfg.DWAOwnershipBeforeSemantics = true }) + c := opened(t) msg := diam.NewMessage(command, 0, 0, 91, 92, c.cfg.Dictionary) n := find(t, c.Handle(receive(msg)), Notify).Notification want := diam.ProtocolInvalidDWA @@ -41,7 +41,6 @@ func TestUnownedControlAnswerRetainsGrammarDiagnostics(t *testing.T) { for _, command := range []uint32{diam.DeviceWatchdog, diam.DisconnectPeer} { t.Run(controlTestName(command), func(t *testing.T) { c := opened(t, func(cfg *Config) { - cfg.DWAOwnershipBeforeSemantics = true cfg.Watchdog = true cfg.WatchdogInterval = 6 * time.Second }) @@ -75,7 +74,6 @@ func TestUnownedControlGrammarAfterCompletedExchange(t *testing.T) { for _, variant := range []string{"malformed-duplicate", "malformed-stray", "other-command", "new-generation"} { t.Run(variant, func(t *testing.T) { configure := func(cfg *Config) { - cfg.DWAOwnershipBeforeSemantics = true cfg.Watchdog = true cfg.WatchdogInterval = 6 * time.Second } @@ -110,7 +108,6 @@ func TestUnownedControlGrammarAfterCompletedExchange(t *testing.T) { func TestRejectedDWAIdentityPrecedesResult(t *testing.T) { c := opened(t, func(cfg *Config) { - cfg.DWAIdentityBeforeResult = true cfg.Watchdog = true cfg.WatchdogInterval = 6 * time.Second }) @@ -134,7 +131,6 @@ func TestUnownedDWAResultCannotRejectExchange(t *testing.T) { for _, state := range []string{"initial", "current", "completed"} { t.Run(state, func(t *testing.T) { c := opened(t, func(cfg *Config) { - cfg.DWAOwnershipBeforeSemantics = true cfg.Watchdog = true cfg.WatchdogInterval = 6 * time.Second }) diff --git a/diam/internal/protocol/watchdog.go b/diam/internal/protocol/watchdog.go index 4900f6b..600c170 100644 --- a/diam/internal/protocol/watchdog.go +++ b/diam/internal/protocol/watchdog.go @@ -62,11 +62,6 @@ func (c *Core) activity(es []Effect, in Input) []Effect { if c.phase != Open || !c.cfg.Watchdog || base.FatalHeaderError(in.Error) { return es } - // Fatal body callbacks can retain the existing deadline. Invalid base - // headers still take the decoded-header activity path. - if c.cfg.FatalBodyErrorKeepsWatchdogDeadline && in.ErrorInput && in.Error != nil && in.Error.Fatal && base.ValidateHeader(in.Message) == nil { - return es - } return c.watchdogArm(es, in.At, in.Interval) } @@ -122,15 +117,17 @@ func (c *Core) dwa(es []Effect, in Input) []Effect { } // Grammar diagnostics precede ownership classification as library policy. // RFC 6733 §§3, 6.2.1: only a correlated answer owns the local exchange. - if c.cfg.DWAOwnershipBeforeSemantics && !c.matches(c.controls[diam.DeviceWatchdog], m) { - return c.discard(es, in, diam.ProtocolUnexpectedMessage, 0) + if !c.matches(c.controls[diam.DeviceWatchdog], m) { + es = c.discard(es, in, diam.ProtocolUnexpectedMessage, 0) + es[len(es)-1].Notification.Diagnostic = "dwa-correlation" + return es } var parsed base.DWA if err := parsed.Parse(m); err != nil { return c.discardCause(es, in, diam.ProtocolInvalidDWA, err) } r := c.controls[diam.DeviceWatchdog] - if c.cfg.DWAIdentityBeforeResult && parsed.ResultCode != diam.Success && !c.bound(m, true) { + if parsed.ResultCode != diam.Success && !c.bound(m, true) { var peer base.DWR _ = peer.Parse(m) err := errors.New("DWA Origin-Realm mismatch") @@ -140,10 +137,7 @@ func (c *Core) dwa(es []Effect, in Input) []Effect { return c.discardCause(es, in, diam.ProtocolInvalidDWA, err) } if parsed.ResultCode != diam.Success { - - if c.matches(r, m) && c.bound(m, true) { - es = c.consumeControl(es, r, m) - } + es = c.consumeControl(es, r, m) es = c.discard(es, in, diam.ProtocolRejectedDWA, parsed.ResultCode) es[len(es)-1].Notification.CauseError = fmt.Errorf("DWA Result-Code %d", parsed.ResultCode) return es @@ -163,9 +157,6 @@ func (c *Core) dwa(es []Effect, in Input) []Effect { } return c.discardCause(es, in, diam.ProtocolInvalidDWA, err) } - if !c.matches(r, m) { - return c.discard(es, in, diam.ProtocolUnexpectedMessage, 0) - } current := c.wd // A fast answer can precede the writer's return. Apply its credit to the // tentative send transition; the late completion cannot restore Pending. diff --git a/diam/peer/answer_correlation_test.go b/diam/peer/answer_correlation_test.go index c5ac36c..2460ce4 100644 --- a/diam/peer/answer_correlation_test.go +++ b/diam/peer/answer_correlation_test.go @@ -211,7 +211,7 @@ func TestManagerDPARequiresLocalDPROnWire(t *testing.T) { unsolicited.Header.HopByHopID, unsolicited.Header.EndToEndID = 0, 0 for range 2 { write(t, c, unsolicited) - correlationEvent(t, events, diam.ProtocolInvalidDPA, false) + correlationEvent(t, events, diam.ProtocolUnexpectedMessage, false) if m.Peers()[0].State != Closing { t.Fatal("unsolicited zero-ID DPA completed disconnect") } @@ -240,7 +240,7 @@ func TestManagerDPAIdentityPrecedesResultOnWire(t *testing.T) { result, _ := bad.FindAVP(avp.ResultCode, 0) result.Data = datatype.Unsigned32(diam.UnableToComply) write(t, c, bad) - e := correlationEvent(t, events, diam.ProtocolInvalidDPA, false) + e := correlationEvent(t, events, diam.ProtocolUnexpectedMessage, false) if e.Err.ResultCode() != 0 { t.Fatalf("unbound peer result interpreted: %+v", e) } @@ -313,7 +313,7 @@ func TestManagerDPAOwnership(t *testing.T) { } a.onWire(event{s: s, msg: answer}) if fault == "no request" { - if a.state != Closing || s.c.Closed() || len(events) != 1 || events[0].Err.Kind() != diam.ProtocolInvalidDPA { + if a.state != Closing || s.c.Closed() || len(events) != 1 || events[0].Err.Kind() != diam.ProtocolUnexpectedMessage { t.Fatalf("unsolicited DPA: state=%s events=%v", a.state, events) } return diff --git a/diam/peer/dwa_diagnostic_contract_test.go b/diam/peer/dwa_diagnostic_contract_test.go index dd3a72f..8dcd362 100644 --- a/diam/peer/dwa_diagnostic_contract_test.go +++ b/diam/peer/dwa_diagnostic_contract_test.go @@ -55,7 +55,7 @@ func TestManagerDWAOwnershipPreservesDiagnosticsTCP(t *testing.T) { t.Fatal(err) } field.Data = datatype.Unsigned32(diam.UnableToComply) - result = 0 + kind, result = diam.ProtocolUnexpectedMessage, 0 if phase == "current" { kind, result = diam.ProtocolRejectedDWA, diam.UnableToComply } diff --git a/diam/peer/local_disconnect_cause_test.go b/diam/peer/local_disconnect_cause_test.go index c9b3098..305acd0 100644 --- a/diam/peer/local_disconnect_cause_test.go +++ b/diam/peer/local_disconnect_cause_test.go @@ -3,51 +3,32 @@ package peer import ( "context" "fmt" + "reflect" "testing" "time" - "github.com/gomaja/go-diameter/diam" - "github.com/gomaja/go-diameter/diam/avp" - "github.com/gomaja/go-diameter/diam/datatype" - "github.com/gomaja/go-diameter/diam/internal/base" "github.com/gomaja/go-diameter/diam/sm" ) -// Close passes its uint32 cause through to the DPR. This is API -// compatibility for local callers; RFC 6733 §5.4.3 defines only values 0–2. -func TestManagerClosePreservesLocalDisconnectCause(t *testing.T) { +func TestManagerCloseRejectsInvalidLocalDisconnectCause(t *testing.T) { for _, dialed := range []bool{false, true} { for _, cause := range []uint32{3, ^uint32(0)} { t.Run(fmt.Sprintf("dialed=%t/cause=%d", dialed, cause), func(t *testing.T) { m, c, _, _ := closingPeer(t, dialed) + before := m.Peers() ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() - done := make(chan error, 1) - go func() { done <- m.Close(ctx, sm.DisconnectCause(cause)) }() - if err := c.SetReadDeadline(time.Now().Add(300 * time.Millisecond)); err != nil { - t.Fatal(err) + err := m.Close(ctx, sm.DisconnectCause(cause)) + if fmt.Sprint(err) != fmt.Sprintf("invalid Disconnect-Cause %d", cause) || fmt.Sprintf("%T", err) != "*errors.errorString" { + t.Fatalf("Close(%d) error=%T %v", cause, err, err) } - dpr, err := diam.ReadMessage(c, m.dictionary()) - if err != nil { - t.Fatalf("Close(%d) did not send DPR: %v", cause, err) + if m.closing || !reflect.DeepEqual(m.Peers(), before) { + t.Fatal("invalid cause changed manager shutdown or peer state") } - if dpr.Header.CommandCode != diam.DisconnectPeer || dpr.Header.CommandFlags&diam.RequestFlag == 0 { - t.Fatalf("Close sent %+v", dpr.Header) - } - value, err := dpr.FindAVP(avp.DisconnectCause, 0) - if err != nil || value.Data != datatype.Enumerated(cause) { - t.Fatalf("Disconnect-Cause=%v, %v; want raw %d", value, err, cause) - } - answer, err := base.BuildDPA(dpr, testBase("known.example.net")) - if err != nil { - t.Fatal(err) - } - write(t, c, answer) - if err := <-done; err != nil { - t.Fatal(err) - } - if p := m.Peers()[0]; p.State != Closed || p.Eligible { - t.Fatalf("Close did not finish: %+v", p) + probe := watchdogRequest(t) + write(t, c, probe) + if answer := read(t, c); answer.Header.CommandCode != probe.Header.CommandCode || answer.Header.HopByHopID != probe.Header.HopByHopID { + t.Fatalf("invalid Close sent a DPR or stopped the peer: %+v", answer.Header) } }) } diff --git a/diam/peer/manager.go b/diam/peer/manager.go index ac43168..c812295 100644 --- a/diam/peer/manager.go +++ b/diam/peer/manager.go @@ -364,7 +364,13 @@ func sortSnapshots(p []PeerSnapshot) { } } +// Close disconnects every peer and waits for shutdown or ctx cancellation. +// A cause outside the values defined by RFC 6733 §5.4.3 returns an +// "invalid Disconnect-Cause" error before changing shutdown or peer state. func (m *Manager) Close(ctx context.Context, cause sm.DisconnectCause) error { + if cause > sm.DisconnectDoNotWantToTalkToYou { + return fmt.Errorf("invalid Disconnect-Cause %d", cause) + } m.mu.Lock() if m.closing { m.mu.Unlock() diff --git a/diam/peer/message_error_callback_test.go b/diam/peer/message_error_callback_test.go index 58f5168..69f8304 100644 --- a/diam/peer/message_error_callback_test.go +++ b/diam/peer/message_error_callback_test.go @@ -78,7 +78,7 @@ func TestManagerBoundHeaderlessErrorCallbackIgnoredTCP(t *testing.T) { // Closing in a public decoder notification describes the supplied fatality. // The native reader owns transport closure; a direct callback can still send -// without moving the existing watchdog deadline. +// while receipt rearms the watchdog without granting health or probe credit. func TestManagerFatalBodyCallbackPreservesLiveTransportTCP(t *testing.T) { for _, dialed := range []bool{false, true} { for _, command := range []uint32{diam.CapabilitiesExchange, diam.DeviceWatchdog, diam.DisconnectPeer} { @@ -116,8 +116,8 @@ func TestManagerFatalBodyCallbackPreservesLiveTransportTCP(t *testing.T) { t.Fatalf("callback error answer=%+v", answer.Header) } } - if after := p.state(t); after.timer != before.timer || after.token != before.token || after.health != before.health || after.pending != before.pending || after.num != before.num { - t.Fatalf("fatal callback changed watchdog state/deadline: before=%+v after=%+v", before, after) + if after := p.state(t); after.timer == before.timer || after.token == before.token || after.health != before.health || after.pending != before.pending || after.num != before.num { + t.Fatalf("fatal callback did not reset Tw independently of watchdog credit: before=%+v after=%+v", before, after) } publicErrorDWR(t, p) if p.s.c.Closed() || len(records.Records()) != start { @@ -134,16 +134,15 @@ func TestManagerFatalBodyCallbackPreservesLiveTransportTCP(t *testing.T) { } } -func TestManagerOpenInitiatorCEAPreservesValidationDiagnosticTCP(t *testing.T) { +func TestManagerCompletedCEARetainsCorrelationDiagnosticTCP(t *testing.T) { for _, header := range []bool{false, true} { t.Run(fmt.Sprintf("header=%t", header), func(t *testing.T) { records := publicErrorLog(t) p := newCoreActivityWire(t, true) message := correlationCEA(t, capabilitiesExchangeRequest(t), "known.example.net") - kind, result, text := diam.ProtocolCEARejected, uint32(diam.MissingAVP), "diameter message error 5005: Diameter validation failed (5005): required AVP missing" + kind, result, text := diam.ProtocolUnexpectedMessage, uint32(0), "peer: unmatched answer (RFC 6733 §6.2.1)" if header { message.Header.CommandFlags |= diam.ProxiableFlag - kind, result, text = diam.ProtocolInvalidHeader, diam.InvalidHDRBits, "diameter message error 3008: invalid base command flags (RFC 6733 §§3, 3.2)" } else { message.AVP = nil } @@ -154,7 +153,7 @@ func TestManagerOpenInitiatorCEAPreservesValidationDiagnosticTCP(t *testing.T) { expected.MessageLength = uint32(message.Len()) protocolErrorCheck(t, event, p.s.c, expected, kind, result, false, diam.ProtocolReplyNone, 0, 0) publicErrorDWR(t, p) - publicErrorLogCheck(t, records, start, diam.CapabilitiesExchange, "*diam.MessageError", text) + publicErrorLogCheck(t, records, start, diam.CapabilitiesExchange, "*errors.errorString", text) }) } } @@ -180,7 +179,7 @@ func TestManagerClosingDPRPreservesDecoderDiagnosticTCP(t *testing.T) { write(t, p.c, message) } else { message.Header.CommandFlags |= diam.ProxiableFlag - kind, result, text = diam.ProtocolMalformedMessage, diam.InvalidHDRBits, "diameter message error 3008: invalid base command flags (RFC 6733 §§3, 3.2)" + kind, result, text = diam.ProtocolInvalidHeader, diam.InvalidHDRBits, "diameter message error 3008: invalid base command flags (RFC 6733 §§3, 3.2)" if err := p.m.HandleMessageError(p.s.c, message, &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: fatal}); err != nil { t.Fatal(err) } @@ -223,7 +222,7 @@ func TestManagerControlAnswerErrorIdentityTCP(t *testing.T) { message = correlationCEA(t, capabilitiesExchangeRequest(t), "known.example.net") case "unowned-dwa": message, err = base.BuildDWA(watchdogRequest(t), testBase("known.example.net")) - kind, text = diam.ProtocolInvalidDWA, "peer: unsolicited or mismatched DWA" + kind, text = diam.ProtocolUnexpectedMessage, "peer: unsolicited or mismatched DWA" case "unmatched-dpa": message, err = base.BuildDPA(closingDPR(t, p.m), testBase("known.example.net")) default: @@ -235,14 +234,14 @@ func TestManagerControlAnswerErrorIdentityTCP(t *testing.T) { switch scenario { case "dpa-identifiers": message.Header.HopByHopID++ - text = "peer: DPA identifiers mismatch" + kind, text = diam.ProtocolUnexpectedMessage, "peer: DPA identifiers mismatch" case "dpa-host": field, e := message.FindAVP(avp.OriginHost, 0) if e != nil { t.Fatal(e) } field.Data = datatype.DiameterIdentity("other.example.net") - text = "peer: DPA Origin-Host mismatch" + kind, text = diam.ProtocolUnexpectedMessage, "peer: DPA Origin-Host mismatch" case "dpa-result": field, e := message.FindAVP(avp.ResultCode, 0) if e != nil { diff --git a/diam/peer/protocol_error_test.go b/diam/peer/protocol_error_test.go index feadc64..e5319aa 100644 --- a/diam/peer/protocol_error_test.go +++ b/diam/peer/protocol_error_test.go @@ -75,7 +75,7 @@ func TestProtocolErrorManagerWireDecisions(t *testing.T) { t.Fatal(err) } return m - }, diam.ProtocolInvalidDWA, 0, 0, false, false}, + }, diam.ProtocolUnexpectedMessage, 0, 0, false, false}, {"invalid base header", func(t *testing.T) *diam.Message { m := watchdogRequest(t); m.Header.ApplicationID = 4; return m }, diam.ProtocolInvalidHeader, 3008, 3008, false, false}, {"invalid answer header", func(t *testing.T) *diam.Message { m := watchdogRequest(t).Answer(2001) @@ -301,7 +301,7 @@ func TestProtocolErrorManagerDWAAndDPA(t *testing.T) { result uint32 mutate func(*diam.Message) }{ - {"DWA identifier", diam.DeviceWatchdog, diam.ProtocolInvalidDWA, 0, func(m *diam.Message) { m.Header.EndToEndID++ }}, + {"DWA identifier", diam.DeviceWatchdog, diam.ProtocolUnexpectedMessage, 0, func(m *diam.Message) { m.Header.EndToEndID++ }}, {"DWA identity", diam.DeviceWatchdog, diam.ProtocolInvalidDWA, 0, func(m *diam.Message) { a, _ := m.FindAVP(avp.OriginHost, 0) a.Data = datatype.DiameterIdentity("other.example.net") @@ -319,8 +319,8 @@ func TestProtocolErrorManagerDWAAndDPA(t *testing.T) { a, _ = m.FindAVP(avp.ResultCode, 0) a.Data = datatype.Unsigned32(3002) }}, - {"DPA identifier", diam.DisconnectPeer, diam.ProtocolInvalidDPA, 0, func(m *diam.Message) { m.Header.EndToEndID++ }}, - {"DPA identity", diam.DisconnectPeer, diam.ProtocolInvalidDPA, 0, func(m *diam.Message) { + {"DPA identifier", diam.DisconnectPeer, diam.ProtocolUnexpectedMessage, 0, func(m *diam.Message) { m.Header.EndToEndID++ }}, + {"DPA identity", diam.DisconnectPeer, diam.ProtocolUnexpectedMessage, 0, func(m *diam.Message) { a, _ := m.FindAVP(avp.OriginHost, 0) a.Data = datatype.DiameterIdentity("other.example.net") }}, @@ -708,9 +708,14 @@ func TestProtocolErrorManagerWirePendingAnswers(t *testing.T) { switch fault { case "identifiers": answer.Header.EndToEndID++ + kind = diam.ProtocolUnexpectedMessage case "hop identifiers": answer.Header.HopByHopID++ + kind = diam.ProtocolUnexpectedMessage case "identity": + if command == diam.DisconnectPeer { + kind = diam.ProtocolUnexpectedMessage + } a, _ := answer.FindAVP(avp.OriginHost, 0) a.Data = datatype.DiameterIdentity("other.example.net") case "missing AVP": diff --git a/diam/peer/session_protocol.go b/diam/peer/session_protocol.go index 3c6632d..e793aac 100644 --- a/diam/peer/session_protocol.go +++ b/diam/peer/session_protocol.go @@ -55,23 +55,11 @@ func (s *session) initProtocol() error { Dictionary: s.m.dictionary(), ExpectedHost: expected, TLS: s.c.TLS() != nil, QueueWrites: true, HandshakeTimeout: s.m.cfg.Timers.CER, ClosingTimeout: s.m.cfg.Timers.Closing, Watchdog: true, WatchdogInterval: max(s.m.cfg.Timers.TwInit, watchdog.MinimumTw), Recovery: recovery, - CandidateDeadlineOwnedByPeer: true, - IgnoreBoundHeaderlessInput: true, - InitialNonCERHeaderOnError: true, - RejectFatalRepeatedCER: true, - ValidateCompletedCEA: true, - PreserveDecoderFatalityOnHeaderError: true, - ReaderOwnsMessageErrorClosure: true, - RetainClosingErrorFatality: true, - SeparateHeaderErrorDiagnostics: true, - CloseCEROnInvalidHeaderBitsOnly: true, - FatalBodyErrorKeepsWatchdogDeadline: true, - FatalCEAErrorBeforeCorrelation: true, - DWAOwnershipBeforeSemantics: true, - DWAIdentityBeforeResult: true, - PreserveReplyErrorFormatWrapper: true, - AllowAnyLocalDisconnectCause: true, - ValidateRequests: s.m.cfg.Settings.ValidateRequests, RejectUnknownMandatoryAVPs: s.m.cfg.Settings.RejectUnknownMandatoryAVPs}) + CandidateDeadlineOwnedByPeer: true, + IgnoreBoundHeaderlessInput: true, + ReaderOwnsMessageErrorClosure: true, + CloseCEROnInvalidHeaderBitsOnly: true, + ValidateRequests: s.m.cfg.Settings.ValidateRequests, RejectUnknownMandatoryAVPs: s.m.cfg.Settings.RejectUnknownMandatoryAVPs}) if err != nil { return err } @@ -553,7 +541,7 @@ func (a *actor) syncProtocol(s *session, in protocol.Input) { } func (s *session) recordProtocol(in protocol.Input, n protocol.Notification) { - if n.Protocol == "" { + if n.Protocol == "" || n.Diagnostic == "duplicate-cea" { return } a, _ := s.binding() @@ -609,13 +597,14 @@ func (s *session) recordProtocol(in protocol.Input, n protocol.Notification) { s.m.report(s, msg, err) return } - if n.Diagnostic == "header-error" || n.Diagnostic == "reject-cer-header" { + headerError := n.Protocol == diam.ProtocolInvalidHeader && (n.Diagnostic == "external-message-error" || n.Diagnostic == "external-reject-cer") + if headerError { s.m.report(s, msg, err) - if n.Diagnostic == "header-error" { + if n.Diagnostic == "external-message-error" { return } } - if n.Diagnostic == "external-reject-cer" || n.Diagnostic == "reject-cer-header" { + if n.Diagnostic == "external-reject-cer" { s.m.report(s, msg, fmt.Errorf("peer: invalid CER rejected; closing connection: %w", err)) return } @@ -739,22 +728,10 @@ func (s *session) notifyProtocol(in protocol.Input, n protocol.Notification, d * if in.Kind == protocol.WriteResult && in.Stage == protocol.WrittenStage && in.Err != nil && n.Protocol == diam.ProtocolReplyFailure { return } - msg := in.Message - if msg == nil && n.Message.HasHeader { - msg = &diam.Message{Header: &n.Message.Header} - } kind := n.Protocol if in.ErrorInput && kind == diam.ProtocolCEARejected { kind = diam.ProtocolHandshakeState } - if kind == diam.ProtocolUnexpectedMessage && msg != nil && msg.Header.CommandCode == diam.DeviceWatchdog { - kind = diam.ProtocolInvalidDWA - } - if kind == diam.ProtocolUnexpectedMessage && msg != nil && msg.Header.CommandCode == diam.DisconnectPeer { - if s.closing.Load() { - kind = diam.ProtocolInvalidDPA - } - } if s.m.ProtocolErrorEnabled() { d.notifications = append(d.notifications, diam.ProtocolErrorEvent{Conn: s.c, Err: diam.NewProtocolError(kind, n.ResultCode), Message: n.Message, Closing: n.Closing, Reply: n.Reply, ReplyCode: n.ReplyCode}) } diff --git a/diam/peer/tcp_parity_test.go b/diam/peer/tcp_parity_test.go index e3035f9..3359bea 100644 --- a/diam/peer/tcp_parity_test.go +++ b/diam/peer/tcp_parity_test.go @@ -6,8 +6,8 @@ import ( "errors" "fmt" "io" - "log/slog" "net" + "strings" "sync" "testing" "time" @@ -16,6 +16,7 @@ import ( "github.com/gomaja/go-diameter/diam/avp" "github.com/gomaja/go-diameter/diam/datatype" "github.com/gomaja/go-diameter/diam/dict" + "github.com/gomaja/go-diameter/diam/internal/logtest" "github.com/gomaja/go-diameter/diam/peer" "github.com/gomaja/go-diameter/diam/sm" "github.com/gomaja/go-diameter/diam/sm/smpeer" @@ -34,15 +35,22 @@ var parityFrontends = []parityFrontend{ } type parityTCP struct { - frontend parityFrontend - wire net.Conn - dict *dict.Parser - events chan diam.ProtocolErrorEvent - ready chan struct{} - conn diam.Conn - done <-chan error - close func() <-chan error - send func() <-chan parityApplicationResult + frontend parityFrontend + wire net.Conn + dict *dict.Parser + events chan diam.ProtocolErrorEvent + ready chan struct{} + conn diam.Conn + done <-chan error + close func() <-chan error + send func() <-chan parityApplicationResult + handler diam.MessageErrorHandler + closeCause func(uint32) <-chan error + logs *logtest.Recorder + watchdog chan sm.WatchdogEvent + manager *peer.Manager + completedCEA diam.Header + peers chan peer.PeerEvent } type parityApplicationResult struct { @@ -66,13 +74,35 @@ func parityAwait[T any](t *testing.T, ch <-chan T) T { // policy. Only the public construction/admission paths differ. Long protocol // timers keep completion dependent on messages and callbacks, not elapsed time. func parityStart(t *testing.T, frontend parityFrontend, handshake bool, application ...bool) *parityTCP { + return parityStartConfigured(t, frontend, handshake, paritySettings{application: len(application) != 0 && application[0]}) +} + +type paritySettings struct { + application bool + probe bool + clock *parityClock + interval time.Duration + writeGate *parityReplyGate + onCEA diam.HandlerFunc +} + +func parityStartConfigured(t *testing.T, frontend parityFrontend, handshake bool, options paritySettings) *parityTCP { t.Helper() h := &parityTCP{frontend: frontend, dict: contractDict(t), events: make(chan diam.ProtocolErrorEvent, 64), ready: make(chan struct{})} + h.logs, h.watchdog, h.peers = logtest.New(), make(chan sm.WatchdogEvent, 32), make(chan peer.PeerEvent, 64) + interval := time.Hour + if options.probe { + interval = 6 * time.Second + } + if options.interval != 0 { + interval = options.interval + } settings := contractSettings(h.dict, true, true) settings.DPRCloseTimeout = time.Hour settings.HandshakeTimeout = time.Hour settings.OnProtocolError = func(event diam.ProtocolErrorEvent) { h.events <- event } - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + settings.OnCEA = options.onCEA + logger := h.logs.Logger() listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) @@ -80,29 +110,36 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat t.Cleanup(func() { _ = listener.Close() }) var ready sync.Once var server *diam.Server + native := make(chan diam.Conn, 1) if frontend.manager { - manager, err := peer.New(peer.Config{ + config := peer.Config{ Settings: settings, Logger: logger, - Timers: peer.Timers{TwInit: time.Hour, CER: time.Hour, Closing: time.Hour}, + Timers: peer.Timers{TwInit: interval, CER: time.Hour, Closing: time.Hour}, OnPeerEvent: func(event peer.PeerEvent) { + h.peers <- event if event.Peer.Eligible { ready.Do(func() { close(h.ready) }) } }, - }) + } + if options.clock != nil { + config.Clock = options.clock + } + manager, err := peer.New(config) if err != nil { t.Fatal(err) } - config := peer.PeerConfig{Host: "remote.example.net", NoAutoReconnect: true} + h.manager, h.handler = manager, manager + peerConfig := peer.PeerConfig{Host: "remote.example.net", NoAutoReconnect: true} if frontend.initiator { - config.Endpoints = []peer.Endpoint{{Network: "tcp", Address: listener.Addr().String()}} + peerConfig.Endpoints = []peer.Endpoint{{Network: "tcp", Address: listener.Addr().String()}} } else { - server = &diam.Server{Dict: h.dict, Logger: logger} + server = &diam.Server{Dict: h.dict, Logger: logger, OnNewConnection: func(c diam.Conn) { native <- c }} if err := manager.BindServer(server); err != nil { t.Fatal(err) } } - if err := manager.AddPeer(config); err != nil { + if err := manager.AddPeer(peerConfig); err != nil { t.Fatal(err) } h.send = func() <-chan parityApplicationResult { @@ -115,12 +152,12 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat }() return done } - h.close = func() <-chan error { + h.closeCause = func(cause uint32) <-chan error { done := make(chan error, 1) go func() { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - done <- manager.Close(ctx, sm.DisconnectRebooting) + done <- manager.Close(ctx, sm.DisconnectCause(cause)) }() return done } @@ -136,7 +173,8 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat } } else { if !frontend.initiator { - settings.EnableWatchdog, settings.WatchdogInterval = true, time.Hour + settings.EnableWatchdog, settings.WatchdogInterval = true, interval + settings.OnWatchdogConnEvent = func(_ diam.Conn, event sm.WatchdogEvent) { h.watchdog <- event } } admitted := make(chan diam.Conn, 1) settings.OnHandshake = func(conn diam.Conn, _ *smpeer.Metadata) { @@ -147,7 +185,8 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat if err != nil { t.Fatal(err) } - if len(application) != 0 && application[0] { + h.handler = state + if options.application { answers := make(chan *diam.Message, 8) state.HandleIdx(diam.CommandIndex{AppID: contractApp, Code: contractCommand}, diam.HandlerFunc(func(_ diam.Conn, message *diam.Message) { answers <- message @@ -178,20 +217,30 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat return done } } - h.close = func() <-chan error { + h.closeCause = func(cause uint32) <-chan error { done := make(chan error, 1) - go func() { done <- state.Disconnect(h.conn, sm.DisconnectRebooting, 10*time.Second) }() + go func() { done <- state.Disconnect(h.conn, sm.DisconnectCause(cause), 10*time.Second) }() return done } if frontend.initiator { - client := &sm.Client{Dict: h.dict, Handler: state, EnableWatchdog: true, WatchdogInterval: time.Hour, - RetransmitInterval: time.Hour, Logger: logger} + client := &sm.Client{Dict: h.dict, Handler: state, EnableWatchdog: true, WatchdogInterval: interval, + RetransmitInterval: time.Hour, Logger: logger, OnWatchdogEvent: func(event sm.WatchdogEvent) { h.watchdog <- event }} done := make(chan error, 1) h.done = done ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) go func() { - conn, err := client.DialContext(ctx, "tcp", listener.Addr().String(), nil) + var conn diam.Conn + var err error + if options.writeGate == nil { + conn, err = client.DialContext(ctx, "tcp", listener.Addr().String(), nil) + } else { + var transport net.Conn + transport, err = (&net.Dialer{}).DialContext(ctx, "tcp", listener.Addr().String()) + if err == nil { + conn, err = client.NewConn(&parityReplyConn{Conn: transport, gate: options.writeGate}, listener.Addr().String()) + } + } if conn != nil { // Cleanup registration is synchronized by receiving done below. t.Cleanup(conn.Close) @@ -199,7 +248,7 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat done <- err }() } else { - server = &diam.Server{Dict: h.dict, Handler: state, Logger: logger} + server = &diam.Server{Dict: h.dict, Handler: state, Logger: logger, OnNewConnection: func(c diam.Conn) { native <- c }} } // The callback's channel publishes the connection before ready is closed. // The test consumes it only after the wire handshake has succeeded. @@ -216,7 +265,11 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat } if server != nil { served := make(chan error, 1) - go func() { served <- server.Serve(listener) }() + servedListener := listener + if options.writeGate != nil { + servedListener = &parityReplyListener{Listener: listener, gate: options.writeGate} + } + go func() { served <- server.Serve(servedListener) }() t.Cleanup(func() { _ = server.Close() if err := parityAwait(t, served); err != nil && !errors.Is(err, diam.ErrServerClosed) { @@ -235,6 +288,10 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat if err != nil { t.Fatal(err) } + h.close = func() <-chan error { return h.closeCause(uint32(sm.DisconnectRebooting)) } + if !frontend.initiator { + h.conn = parityAwait(t, native) + } t.Cleanup(func() { _ = h.wire.Close() }) if handshake { if frontend.initiator { @@ -244,7 +301,7 @@ func parityStart(t *testing.T, frontend parityFrontend, handshake bool, applicat } cea := contractBase(h.dict, diam.CapabilitiesExchange, false) cea.Header.HopByHopID, cea.Header.EndToEndID = cer.Header.HopByHopID, cer.Header.EndToEndID - h.write(t, cea) + h.completedCEA = h.write(t, cea) } else { cer := contractBase(h.dict, diam.CapabilitiesExchange, true) h.write(t, cer) @@ -626,12 +683,6 @@ func TestProtocolTCPParityLocalClosing(t *testing.T) { answer := contractBase(h.dict, diam.DisconnectPeer, false) answer.Header.HopByHopID, answer.Header.EndToEndID = hop, end kind, code := diam.ProtocolUnexpectedMessage, uint32(0) - // The core selects the same unowned-answer disposition. - // Manager's adapter preserves its InvalidDPA presentation - // while Closing; no protocol option changes wire handling. - if frontend.manager { - kind = diam.ProtocolInvalidDPA - } switch fault { case "hop": answer.Header.HopByHopID++ @@ -761,53 +812,75 @@ func TestProtocolTCPParityClosingApplicationOwnership(t *testing.T) { } } -func TestProtocolTCPParityCompletedCEAValidateCompletedCEA(t *testing.T) { - for _, frontend := range parityFrontends { - if !frontend.initiator { - continue - } - t.Run(frontend.name, func(t *testing.T) { - h := parityStart(t, frontend, true) - cea := contractBase(h.dict, diam.CapabilitiesExchange, false) - contractDrop(cea, avp.OriginRealm) - header := h.write(t, cea) - // Config.ValidateCompletedCEA is true only for Manager. It preserves - // grammar diagnosis after handshake; Client diagnoses ownership first. - kind, code := diam.ProtocolUnexpectedMessage, uint32(0) - if frontend.manager { - kind, code = diam.ProtocolCEARejected, diam.MissingAVP +func TestProtocolTCPParityCompletedCEA(t *testing.T) { + for _, correlation := range []string{"duplicate", "hop", "end"} { + for _, shape := range []string{"valid", "missing", "flags"} { + for _, frontend := range parityFrontends { + if !frontend.initiator { + continue + } + t.Run(correlation+"/"+shape+"/"+frontend.name, func(t *testing.T) { + h := parityStart(t, frontend, true) + cea := contractBase(h.dict, diam.CapabilitiesExchange, false) + cea.Header.HopByHopID, cea.Header.EndToEndID = h.completedCEA.HopByHopID, h.completedCEA.EndToEndID + if correlation == "hop" { + cea.Header.HopByHopID++ + } + if correlation == "end" { + cea.Header.EndToEndID++ + } + if shape == "missing" { + contractDrop(cea, avp.OriginRealm) + } + if shape == "flags" { + cea.Header.CommandFlags |= diam.ProxiableFlag + } + before := len(h.logs.Records()) + header := h.write(t, cea) + h.event(t, header, frontend.replyDisposition(diam.ProtocolUnexpectedMessage, 0, false, false)) + h.barrier(t) + h.noEvents(t) + if correlation == "duplicate" && len(h.logs.Records()) != before { + t.Fatal("completed duplicate CEA was logged") + } + }) } - h.event(t, header, frontend.replyDisposition(kind, code, false, false)) - h.barrier(t) - h.noEvents(t) - }) + } } } -func TestProtocolTCPParityUnownedDWADWAOwnershipBeforeSemantics(t *testing.T) { +func TestProtocolTCPParityUnownedDWA(t *testing.T) { for _, result := range []uint32{diam.Success, diam.UnableToComply} { for _, frontend := range parityFrontends { t.Run(fmt.Sprintf("%d/%s", result, frontend.name), func(t *testing.T) { h := parityStart(t, frontend, true) dwa := contractBase(h.dict, diam.DeviceWatchdog, false) contractField(t, dwa, avp.ResultCode).Data = datatype.Unsigned32(result) + before := len(h.logs.Records()) header := h.write(t, dwa) - // Config.DWAOwnershipBeforeSemantics is true only for Manager. - // All paths discard the unowned answer without a reply or closure; - // the option selects the public diagnostic's kind and result code. - kind, code := diam.ProtocolRejectedDWA, result - if result == diam.Success { - kind, code = diam.ProtocolUnexpectedMessage, 0 - } - if frontend.manager { - // Manager presents the core's unowned DWA as InvalidDWA. - // That adapter-owned classification does not change the - // selected core disposition or its ownership-first result. - kind, code = diam.ProtocolInvalidDWA, 0 - } - h.event(t, header, frontend.replyDisposition(kind, code, false, false)) + h.event(t, header, frontend.replyDisposition(diam.ProtocolUnexpectedMessage, 0, false, false)) h.barrier(t) h.noEvents(t) + found := false + for _, record := range h.logs.Records()[before:] { + detail := record.Message + " " + fmt.Sprint(logtest.Attr(record, "error").Any()) + if strings.Contains(detail, "DWA") && strings.Contains(detail, "mismatch") { + found = true + } + } + if !found { + t.Fatal("unowned DWA lost its correlation diagnostic") + } + for { + select { + case event := <-h.watchdog: + if event == sm.WatchdogInvalidAnswer || event == sm.WatchdogAnswerReceived { + t.Fatalf("unowned DWA emitted watchdog answer callback %s", event) + } + default: + return + } + } }) } } diff --git a/diam/peer/tcp_receive_parity_test.go b/diam/peer/tcp_receive_parity_test.go new file mode 100644 index 0000000..74abdcc --- /dev/null +++ b/diam/peer/tcp_receive_parity_test.go @@ -0,0 +1,659 @@ +package peer_test + +import ( + "bytes" + "errors" + "fmt" + "log/slog" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/gomaja/go-diameter/diam" + "github.com/gomaja/go-diameter/diam/avp" + "github.com/gomaja/go-diameter/diam/datatype" + "github.com/gomaja/go-diameter/diam/dict" + "github.com/gomaja/go-diameter/diam/internal/logtest" + "github.com/gomaja/go-diameter/diam/peer" + "github.com/gomaja/go-diameter/diam/sm" +) + +// The public Manager clock supplies receipt times and records actual scheduled +// deadlines. No peer state or protocol input is manufactured by this clock. +type parityClock struct { + mu sync.Mutex + now time.Time + timers []*parityTimer + changed chan struct{} +} + +type parityTimer struct { + clock *parityClock + deadline time.Time + fire func() + active bool +} + +func (c *parityClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + if c.now.IsZero() { + c.now = time.Unix(1700000000, 0) + } + return c.now +} + +func (c *parityClock) AfterFunc(d time.Duration, f func()) peer.Timer { + now := c.Now() + c.mu.Lock() + defer c.mu.Unlock() + timer := &parityTimer{clock: c, deadline: now.Add(d), fire: f, active: true} + c.timers = append(c.timers, timer) + if c.changed == nil { + c.changed = make(chan struct{}, 1) + } + select { + case c.changed <- struct{}{}: + default: + } + return timer +} + +func (timer *parityTimer) Stop() bool { + timer.clock.mu.Lock() + defer timer.clock.mu.Unlock() + was := timer.active + timer.active = false + return was +} + +func (c *parityClock) current(t *testing.T) *parityTimer { + t.Helper() + deadline := time.NewTimer(10 * time.Second) + defer deadline.Stop() + for { + c.mu.Lock() + var next *parityTimer + for _, timer := range c.timers { + if timer.active && (next == nil || timer.deadline.Before(next.deadline)) { + next = timer + } + } + if c.changed == nil { + c.changed = make(chan struct{}, 1) + } + changed := c.changed + c.mu.Unlock() + if next != nil { + return next + } + select { + case <-changed: + case <-deadline.C: + t.Fatal("no scheduled protocol deadline") + return nil + } + } +} + +func (c *parityClock) advance(d time.Duration) { + c.Now() + c.mu.Lock() + c.now = c.now.Add(d) + c.mu.Unlock() +} + +func (c *parityClock) expire(t *testing.T) { + t.Helper() + timer := c.current(t) + c.mu.Lock() + c.now, timer.active = timer.deadline, false + c.mu.Unlock() + timer.fire() +} + +func (h *parityTCP) connection(t *testing.T) diam.Conn { + t.Helper() + if h.conn == nil { + message := contractBase(h.dict, diam.DisconnectPeer, false) + header := h.write(t, message) + h.event(t, header, h.frontend.replyDisposition(diam.ProtocolUnexpectedMessage, 0, false, false)) + } + return h.conn +} + +func (h *parityTCP) callback(t *testing.T, message *diam.Message, description *diam.MessageError) { + t.Helper() + conn := h.connection(t) + if err := h.handler.HandleMessageError(conn, message, description); err != nil { + t.Fatal(err) + } +} + +func (h *parityTCP) callbackEvent(t *testing.T, message *diam.Message, want parityDisposition) { + t.Helper() + event := parityAwait(t, h.events) + if event.Conn != h.conn || !event.Message.HasHeader || event.Message.Header != *message.Header || event.Message.Stream != message.MessageStream() { + t.Fatalf("callback metadata=%+v; want %+v stream %d", event, message.Header, message.MessageStream()) + } + got := parityDisposition{event.Err.Kind(), event.Err.ResultCode(), event.Closing, event.Reply, event.ReplyCode} + if got != want { + t.Fatalf("callback disposition=%+v; want %+v", got, want) + } +} + +func parityBodyError() *diam.MessageError { + return &diam.MessageError{ResultCode: diam.InvalidAVPLength, Fatal: true, + FailedAVP: diam.NewAVP(avp.InbandSecurityID, avp.Mbit, 0, datatype.Unknown{1, 2}), Err: errors.New("invalid AVP payload")} +} + +func TestProtocolTCPParityInitialHeaderDiagnosis(t *testing.T) { + for _, callback := range []bool{false, true} { + for _, frontend := range parityFrontends { + if frontend.initiator { + continue + } + t.Run(fmt.Sprintf("callback=%t/%s", callback, frontend.name), func(t *testing.T) { + h := parityStart(t, frontend, false) + message := contractBase(h.dict, diam.DeviceWatchdog, true) + message.Header.CommandFlags |= diam.ProxiableFlag + code := uint32(diam.InvalidHDRBits) + if callback { + code = diam.InvalidAVPLength + h.callback(t, message, parityBodyError()) + h.callbackEvent(t, message, frontend.replyDisposition(diam.ProtocolHandshakeState, code, false, true)) + } else { + header := h.write(t, message) + h.event(t, header, frontend.replyDisposition(diam.ProtocolHandshakeState, code, false, true)) + } + h.eof(t) + h.noEvents(t) + }) + } + } +} + +func TestProtocolTCPParityFatalRepeatedCER(t *testing.T) { + for _, frontend := range parityFrontends { + t.Run(frontend.name, func(t *testing.T) { + var ceaCalls atomic.Int32 + options := paritySettings{} + if !frontend.manager { + options.onCEA = func(diam.Conn, *diam.Message) { ceaCalls.Add(1) } + } + h := parityStartConfigured(t, frontend, true, options) + before := ceaCalls.Load() + message := contractBase(h.dict, diam.CapabilitiesExchange, true) + h.callback(t, message, parityBodyError()) + h.callbackEvent(t, message, frontend.replyDisposition(diam.ProtocolMalformedMessage, diam.InvalidAVPLength, true, true)) + if !frontend.manager && ceaCalls.Load() != before+1 { + t.Fatalf("required error CEA invoked OnCEA %d times; want one", ceaCalls.Load()-before) + } + wire := h.answer(t, message, diam.InvalidAVPLength) + answer, err := diam.ReadMessage(bytes.NewReader(wire), h.dict) + if err != nil { + t.Fatal(err) + } + if answer.Header.CommandFlags&diam.ErrorFlag != 0 { + t.Fatal("permanent error used E-bit grammar") + } + failed, err := answer.FindAVP(avp.FailedAVP, 0) + if err != nil { + t.Fatal("invalid-length answer lost Failed-AVP", err) + } + group, ok := failed.Data.(*diam.GroupedAVP) + if !ok || len(group.AVP) != 1 || group.AVP[0].Code != avp.InbandSecurityID { + t.Fatalf("Failed-AVP=%+v", failed) + } + if frontend.manager { + h.barrier(t) + } else { + h.eof(t) + } + h.noEvents(t) + }) + } +} + +func TestProtocolTCPParityHeaderDecoderFatality(t *testing.T) { + for _, request := range []bool{false, true} { + for _, frontend := range parityFrontends { + t.Run(fmt.Sprintf("request=%t/%s", request, frontend.name), func(t *testing.T) { + h := parityStart(t, frontend, true) + message := contractBase(h.dict, diam.DeviceWatchdog, request) + message.Header.CommandFlags |= diam.ProxiableFlag + h.callback(t, message, parityBodyError()) + if request { + h.answer(t, message, diam.InvalidHDRBits) + } + h.callbackEvent(t, message, frontend.replyDisposition(diam.ProtocolInvalidHeader, diam.InvalidHDRBits, request, true)) + found := false + for _, record := range h.logs.Records() { + cause, ok := logtest.Attr(record, "error").Any().(*diam.MessageError) + if ok && cause.ResultCode == diam.InvalidHDRBits { + found = true + if !cause.Fatal { + t.Fatal("header diagnosis erased decoder fatality") + } + } + } + if !found { + t.Fatal("header diagnostic lost its typed error") + } + if frontend.manager { + h.barrier(t) + } else { + h.eof(t) + } + h.noEvents(t) + }) + } + } +} + +func TestProtocolTCPParityClosingDecoderFatality(t *testing.T) { + for _, frontend := range parityFrontends { + t.Run(frontend.name, func(t *testing.T) { + h := parityStart(t, frontend, true) + h.connection(t) + done := h.close() + local := h.read(t) + message := contractBase(h.dict, diam.DisconnectPeer, true) + h.callback(t, message, parityBodyError()) + h.callbackEvent(t, message, frontend.replyDisposition(diam.ProtocolMalformedMessage, diam.InvalidAVPLength, false, true)) + crossed := contractBase(h.dict, diam.DisconnectPeer, true) + h.write(t, crossed) + h.answer(t, crossed, diam.Success) + h.answerPeer(t, local) + if err := parityAwait(t, done); err != nil { + t.Fatal(err) + } + h.eof(t) + h.noEvents(t) + }) + } +} + +func TestProtocolTCPParityClosingHeaderDiagnosis(t *testing.T) { + for _, frontend := range parityFrontends { + t.Run(frontend.name, func(t *testing.T) { + h := parityStart(t, frontend, true) + h.connection(t) + done := h.close() + local := h.read(t) + message := contractBase(h.dict, diam.DisconnectPeer, true) + message.Header.CommandFlags |= diam.ProxiableFlag + h.callback(t, message, &diam.MessageError{ResultCode: diam.MissingAVP}) + h.callbackEvent(t, message, frontend.replyDisposition(diam.ProtocolInvalidHeader, diam.InvalidHDRBits, false, false)) + found := false + for _, record := range h.logs.Records() { + cause, ok := logtest.Attr(record, "error").Any().(*diam.MessageError) + if ok && cause.ResultCode == diam.InvalidHDRBits { + found = true + } + } + if !found { + t.Fatal("header cause was not logged") + } + crossed := contractBase(h.dict, diam.DisconnectPeer, true) + h.write(t, crossed) + h.answer(t, crossed, diam.Success) + h.answerPeer(t, local) + if err := parityAwait(t, done); err != nil { + t.Fatal(err) + } + h.eof(t) + h.noEvents(t) + }) + } +} + +func TestProtocolTCPParityFatalBodyReceipt(t *testing.T) { + for _, frontend := range parityFrontends { + if frontend.manager { + continue + } + t.Run(frontend.name, func(t *testing.T) { + t.Parallel() + parityNativeFatalBodyReceipt(t, frontend) + }) + } + for _, frontend := range parityFrontends { + if !frontend.manager { + continue + } + t.Run(frontend.name, func(t *testing.T) { + clock := new(parityClock) + h := parityStartConfigured(t, frontend, true, paritySettings{clock: clock}) + h.connection(t) + h.barrier(t) + clock.expire(t) + probe := h.read(t) + if probe.Header.CommandCode != diam.DeviceWatchdog || probe.Header.CommandFlags&diam.RequestFlag == 0 { + t.Fatal("watchdog did not publish DWR") + } + clock.expire(t) + for parityAwait(t, h.peers).Peer.Watchdog != peer.WatchdogSuspect { + } + before := clock.current(t) + clock.advance(10 * time.Second) + message := contractBase(h.dict, diam.DeviceWatchdog, true) + h.callback(t, message, parityBodyError()) + h.answer(t, message, diam.InvalidAVPLength) + h.callbackEvent(t, message, frontend.replyDisposition(diam.ProtocolMalformedMessage, diam.InvalidAVPLength, true, true)) + after := clock.current(t) + if after == before || !after.deadline.After(before.deadline) { + t.Fatal("fatal body receipt did not rearm Tw from receipt") + } + if state := h.manager.Peers()[0].Watchdog; state != peer.WatchdogSuspect { + t.Fatalf("invalid receipt changed health to %s", state) + } + h.answerPeer(t, probe) + for parityAwait(t, h.peers).Peer.Watchdog != peer.WatchdogOkay { + } + h.barrier(t) + h.noEvents(t) + }) + } +} + +// Holding the native transport writer exposes the interval between selecting +// a required fatal reply and completing it. The connection and reader are real +// TCP; only the selected answer's write is delayed. +type parityReplyGate struct { + armed, entered, release chan struct{} + once sync.Once +} + +type parityReplyConn struct { + net.Conn + gate *parityReplyGate +} + +func (c *parityReplyConn) Write(wire []byte) (int, error) { + select { + case <-c.gate.armed: + header, err := diam.DecodeHeader(wire) + if err == nil && header.CommandCode == diam.DeviceWatchdog && header.CommandFlags&diam.RequestFlag == 0 { + c.gate.once.Do(func() { close(c.gate.entered) }) + <-c.gate.release + } + default: + } + return c.Conn.Write(wire) +} + +type parityReplyListener struct { + net.Listener + gate *parityReplyGate +} + +func (l *parityReplyListener) Accept() (net.Conn, error) { + conn, err := l.Listener.Accept() + if err != nil { + return nil, err + } + return &parityReplyConn{Conn: conn, gate: l.gate}, nil +} + +func parityNativeFatalBodyReceipt(t *testing.T, frontend parityFrontend) { + t.Helper() + gate := &parityReplyGate{armed: make(chan struct{}), entered: make(chan struct{}), release: make(chan struct{})} + release := sync.OnceFunc(func() { close(gate.release) }) + defer release() + h := parityStartConfigured(t, frontend, true, paritySettings{interval: 12 * time.Second, writeGate: gate}) + h.barrier(t) + // Native watchdogs use the public interval with its production ±2s jitter. + // A 20s boundary includes the maximum 14s Tw without shortening policy. + if err := h.wire.SetReadDeadline(time.Now().Add(20 * time.Second)); err != nil { + t.Fatal(err) + } + probe, err := diam.ReadMessage(h.wire, h.dict) + if err != nil { + t.Fatal(err) + } + if probe.Header.CommandCode != diam.DeviceWatchdog || probe.Header.CommandFlags&diam.RequestFlag == 0 { + t.Fatal("watchdog did not publish DWR") + } + deadline := time.NewTimer(20 * time.Second) + defer deadline.Stop() + for suspect := false; !suspect; { + select { + case event := <-h.watchdog: + suspect = event == sm.WatchdogSuspect + case <-deadline.C: + t.Fatal("watchdog did not become SUSPECT with its DWR pending") + } + } + time.Sleep(8 * time.Second) + close(gate.armed) + message := contractBase(h.dict, diam.DeviceWatchdog, true) + callback := make(chan error, 1) + receipt := time.Now() + go func() { callback <- h.handler.HandleMessageError(h.conn, message, parityBodyError()) }() + parityAwait(t, gate.entered) + // Original Tw must have expired by 14s from SUSPECT. The replacement + // cannot expire before 10s from this receipt. At 8+8s only a reset survives. + time.Sleep(time.Until(receipt.Add(8 * time.Second))) + if elapsed := time.Since(receipt); elapsed >= 10*time.Second { + t.Fatalf("timer observation missed the receipt window: %s", elapsed) + } + if h.conn.Closed() { + t.Fatal("fatal valid-header receipt did not replace the old Tw before the required reply") + } + tracker, ok := diam.ConnAs[interface{ PendingRequests() *diam.PendingRequests }](h.conn) + if !ok || !tracker.PendingRequests().HasPending() { + t.Fatal("invalid fatal receipt credited the outstanding DWR") + } + select { + case event := <-h.watchdog: + t.Fatalf("invalid receipt changed watchdog health: %s", event) + default: + } + release() + h.answer(t, message, diam.InvalidAVPLength) + if err := parityAwait(t, callback); err != nil { + t.Fatal(err) + } + h.callbackEvent(t, message, frontend.replyDisposition(diam.ProtocolMalformedMessage, diam.InvalidAVPLength, true, true)) + h.eof(t) + // Completion disposes the refreshed Tw; neither it nor a stale expiry + // may publish a watchdog observation after fatal transport closure. + time.Sleep(15 * time.Second) + select { + case event := <-h.watchdog: + t.Fatalf("fatal completion left a watchdog callback active: %s", event) + default: + } + h.noEvents(t) +} + +// In-tree body decoders report recoverable errors. Fatal valid-header body +// errors therefore exercise the public direct-callback contract here; they do +// not run the native reader's independent fatal-error closure path. +func TestProtocolTCPParityUncorrelatedFatalCEA(t *testing.T) { + for _, fatal := range []bool{false, true} { + for _, coordinate := range []string{"hop", "end"} { + for _, frontend := range parityFrontends { + if !frontend.initiator { + continue + } + t.Run(fmt.Sprintf("direct-callback/fatal=%t/%s/%s", fatal, coordinate, frontend.name), func(t *testing.T) { + options := paritySettings{} + if frontend.manager { + options.clock = new(parityClock) + } + h := parityStartConfigured(t, frontend, false, options) + cer := h.read(t) + answer := contractBase(h.dict, diam.CapabilitiesExchange, false) + answer.Header.HopByHopID, answer.Header.EndToEndID = cer.Header.HopByHopID, cer.Header.EndToEndID + if coordinate == "hop" { + answer.Header.HopByHopID++ + } else { + answer.Header.EndToEndID++ + } + header := h.write(t, answer) + h.event(t, header, frontend.replyDisposition(diam.ProtocolUnexpectedMessage, 0, false, false)) + var deadline *parityTimer + if options.clock != nil { + deadline = options.clock.current(t) + } + description := parityBodyError() + description.Fatal = fatal + h.callback(t, answer, description) + h.callbackEvent(t, answer, frontend.replyDisposition(diam.ProtocolUnexpectedMessage, 0, false, fatal)) + select { + case <-h.ready: + t.Fatal("unowned error admitted connection") + default: + } + if fatal { + if frontend.manager { + if options.clock.current(t) != deadline || h.conn.Closed() { + t.Fatal("direct Manager callback took reader-owned closure or replaced CER deadline") + } + // A direct invocation has no reader to close the transport. + h.conn.Close() + } else if err := parityAwait(t, h.done); err == nil { + t.Fatal("fatal callback completed Client handshake successfully") + } + h.eof(t) + h.noEvents(t) + return + } + // Recoverable unmatched reports preserve ownership and the deadline. + if options.clock != nil && options.clock.current(t) != deadline { + t.Fatal("uncorrelated body error replaced the CER deadline") + } + answer.Header.HopByHopID, answer.Header.EndToEndID = cer.Header.HopByHopID, cer.Header.EndToEndID + h.write(t, answer) + parityAwait(t, h.ready) + if h.done != nil { + if err := parityAwait(t, h.done); err != nil { + t.Fatal(err) + } + } + h.barrier(t) + h.noEvents(t) + }) + } + } + } +} + +func TestProtocolTCPParityOwnedDWAIdentity(t *testing.T) { + for _, identity := range []uint32{avp.OriginHost, avp.OriginRealm} { + for _, frontend := range parityFrontends { + t.Run(fmt.Sprintf("%d/%s", identity, frontend.name), func(t *testing.T) { + t.Parallel() + options := paritySettings{probe: true} + if frontend.manager { + options.clock = new(parityClock) + } + h := parityStartConfigured(t, frontend, true, options) + if options.clock != nil { + h.barrier(t) + options.clock.expire(t) + } + probe := h.read(t) + if probe.Header.CommandCode != diam.DeviceWatchdog || probe.Header.CommandFlags&diam.RequestFlag == 0 { + t.Fatalf("probe=%+v", probe.Header) + } + answer := contractBase(h.dict, diam.DeviceWatchdog, false) + answer.Header.HopByHopID, answer.Header.EndToEndID = probe.Header.HopByHopID, probe.Header.EndToEndID + contractField(t, answer, avp.ResultCode).Data = datatype.Unsigned32(diam.UnableToComply) + field := contractField(t, answer, identity) + original := field.Data + field.Data = datatype.DiameterIdentity("other.example.net") + header := h.write(t, answer) + h.event(t, header, frontend.replyDisposition(diam.ProtocolInvalidDWA, 0, false, false)) + field.Data = original + contractField(t, answer, avp.ResultCode).Data = datatype.Unsigned32(diam.Success) + h.write(t, answer) + if !frontend.manager { + for parityAwait(t, h.watchdog) != sm.WatchdogAnswerReceived { + } + } + h.barrier(t) + h.noEvents(t) + }) + } + } +} + +func TestProtocolTCPParityRequiredReplyBuildFailure(t *testing.T) { + for _, frontend := range parityFrontends { + t.Run(frontend.name, func(t *testing.T) { + h := parityStart(t, frontend, true) + h.connection(t) + empty, err := dict.NewParser() + if err != nil { + t.Fatal(err) + } + message := diam.NewMessage(diam.DeviceWatchdog, diam.RequestFlag, 0, 17, 18, empty) + h.callback(t, message, &diam.MessageError{ResultCode: diam.InvalidAVPValue, Err: errors.New("invalid AVP payload")}) + h.callbackEvent(t, message, parityDisposition{Kind: diam.ProtocolMalformedMessage, Code: diam.InvalidAVPValue, Closing: true, Reply: diam.ProtocolReplyFailed, ReplyCode: diam.InvalidAVPValue}) + h.eof(t) + h.noEvents(t) + found := false + for _, record := range h.logs.Records() { + if record.Level != slog.LevelError { + continue + } + found = true + if frontend.manager { + cause := logtest.Attr(record, "error").Any() + if fmt.Sprintf("%T", cause) != "*fmt.wrapError" { + t.Fatalf("builder failure cause=%T, want standard wrapped error", cause) + } + } + } + if !found { + t.Fatal("required reply failure was not logged") + } + }) + } +} + +func TestProtocolTCPParityLocalDisconnectCause(t *testing.T) { + for _, frontend := range parityFrontends { + t.Run(frontend.name, func(t *testing.T) { + h := parityStart(t, frontend, true) + h.connection(t) + for _, cause := range []uint32{3, ^uint32(0)} { + err := parityAwait(t, h.closeCause(cause)) + if err == nil || err.Error() != fmt.Sprintf("invalid Disconnect-Cause %d", cause) || fmt.Sprintf("%T", err) != "*errors.errorString" { + t.Fatalf("invalid cause returned %v", err) + } + h.barrier(t) + h.noEvents(t) + } + done := h.close() + dpr := h.read(t) + if cause := contractField(t, dpr, avp.DisconnectCause).Data; cause != datatype.Enumerated(sm.DisconnectRebooting) { + t.Fatalf("DPR cause=%v", cause) + } + h.answerPeer(t, dpr) + if err := parityAwait(t, done); err != nil { + t.Fatal(err) + } + h.eof(t) + for _, cause := range []uint32{3, ^uint32(0)} { + err := parityAwait(t, h.closeCause(cause)) + if err == nil || err.Error() != fmt.Sprintf("invalid Disconnect-Cause %d", cause) || fmt.Sprintf("%T", err) != "*errors.errorString" { + t.Fatalf("invalid cause on closed connection returned %v", err) + } + } + h.noEvents(t) + }) + } +} + +func (h *parityTCP) answerPeer(t *testing.T, request *diam.Message) { + t.Helper() + answer := contractBase(h.dict, request.Header.CommandCode, false) + answer.Header.HopByHopID, answer.Header.EndToEndID = request.Header.HopByHopID, request.Header.EndToEndID + h.write(t, answer) +} diff --git a/diam/sm/accepted_log.go b/diam/sm/accepted_log.go index c468c20..e2e5d29 100644 --- a/diam/sm/accepted_log.go +++ b/diam/sm/accepted_log.go @@ -96,6 +96,10 @@ func (s *acceptedHandshake) record(in protocol.Input, n protocol.Notification) { text = "sm: CEA identifiers mismatch; answer discarded" break } + if n.Diagnostic == "dwa-correlation" { + text = "sm: DWA identifiers mismatch; answer discarded" + break + } if n.Diagnostic == "dpa-correlation" { text = "sm: DPA identifiers mismatch; answer discarded" break diff --git a/diam/sm/client_core_watchdog_test.go b/diam/sm/client_core_watchdog_test.go index 647add6..7edc97e 100644 --- a/diam/sm/client_core_watchdog_test.go +++ b/diam/sm/client_core_watchdog_test.go @@ -127,7 +127,7 @@ func TestClientCoreWatchdogConcurrentDuplicateDWA(t *testing.T) { if len(observed) != 1 || observed[0] != WatchdogAnswerReceived || after.Pending { t.Fatalf("concurrent DWA callbacks=%v watchdog=%+v; want one answer and no pending probe", observed, after) } - if got := len(logs.RecordsWithMessage("sm: unhandled answer discarded")); got != copies-1 { + if got := len(logs.RecordsWithMessage("sm: DWA identifiers mismatch; answer discarded")); got != copies-1 { t.Fatalf("duplicate DWA warnings = %d, want %d", got, copies-1) } } diff --git a/diam/sm/closing_contract_test.go b/diam/sm/closing_contract_test.go index 3e11493..767be38 100644 --- a/diam/sm/closing_contract_test.go +++ b/diam/sm/closing_contract_test.go @@ -370,6 +370,12 @@ func TestClosingContractRejectedDWAInFlightWire(t *testing.T) { if fault == "malformed" { wantKind, wantCode = diam.ProtocolInvalidDWA, diam.MissingAVP } + if fault == "mismatch" { + wantKind, wantCode = diam.ProtocolUnexpectedMessage, 0 + } + if fault == "identity" { + wantKind, wantCode = diam.ProtocolInvalidDWA, 0 + } if e.Err.Kind() != wantKind || e.Err.ResultCode() != wantCode || e.Reply != diam.ProtocolReplyNone { t.Fatalf("rejected DWA=%+v", e) } diff --git a/diam/sm/closing_watchdog_test.go b/diam/sm/closing_watchdog_test.go index bc0089b..7d07166 100644 --- a/diam/sm/closing_watchdog_test.go +++ b/diam/sm/closing_watchdog_test.go @@ -156,8 +156,10 @@ func TestAcceptedWatchdogEventsHoldStopLock(t *testing.T) { if err != nil { t.Fatal(err) } + invalid := acceptedDWA(t, request, diam.Success) + invalid.AVP = nil + receiveAcceptedTestWatchdog(w, invalid) receiveAcceptedTestWatchdog(w, acceptedDWA(t, request, diam.Success)) - receiveAcceptedTestWatchdog(w, acceptedDWA(t, regressionDWR(t), diam.TooBusy)) for _, event := range []WatchdogEvent{ WatchdogWriteFailed, WatchdogSuspect, WatchdogRecovered, WatchdogTimedOut, } { diff --git a/diam/sm/protocol_error_delivery_test.go b/diam/sm/protocol_error_delivery_test.go index dc2fd37..3853916 100644 --- a/diam/sm/protocol_error_delivery_test.go +++ b/diam/sm/protocol_error_delivery_test.go @@ -236,13 +236,18 @@ func TestProtocolErrorCallbacksOverlapOnSameConnection(t *testing.T) { func TestProtocolErrorWatchdogCallbackCanStopSupervision(t *testing.T) { testDict := newSMTestDictionary(creditcontrol.Dictionary()) - for _, invalid := range []bool{false, true} { - t.Run(fmt.Sprintf("invalid=%t", invalid), func(t *testing.T) { + for _, variant := range []string{"unowned", "rejected", "invalid"} { + t.Run(variant, func(t *testing.T) { + invalid := variant == "invalid" events := make(chan diam.ProtocolErrorEvent, 1) watchdog := make(chan WatchdogEvent, 4) var machine atomic.Pointer[StateMachine] p := newProtocolDeliveryWire(t, 0, true, func(cfg *Settings) { cfg.EnableWatchdog, cfg.WatchdogInterval = true, time.Hour + if variant != "unowned" { + cfg.WatchdogInterval = 200 * time.Millisecond + cfg.watchdogTiming = &watchdogTiming{floor: time.Millisecond} + } cfg.Dict = newTestDictionary(dict.Base(), nasreq.Dictionary()) cfg.Dict.SetStrict(false) cfg.AuthApplicationID = []*diam.AVP{diam.NewAVP(avp.AuthApplicationID, avp.Mbit, 0, datatype.Unsigned32(1001))} @@ -261,8 +266,18 @@ func TestProtocolErrorWatchdogCallbackCanStopSupervision(t *testing.T) { }) machine.Store(p.sm) m := diam.NewMessage(diam.DeviceWatchdog, 0, 0, 0x86a0, 0x86a1, testDict) + if variant != "unowned" { + request := acceptedWatchdogOnWire(t, p.peer) + if observed := protocolGenerationAwait(t, watchdog); observed != WatchdogRequestSent { + t.Fatalf("watchdog event = %s", observed) + } + m.Header.HopByHopID, m.Header.EndToEndID = request.Header.HopByHopID, request.Header.EndToEndID + } var payload datatype.Type = datatype.Unsigned32(diam.UnableToComply) kind, code, logMessage := diam.ProtocolRejectedDWA, uint32(diam.UnableToComply), "sm: rejected DWA" + if variant == "unowned" { + kind, code, logMessage = diam.ProtocolUnexpectedMessage, 0, "sm: DWA identifiers mismatch; answer discarded" + } if invalid { payload = localhostAddress kind, code, logMessage = diam.ProtocolInvalidDWA, diam.InvalidAVPLength, "sm: invalid DWA" @@ -273,7 +288,13 @@ func TestProtocolErrorWatchdogCallbackCanStopSupervision(t *testing.T) { header := p.send(t, m) event := protocolGenerationAwait(t, events) assertProtocolEvent(t, event, p.conn, &header, kind, code, false, diam.ProtocolReplyNone, 0) - if observed := protocolGenerationAwait(t, watchdog); observed != WatchdogInvalidAnswer { + if variant == "unowned" { + select { + case observed := <-watchdog: + t.Fatalf("unowned DWA received watchdog credit: %s", observed) + default: + } + } else if observed := protocolGenerationAwait(t, watchdog); observed != WatchdogInvalidAnswer { t.Fatalf("watchdog event = %s", observed) } p.usable(t) diff --git a/diam/sm/protocol_error_test.go b/diam/sm/protocol_error_test.go index 63c4cef..6daa785 100644 --- a/diam/sm/protocol_error_test.go +++ b/diam/sm/protocol_error_test.go @@ -89,13 +89,13 @@ func TestProtocolErrorRetainedWire(t *testing.T) { m.AddAVP(diam.NewAVP(avp.OriginRealm, avp.Mbit, 0, datatype.DiameterIdentity("example"))) return m }, diam.ProtocolInvalidDWA, diam.InvalidAVPLength, false, "sm: invalid DWA", slog.LevelWarn}, - {"A10_rejected_DWA", func(c *Settings) { c.EnableWatchdog = true; c.WatchdogInterval = time.Hour }, func(*testing.T) *diam.Message { + {"A10_unowned_DWA", func(c *Settings) { c.EnableWatchdog = true; c.WatchdogInterval = time.Hour }, func(*testing.T) *diam.Message { m := diam.NewMessage(diam.DeviceWatchdog, 0, 0, 17, 18, testDict) m.AddAVP(diam.NewAVP(avp.ResultCode, avp.Mbit, 0, datatype.Unsigned32(5012))) m.AddAVP(diam.NewAVP(avp.OriginHost, avp.Mbit, 0, datatype.DiameterIdentity("peer.example"))) m.AddAVP(diam.NewAVP(avp.OriginRealm, avp.Mbit, 0, datatype.DiameterIdentity("example"))) return m - }, diam.ProtocolRejectedDWA, 5012, false, "sm: rejected DWA", slog.LevelWarn}, + }, diam.ProtocolUnexpectedMessage, 0, false, "sm: DWA identifiers mismatch; answer discarded", slog.LevelWarn}, {"A11_invalid_DPR", nil, func(*testing.T) *diam.Message { return diam.NewMessage(diam.DisconnectPeer, diam.RequestFlag, 0, 17, 18, testDict) }, diam.ProtocolInvalidRequest, diam.MissingAVP, true, "sm: invalid request", slog.LevelWarn}, diff --git a/diam/sm/testdata/message_error_disposition.json b/diam/sm/testdata/message_error_disposition.json index 8b4a13a..374f60a 100644 --- a/diam/sm/testdata/message_error_disposition.json +++ b/diam/sm/testdata/message_error_disposition.json @@ -107,7 +107,15 @@ "err= closed=true replies=[] events=[malformed_message/5014/close=false/reply=0/0] logs=[] callbacks=0/0/0", "err= closed=true replies=[] events=[invalid_header/3008/close=true/reply=3/3008] logs=[] callbacks=0/0/0", "err= closed=true replies=[] events=[invalid_header/3008/close=false/reply=0/0] logs=[sm: invalid message header] callbacks=0/0/0", - "err= closed=true replies=[] events=[unexpected_message/0/close=false/reply=0/0] logs=[sm: ignoring CER on admitted connection; accepted CER renewal is disabled or ineligible] callbacks=0/0/0" + "err= closed=true replies=[] events=[unexpected_message/0/close=false/reply=0/0] logs=[sm: ignoring CER on admitted connection; accepted CER renewal is disabled or ineligible] callbacks=0/0/0", + "err= closed=true replies=[280/3008/20/[]] events=[invalid_header/3008/close=true/reply=2/3008] logs=[sm: invalid message header] callbacks=0/0/0", + "err= closed=true replies=[] events=[invalid_header/3008/close=true/reply=0/0] logs=[sm: invalid message header] callbacks=0/0/0", + "err= closed=true replies=[257/5005/0/[]] events=[malformed_message/5005/close=true/reply=2/5005] logs=[] callbacks=0/1/0", + "err= closed=true replies=[257/3001/20/[]] events=[malformed_message/3001/close=true/reply=2/3001] logs=[] callbacks=0/1/0", + "err= closed=true replies=[257/5004/0/[]] events=[malformed_message/5004/close=true/reply=2/5004] logs=[] callbacks=0/1/0", + "err= closed=true replies=[257/5014/0/[263]] events=[malformed_message/5014/close=true/reply=2/5014] logs=[] callbacks=0/1/0", + "err= closed=true replies=[282/3008/20/[]] events=[invalid_header/3008/close=true/reply=2/3008] logs=[sm: invalid message header] callbacks=0/0/0", + "err= closed=false replies=[] events=[invalid_header/3008/close=true/reply=0/0] logs=[sm: invalid message header] callbacks=0/0/0" ], "cases": { "manual/closed/CEA": [64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64], @@ -132,8 +140,8 @@ "manual/closed/nil": [64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64, 64], "manual/closing/CEA": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "manual/closing/CER": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], - "manual/closing/DPA": [2, 3, 3, 5, 5, 7, 7, 9, 9, 11, 11, 13, 13, 13, 13], - "manual/closing/DPR": [2, 3, 3, 5, 5, 7, 7, 9, 9, 11, 11, 13, 13, 13, 13], + "manual/closing/DPA": [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 13, 14], + "manual/closing/DPR": [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 13, 14], "manual/closing/DWA": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "manual/closing/DWR": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "manual/closing/app-CEA": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], @@ -145,8 +153,8 @@ "manual/closing/headerless": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "manual/closing/invalid-CEA": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "manual/closing/invalid-CER": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], - "manual/closing/invalid-DPA": [30, 30, 30, 30, 30, 30, 30, 30, 9, 30, 11, 30, 30, 30, 30], - "manual/closing/invalid-DPR": [30, 30, 30, 30, 30, 30, 30, 30, 9, 30, 11, 30, 30, 30, 30], + "manual/closing/invalid-DPA": [30, 30, 31, 30, 31, 30, 31, 30, 10, 30, 12, 30, 31, 30, 31], + "manual/closing/invalid-DPR": [30, 30, 31, 30, 31, 30, 31, 30, 10, 30, 12, 30, 31, 30, 31], "manual/closing/invalid-DWA": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "manual/closing/invalid-DWR": [63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "manual/closing/nil": [0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], @@ -171,29 +179,29 @@ "manual/open/invalid-DWR": [28, 28, 29, 28, 29, 28, 29, 28, 22, 28, 24, 28, 29, 28, 29], "manual/open/nil": [0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], "native/closed/CEA": [90, 98, 71, 99, 72, 100, 73, 101, 74, 102, 75, 103, 76, 103, 76], - "native/closed/CER": [90, 106, 106, 106, 106, 106, 106, 106, 94, 106, 95, 96, 96, 106, 106], + "native/closed/CER": [90, 106, 91, 106, 92, 106, 93, 106, 94, 106, 95, 96, 96, 106, 97], "native/closed/DPA": [90, 98, 71, 99, 72, 100, 73, 101, 74, 102, 75, 103, 76, 103, 76], "native/closed/DPR": [90, 91, 91, 92, 92, 93, 93, 94, 94, 95, 95, 96, 96, 97, 97], "native/closed/DWA": [90, 98, 71, 99, 72, 100, 73, 101, 74, 102, 75, 103, 76, 103, 76], "native/closed/DWR": [90, 91, 91, 92, 92, 93, 93, 94, 94, 95, 95, 96, 96, 97, 97], - "native/closed/app-CEA": [90, 105, 105, 105, 105, 105, 105, 105, 74, 105, 75, 105, 105, 105, 105], + "native/closed/app-CEA": [90, 105, 108, 105, 108, 105, 108, 105, 74, 105, 75, 105, 108, 105, 108], "native/closed/app-CER": [90, 104, 104, 104, 104, 104, 104, 104, 94, 104, 95, 96, 96, 104, 104], - "native/closed/app-DWA": [90, 105, 105, 105, 105, 105, 105, 105, 74, 105, 75, 105, 105, 105, 105], + "native/closed/app-DWA": [90, 105, 108, 105, 108, 105, 108, 105, 74, 105, 75, 105, 108, 105, 108], "native/closed/app-DWR": [90, 104, 104, 104, 104, 104, 104, 104, 94, 104, 95, 96, 96, 104, 104], "native/closed/app-answer": [90, 98, 71, 99, 72, 100, 73, 101, 74, 102, 75, 103, 76, 103, 76], "native/closed/app-request": [90, 56, 56, 92, 92, 59, 59, 60, 60, 61, 61, 96, 96, 62, 62], "native/closed/headerless": [90, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], - "native/closed/invalid-CEA": [90, 105, 105, 105, 105, 105, 105, 105, 74, 105, 75, 105, 105, 105, 105], - "native/closed/invalid-CER": [90, 106, 106, 106, 106, 106, 106, 106, 94, 106, 95, 96, 96, 106, 106], - "native/closed/invalid-DPA": [90, 105, 105, 105, 105, 105, 105, 105, 74, 105, 75, 105, 105, 105, 105], + "native/closed/invalid-CEA": [90, 105, 108, 105, 108, 105, 108, 105, 74, 105, 75, 105, 108, 105, 108], + "native/closed/invalid-CER": [90, 106, 104, 106, 104, 106, 104, 106, 94, 106, 95, 96, 96, 106, 104], + "native/closed/invalid-DPA": [90, 105, 108, 105, 108, 105, 108, 105, 74, 105, 75, 105, 108, 105, 108], "native/closed/invalid-DPR": [90, 104, 104, 104, 104, 104, 104, 104, 94, 104, 95, 96, 96, 104, 104], - "native/closed/invalid-DWA": [90, 105, 105, 105, 105, 105, 105, 105, 74, 105, 75, 105, 105, 105, 105], + "native/closed/invalid-DWA": [90, 105, 108, 105, 108, 105, 108, 105, 74, 105, 75, 105, 108, 105, 108], "native/closed/invalid-DWR": [90, 104, 104, 104, 104, 104, 104, 104, 94, 104, 95, 96, 96, 104, 104], "native/closed/nil": [90, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], "native/closing/CEA": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "native/closing/CER": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 25, 25, 63, 63], - "native/closing/DPA": [0, 3, 3, 5, 5, 7, 7, 9, 9, 11, 11, 13, 13, 13, 13], - "native/closing/DPR": [0, 3, 3, 5, 5, 7, 7, 9, 9, 11, 11, 25, 25, 13, 13], + "native/closing/DPA": [0, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 13, 14], + "native/closing/DPR": [0, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 25, 25, 13, 14], "native/closing/DWA": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "native/closing/DWR": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 25, 25, 63, 63], "native/closing/app-CEA": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], @@ -205,30 +213,30 @@ "native/closing/headerless": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "native/closing/invalid-CEA": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "native/closing/invalid-CER": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 25, 25, 63, 63], - "native/closing/invalid-DPA": [0, 78, 78, 78, 78, 78, 78, 78, 9, 78, 11, 78, 78, 78, 78], - "native/closing/invalid-DPR": [0, 78, 78, 78, 78, 78, 78, 78, 9, 78, 11, 25, 25, 78, 78], + "native/closing/invalid-DPA": [0, 78, 114, 78, 114, 78, 114, 78, 10, 78, 12, 78, 114, 78, 114], + "native/closing/invalid-DPR": [0, 78, 114, 78, 114, 78, 114, 78, 10, 78, 12, 25, 25, 78, 114], "native/closing/invalid-DWA": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "native/closing/invalid-DWR": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 25, 25, 63, 63], "native/closing/nil": [0, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63, 63], "native/open/CEA": [0, 3, 71, 5, 72, 7, 73, 9, 74, 11, 75, 13, 76, 13, 76], - "native/open/CER": [0, 32, 32, 32, 32, 32, 32, 32, 79, 32, 80, 25, 25, 32, 32], + "native/open/CER": [0, 32, 109, 32, 110, 32, 111, 32, 79, 32, 80, 25, 25, 32, 112], "native/open/DPA": [0, 3, 71, 5, 72, 7, 73, 9, 74, 11, 75, 13, 76, 13, 76], "native/open/DPR": [0, 42, 82, 44, 83, 46, 84, 48, 85, 50, 86, 25, 25, 52, 87], "native/open/DWA": [0, 3, 71, 5, 72, 7, 73, 9, 74, 11, 75, 13, 76, 13, 76], "native/open/DWR": [0, 15, 65, 17, 66, 19, 67, 21, 68, 23, 69, 25, 25, 26, 70], - "native/open/app-CEA": [0, 78, 78, 78, 78, 78, 78, 78, 74, 78, 75, 78, 78, 78, 78], + "native/open/app-CEA": [0, 78, 108, 78, 108, 78, 108, 78, 74, 78, 75, 78, 108, 78, 108], "native/open/app-CER": [0, 81, 81, 81, 81, 81, 81, 81, 40, 81, 41, 25, 25, 81, 81], - "native/open/app-DWA": [0, 78, 78, 78, 78, 78, 78, 78, 74, 78, 75, 78, 78, 78, 78], - "native/open/app-DWR": [0, 77, 77, 77, 77, 77, 77, 77, 68, 77, 69, 25, 25, 77, 77], + "native/open/app-DWA": [0, 78, 108, 78, 108, 78, 108, 78, 74, 78, 75, 78, 108, 78, 108], + "native/open/app-DWR": [0, 77, 107, 77, 107, 77, 107, 77, 68, 77, 69, 25, 25, 77, 107], "native/open/app-answer": [0, 3, 71, 5, 72, 7, 73, 9, 74, 11, 75, 13, 76, 13, 76], "native/open/app-request": [0, 56, 56, 57, 89, 59, 59, 60, 60, 61, 61, 25, 25, 62, 62], "native/open/headerless": [0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], - "native/open/invalid-CEA": [0, 78, 78, 78, 78, 78, 78, 78, 74, 78, 75, 78, 78, 78, 78], - "native/open/invalid-CER": [0, 32, 32, 32, 32, 32, 32, 32, 40, 32, 41, 25, 25, 32, 32], - "native/open/invalid-DPA": [0, 78, 78, 78, 78, 78, 78, 78, 74, 78, 75, 78, 78, 78, 78], - "native/open/invalid-DPR": [0, 88, 88, 88, 88, 88, 88, 88, 85, 88, 86, 25, 25, 88, 88], - "native/open/invalid-DWA": [0, 78, 78, 78, 78, 78, 78, 78, 74, 78, 75, 78, 78, 78, 78], - "native/open/invalid-DWR": [0, 77, 77, 77, 77, 77, 77, 77, 68, 77, 69, 25, 25, 77, 77], + "native/open/invalid-CEA": [0, 78, 108, 78, 108, 78, 108, 78, 74, 78, 75, 78, 108, 78, 108], + "native/open/invalid-CER": [0, 32, 81, 32, 81, 32, 81, 32, 40, 32, 41, 25, 25, 32, 81], + "native/open/invalid-DPA": [0, 78, 108, 78, 108, 78, 108, 78, 74, 78, 75, 78, 108, 78, 108], + "native/open/invalid-DPR": [0, 88, 113, 88, 113, 88, 113, 88, 85, 88, 86, 25, 25, 88, 113], + "native/open/invalid-DWA": [0, 78, 108, 78, 108, 78, 108, 78, 74, 78, 75, 78, 108, 78, 108], + "native/open/invalid-DWR": [0, 77, 107, 77, 107, 77, 107, 77, 68, 77, 69, 25, 25, 77, 107], "native/open/nil": [0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1] } } diff --git a/diam/sm/watchdog_correlation_test.go b/diam/sm/watchdog_correlation_test.go index e506fb9..09b266b 100644 --- a/diam/sm/watchdog_correlation_test.go +++ b/diam/sm/watchdog_correlation_test.go @@ -77,7 +77,7 @@ func TestWatchdogDWACorrelation(t *testing.T) { awaitAcceptedEvent(t, events, WatchdogSuspect, p.conn) assertProtocolEvent(t, protocolEvent(t, p.events), p.conn, &header, diam.ProtocolUnexpectedMessage, 0, false, diam.ProtocolReplyNone, 0) - records := p.logs.RecordsWithMessage("sm: unhandled answer discarded") + records := p.logs.RecordsWithMessage("sm: DWA identifiers mismatch; answer discarded") if len(records) != 1 || records[0].Level != slog.LevelWarn { t.Fatalf("unmatched DWA log records = %v, want one warning", records) }