diff --git a/cmd/di.go b/cmd/di.go index 8a52520ff..f1aac4fe2 100644 --- a/cmd/di.go +++ b/cmd/di.go @@ -299,14 +299,15 @@ func (di *Dependencies) Bootstrap(nodeOptions node.Options) error { if err = di.handleConnStateChange(); err != nil { return err } + if err := di.subscribeNATStatusForPublicIP(); err != nil { + return err + } if err := di.Node.Start(); err != nil { return err } config.Current.EnableEventPublishing(di.EventBus) - di.handleNATStatusForPublicIP() - log.Info().Msg("Mysterium node started!") return nil } @@ -1103,7 +1104,16 @@ func (di *Dependencies) handleConnStateChange() error { }) } -func (di *Dependencies) handleNATStatusForPublicIP() { +// subscribeNATStatusForPublicIP reports the "public_ip" NAT traversal status once an identity +// is unlocked. The quality metric derived from this event is owned and signed by that identity, +// so it must not be published before an identity is available. +func (di *Dependencies) subscribeNATStatusForPublicIP() error { + return di.EventBus.SubscribeAsync(identity.AppTopicIdentityUnlock, func(e identity.AppEventIdentityUnlock) { + di.handleNATStatusForPublicIP(e.ID.Address) + }) +} + +func (di *Dependencies) handleNATStatusForPublicIP(id string) { outIP, err := di.IPResolver.GetOutboundIP() if err != nil { log.Error().Err(err).Msg("Failed to get outbound IP address") @@ -1115,7 +1125,7 @@ func (di *Dependencies) handleNATStatusForPublicIP() { } if outIP == pubIP && pubIP != "" { - di.EventBus.Publish(event.AppTopicTraversal, event.BuildSuccessfulEvent("", "public_ip")) + di.EventBus.Publish(event.AppTopicTraversal, event.BuildSuccessfulEvent(id, "public_ip")) } } diff --git a/cmd/di_nat_public_ip_test.go b/cmd/di_nat_public_ip_test.go new file mode 100644 index 000000000..1e28e413c --- /dev/null +++ b/cmd/di_nat_public_ip_test.go @@ -0,0 +1,68 @@ +/* + * Copyright (C) 2026 The "MysteriumNetwork/node" Authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package cmd + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + + "github.com/mysteriumnetwork/node/core/ip" + "github.com/mysteriumnetwork/node/eventbus" + "github.com/mysteriumnetwork/node/identity" + "github.com/mysteriumnetwork/node/nat/event" +) + +const testIdentity = "0x1111111111111111111111111111111111111111" + +func natEventsFor(t *testing.T, resolver ip.Resolver) <-chan event.Event { + di := &Dependencies{EventBus: eventbus.New(), IPResolver: resolver} + + events := make(chan event.Event, 1) + assert.NoError(t, di.EventBus.Subscribe(event.AppTopicTraversal, func(e event.Event) { events <- e })) + assert.NoError(t, di.subscribeNATStatusForPublicIP()) + + di.EventBus.Publish(identity.AppTopicIdentityUnlock, identity.AppEventIdentityUnlock{ID: identity.FromAddress(testIdentity)}) + return events +} + +// The "public_ip" NAT event feeds a quality metric that is signed by its owner, +// so it must carry the unlocked identity instead of an empty ID. +func TestNATStatusForPublicIP_PublishesWithUnlockedIdentity(t *testing.T) { + events := natEventsFor(t, ip.NewResolverMockMultiple("1.2.3.4", "1.2.3.4")) + + select { + case e := <-events: + assert.Equal(t, testIdentity, e.ID) + assert.Equal(t, "public_ip", e.Stage) + assert.True(t, e.Successful) + case <-time.After(2 * time.Second): + t.Fatal("expected public_ip NAT event after identity unlock") + } +} + +func TestNATStatusForPublicIP_SkipsWhenBehindNAT(t *testing.T) { + events := natEventsFor(t, ip.NewResolverMockMultiple("192.168.1.10", "1.2.3.4")) + + select { + case e := <-events: + t.Fatalf("unexpected NAT event behind NAT: %+v", e) + case <-time.After(300 * time.Millisecond): + } +} diff --git a/core/quality/trace_owner_test.go b/core/quality/trace_owner_test.go new file mode 100644 index 000000000..58644707d --- /dev/null +++ b/core/quality/trace_owner_test.go @@ -0,0 +1,108 @@ +/* + * Copyright (C) 2026 The "MysteriumNetwork/node" Authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package quality + +import ( + "io/fs" + "os" + "path/filepath" + "regexp" + "strings" + "testing" + + "github.com/stretchr/testify/assert" +) + +const ( + testProviderID = "0x1111111111111111111111111111111111111111" + testConsumerID = "0x2222222222222222222222222222222222222222" +) + +func traceContext(stage string) sessionTraceContext { + return sessionTraceContext{ + Stage: stage, + sessionContext: sessionContext{ + ID: "session-id", + Consumer: testConsumerID, + Provider: testProviderID, + }, + } +} + +// Provider-side trace stages must be owned (and therefore signed) by the provider. +// A provider node does not hold the consumer's key, so attributing a provider stage +// to the consumer makes signing fail with keystore.ErrLocked. +func TestTraceEventToMetricsEvent_ProviderStageOwnedByProvider(t *testing.T) { + for _, stage := range []string{ + "Provider session create", + "Provider session validation", + "Provider P2P exchange", + } { + owner, event := traceEventToMetricsEvent(traceContext(stage)) + + assert.Equal(t, testProviderID, owner, stage) + assert.True(t, event.IsProvider, stage) + assert.Equal(t, testConsumerID, event.TargetId, stage) + } +} + +func TestTraceEventToMetricsEvent_ConsumerStageOwnedByConsumer(t *testing.T) { + owner, event := traceEventToMetricsEvent(traceContext("Consumer session creation")) + + assert.Equal(t, testConsumerID, owner) + assert.False(t, event.IsProvider) + assert.Equal(t, testProviderID, event.TargetId) +} + +// traceEventToMetricsEvent derives the event owner from the stage name prefix. +// Every trace stage must therefore start with "Provider" or "Consumer", otherwise +// a provider-side stage is silently attributed to the consumer. +func TestTraceStageNames_HaveOwnerPrefix(t *testing.T) { + stageLiteral := regexp.MustCompile(`StartStage\("([^"]*)"\)`) + root := filepath.Join("..", "..") + + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + switch d.Name() { + case ".git", "vendor", "node_modules", "build": + return filepath.SkipDir + } + return nil + } + if !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { + return nil + } + + src, err := os.ReadFile(path) + if err != nil { + return err + } + for _, m := range stageLiteral.FindAllStringSubmatch(string(src), -1) { + stage := m[1] + assert.True(t, + strings.HasPrefix(stage, "Provider") || strings.HasPrefix(stage, "Consumer"), + "trace stage %q in %s must start with \"Provider\" or \"Consumer\"", stage, path, + ) + } + return nil + }) + assert.NoError(t, err) +} diff --git a/core/service/session_manager.go b/core/service/session_manager.go index 843e1d9b4..aade7c652 100644 --- a/core/service/session_manager.go +++ b/core/service/session_manager.go @@ -175,7 +175,7 @@ func (manager *SessionManager) Start(request *pb.SessionRequest) (_ pb.SessionRe validationWG := sync.WaitGroup{} validationWG.Add(1) go func() { - trace := session.tracer.StartStage("Session validation") + trace := session.tracer.StartStage("Provider session validation") validationError = manager.validateSession(session, prices) session.tracer.EndStage(trace) validationWG.Done() diff --git a/core/service/session_manager_test.go b/core/service/session_manager_test.go index 430a421ff..c91389a0d 100644 --- a/core/service/session_manager_test.go +++ b/core/service/session_manager_test.go @@ -139,7 +139,7 @@ func TestManager_Start_StoresSession(t *testing.T) { for _, key := range []string{ "Provider connect", "Provider session create", - "Session validation", + "Provider session validation", "Provider session create (start)", "Provider session create (payment)", "Provider session create (configure)", @@ -186,7 +186,7 @@ func TestManager_Start_DisconnectsOnPaymentError(t *testing.T) { for _, key := range []string{ "Provider connect", "Provider session create", - "Session validation", + "Provider session validation", "Provider session create (start)", "Provider session create (payment)", } {