From 7d8a01081f0460b15c54f99f42eac2ca6e3c8412 Mon Sep 17 00:00:00 2001 From: Life <33139576+dust-life@users.noreply.github.com> Date: Mon, 26 Oct 2026 18:25:24 +0800 Subject: [PATCH] fix(server): make notify config update transactional UpdateNotifyConfig persisted the new configuration and then replaced the running notification service in two independent steps. When service initialization failed after the write, the RPC returned an error while the configuration on disk was already updated, leaving the persisted configuration and the running service inconsistent until the next restart. Rebuilding the service also mutated the shared notifier in place, so a partial failure could leave a half-initialized service set, and updating to a disabled configuration left the previous services active. Build and validate the complete replacement service set in memory first (notify.Notifier.Prepare), persist the configuration, then activate the replacement with an atomic, infallible swap. Any failure before the swap leaves both the persisted configuration and the running service untouched. The notifier now hands out immutable service snapshots via atomic.Pointer, so in-flight sends never observe a half-initialized set. UpdateNotifyConfig also rejects a nil request with InvalidArgument. --- server/internal/core/event.go | 17 ++- server/internal/core/notify_config_test.go | 35 ++++++ server/internal/notify/notify.go | 100 ++++++++++++++--- server/internal/notify/notify_test.go | 96 +++++++++++++++- server/rpc/rpc-config.go | 15 ++- server/rpc/rpc_notify_config_test.go | 123 +++++++++++++++++++++ 6 files changed, 364 insertions(+), 22 deletions(-) create mode 100644 server/internal/core/notify_config_test.go create mode 100644 server/rpc/rpc_notify_config_test.go diff --git a/server/internal/core/event.go b/server/internal/core/event.go index 6cc5711cb..867da003e 100644 --- a/server/internal/core/event.go +++ b/server/internal/core/event.go @@ -294,7 +294,7 @@ type eventBroker struct { subscribe chan eventSubscription unsubscribe chan chan Event send chan Event - notifier inotify.Notifier + notifier *inotify.Notifier cache *RingCache @@ -813,5 +813,20 @@ var ( ) func (broker *eventBroker) InitService(config *configs.NotifyConfig) error { + if broker == nil || broker.notifier == nil { + return ErrEventBrokerUnavailable + } return broker.notifier.InitService(config) } + +// PrepareNotifyConfig builds and validates the replacement notification +// service for config without changing the running one. The returned commit +// swaps it in atomically and cannot fail; callers persist the configuration +// first and commit only after the write succeeds, so the configuration on +// disk and the running service change together or not at all. +func (broker *eventBroker) PrepareNotifyConfig(config *configs.NotifyConfig) (commit func(), err error) { + if broker == nil || broker.notifier == nil { + return nil, ErrEventBrokerUnavailable + } + return broker.notifier.Prepare(config) +} diff --git a/server/internal/core/notify_config_test.go b/server/internal/core/notify_config_test.go new file mode 100644 index 000000000..4e713ebeb --- /dev/null +++ b/server/internal/core/notify_config_test.go @@ -0,0 +1,35 @@ +package core + +import ( + "errors" + "testing" + + "github.com/chainreactors/malice-network/server/internal/configs" + inotify "github.com/chainreactors/malice-network/server/internal/notify" +) + +func TestPrepareNotifyConfigTransaction(t *testing.T) { + broker := &eventBroker{notifier: inotify.NewNotifier()} + + if _, err := broker.PrepareNotifyConfig(&configs.NotifyConfig{ + Enable: true, + ServerChan: &configs.ServerChanConfig{Enable: true, URL: "not-a-url"}, + }); err == nil { + t.Fatal("expected a validation error") + } + + commit, err := broker.PrepareNotifyConfig(&configs.NotifyConfig{Enable: false}) + if err != nil { + t.Fatalf("PrepareNotifyConfig failed: %v", err) + } + commit() + + for _, unavailable := range []*eventBroker{nil, {}} { + if _, err := unavailable.PrepareNotifyConfig(&configs.NotifyConfig{}); !errors.Is(err, ErrEventBrokerUnavailable) { + t.Fatalf("unavailable broker prepare: %v", err) + } + if err := unavailable.InitService(&configs.NotifyConfig{}); !errors.Is(err, ErrEventBrokerUnavailable) { + t.Fatalf("unavailable broker init: %v", err) + } + } +} diff --git a/server/internal/notify/notify.go b/server/internal/notify/notify.go index 8c15fe902..90c057406 100644 --- a/server/internal/notify/notify.go +++ b/server/internal/notify/notify.go @@ -2,8 +2,11 @@ package notify import ( "context" + "errors" "fmt" "net/url" + "strings" + "sync/atomic" "time" "github.com/chainreactors/logs" @@ -15,39 +18,67 @@ import ( "github.com/nikoksr/notify/service/telegram" ) +// Notifier publishes complete, immutable service sets: a replacement is built +// and validated in memory first, then atomically swapped in, so a failed +// rebuild never touches the running services and a send never observes a +// half-initialized set. A Notifier must not be copied after first use. type Notifier struct { - notify *notify.Notify - enable bool + active atomic.Pointer[notify.Notify] } -func NewNotifier() Notifier { - return Notifier{notify: notify.New(), enable: false} +func NewNotifier() *Notifier { + return &Notifier{} } -func (n *Notifier) InitService(config *configs.NotifyConfig) error { +// buildServices validates cfg and constructs a complete replacement service +// set without touching the currently active one. +func buildServices(config *configs.NotifyConfig) (*notify.Notify, error) { + next := notify.New() if config == nil || !config.Enable { - return nil + next.Disabled = true + return next, nil + } + // Validate every enabled channel before constructing any provider, so a + // misconfiguration fails the whole update instead of partially building it. + if config.Telegram != nil && config.Telegram.Enable && + (strings.TrimSpace(config.Telegram.APIKey) == "" || config.Telegram.ChatID == 0) { + return nil, errors.New("server.notify.telegram requires api_key and a nonzero chat_id") + } + if config.DingTalk != nil && config.DingTalk.Enable && strings.TrimSpace(config.DingTalk.Token) == "" { + return nil, errors.New("server.notify.dingtalk requires token") + } + if config.Lark != nil && config.Lark.Enable { + if err := validateWebhook("server.notify.lark.webhook_url", config.Lark.WebHookUrl); err != nil { + return nil, err + } + } + if config.ServerChan != nil && config.ServerChan.Enable { + if err := validateWebhook("server.notify.serverchan.url", config.ServerChan.URL); err != nil { + return nil, err + } + } + if config.PushPlus != nil && config.PushPlus.Enable && strings.TrimSpace(config.PushPlus.Token) == "" { + return nil, errors.New("server.notify.pushplus requires token") } - n.enable = true if config.Telegram != nil && config.Telegram.Enable { tg, err := telegram.New(config.Telegram.APIKey) if err != nil { - return err + return nil, err } tg.SetParseMode(telegram.ModeMarkdown) tg.AddReceivers(config.Telegram.ChatID) - n.notify.UseServices(tg) + next.UseServices(tg) } if config.DingTalk != nil && config.DingTalk.Enable { dt := dingding.New(&dingding.Config{ Token: config.DingTalk.Token, Secret: config.DingTalk.Secret, }) - n.notify.UseServices(dt) + next.UseServices(dt) } if config.Lark != nil && config.Lark.Enable { larkSvc := NewLarkWebhookService(config.Lark.WebHookUrl, config.Lark.Secret) - n.notify.UseServices(larkSvc) + next.UseServices(larkSvc) } if config.ServerChan != nil && config.ServerChan.Enable { sc := http.New() @@ -62,7 +93,7 @@ func (n *Notifier) InitService(config *configs.NotifyConfig) error { return data.Encode() }, }) - n.notify.UseServices(sc) + next.UseServices(sc) } if config.PushPlus != nil && config.PushPlus.Enable { pp := http.New() @@ -81,17 +112,56 @@ func (n *Notifier) InitService(config *configs.NotifyConfig) error { } }, }) - n.notify.UseServices(pp) + next.UseServices(pp) + } + return next, nil +} + +func validateWebhook(path, raw string) error { + parsed, err := url.Parse(raw) + if err != nil || parsed.Hostname() == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") { + return fmt.Errorf("%s: expected an absolute http(s) url", path) } return nil } +// Prepare builds and validates the replacement service set for config without +// changing the active one. The returned commit activates the replacement +// atomically and cannot fail; callers that persist the configuration first +// invoke commit only after the write succeeds, keeping the persisted +// configuration and the running service in lockstep. +func (n *Notifier) Prepare(config *configs.NotifyConfig) (commit func(), err error) { + if n == nil { + return nil, errors.New("notifier is unavailable") + } + next, err := buildServices(config) + if err != nil { + return nil, err + } + return func() { n.active.Store(next) }, nil +} + +// InitService atomically replaces the active service set. It is the boot-time +// equivalent of Prepare+commit for callers that have no persistence step. +func (n *Notifier) InitService(config *configs.NotifyConfig) error { + commit, err := n.Prepare(config) + if err != nil { + return err + } + commit() + return nil +} + func (n *Notifier) Send(eventType, op, message string) { - if !n.enable { + if n == nil { + return + } + active := n.active.Load() + if active == nil || active.Disabled { return } title := fmt.Sprintf("[%s] %s", eventType, op) - err := n.notify.Send(context.Background(), title, message) + err := active.Send(context.Background(), title, message) if err != nil { logs.Log.Errorf("Failed to send notification: %s", err) } diff --git a/server/internal/notify/notify_test.go b/server/internal/notify/notify_test.go index a16ae08de..3f63056b2 100644 --- a/server/internal/notify/notify_test.go +++ b/server/internal/notify/notify_test.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "net/http/httptest" + "sync/atomic" "testing" "github.com/chainreactors/malice-network/server/internal/configs" @@ -104,7 +105,8 @@ func TestNotifierInitServiceUsesConfiguredChannels(t *testing.T) { if err := n.InitService(cfg); err != nil { t.Fatalf("InitService failed: %v", err) } - if !n.enable { + active := n.active.Load() + if active == nil || active.Disabled { t.Fatal("expected notifier to be enabled after loading config") } } @@ -115,7 +117,95 @@ func TestNotifierInitServiceSkipsDisabledConfig(t *testing.T) { if err := n.InitService(&configs.NotifyConfig{Enable: false}); err != nil { t.Fatalf("InitService failed: %v", err) } - if n.enable { - t.Fatal("expected notifier to remain disabled") + active := n.active.Load() + if active == nil || !active.Disabled { + t.Fatal("expected notifier to be disabled") + } +} + +func TestBuildServicesValidatesEnabledChannels(t *testing.T) { + cases := []struct { + name string + cfg *configs.NotifyConfig + }{ + {"telegram missing credentials", &configs.NotifyConfig{ + Enable: true, + Telegram: &configs.TelegramConfig{Enable: true}, + }}, + {"dingtalk missing token", &configs.NotifyConfig{ + Enable: true, + DingTalk: &configs.DingTalkConfig{Enable: true}, + }}, + {"lark invalid webhook", &configs.NotifyConfig{ + Enable: true, + Lark: &configs.LarkConfig{Enable: true, WebHookUrl: "not-a-url"}, + }}, + {"serverchan invalid url", &configs.NotifyConfig{ + Enable: true, + ServerChan: &configs.ServerChanConfig{Enable: true, URL: "not-a-url"}, + }}, + {"pushplus missing token", &configs.NotifyConfig{ + Enable: true, + PushPlus: &configs.PushPlusConfig{Enable: true}, + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if _, err := buildServices(tc.cfg); err == nil { + t.Fatal("expected a validation error") + } + }) + } +} + +func TestNotifierPrepareKeepsActiveServiceOnFailure(t *testing.T) { + n := NewNotifier() + good := &configs.NotifyConfig{ + Enable: true, + ServerChan: &configs.ServerChanConfig{Enable: true, URL: "https://serverchan.example/send"}, + } + if err := n.InitService(good); err != nil { + t.Fatalf("InitService failed: %v", err) + } + active := n.active.Load() + if _, err := n.Prepare(&configs.NotifyConfig{ + Enable: true, + ServerChan: &configs.ServerChanConfig{Enable: true, URL: "not-a-url"}, + }); err == nil { + t.Fatal("expected a validation error") + } + if n.active.Load() != active { + t.Fatal("failed prepare replaced the active service") + } +} + +func TestNotifierInitServiceReplacesPreviousServices(t *testing.T) { + var deliveries int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&deliveries, 1) + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"StatusCode":0,"StatusMessage":"success"}`)) + })) + defer srv.Close() + + n := NewNotifier() + if err := n.InitService(&configs.NotifyConfig{ + Enable: true, + Lark: &configs.LarkConfig{Enable: true, WebHookUrl: srv.URL}, + }); err != nil { + t.Fatalf("InitService failed: %v", err) + } + n.Send("test", "op", "message") + if atomic.LoadInt32(&deliveries) != 1 { + t.Fatal("expected the configured channel to deliver") + } + + // A disabled replacement must swap the old service set out, not leave it active. + if err := n.InitService(&configs.NotifyConfig{Enable: false}); err != nil { + t.Fatalf("InitService failed: %v", err) + } + n.Send("test", "op", "message") + if atomic.LoadInt32(&deliveries) != 1 { + t.Fatal("disabled config left the previous service active") } } diff --git a/server/rpc/rpc-config.go b/server/rpc/rpc-config.go index b1772e5ca..536dc1695 100644 --- a/server/rpc/rpc-config.go +++ b/server/rpc/rpc-config.go @@ -2,11 +2,14 @@ package rpc import ( "context" + "github.com/chainreactors/IoM-go/proto/client/clientpb" "github.com/chainreactors/IoM-go/types" "github.com/chainreactors/malice-network/helper/utils/configutil" "github.com/chainreactors/malice-network/server/internal/configs" "github.com/chainreactors/malice-network/server/internal/core" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func (rpc *Server) GetGithubConfig(ctx context.Context, req *clientpb.Empty) (*clientpb.GithubActionBuildConfig, error) { @@ -98,6 +101,9 @@ func (rpc *Server) GetNotifyConfig(ctx context.Context, req *clientpb.Empty) (*c } func (rpc *Server) UpdateNotifyConfig(ctx context.Context, req *clientpb.Notify) (*clientpb.Empty, error) { + if req == nil { + return nil, status.Error(codes.InvalidArgument, "notify config is required") + } notifyConfig := &configs.NotifyConfig{ Enable: req.TelegramEnable || req.DingtalkEnable || req.LarkEnable || req.ServerchanEnable || req.PushplusEnable, Telegram: &configs.TelegramConfig{ @@ -126,14 +132,17 @@ func (rpc *Server) UpdateNotifyConfig(ctx context.Context, req *clientpb.Notify) Channel: req.PushplusChannel, }, } - err := configs.UpdateNotifyConfig(notifyConfig) + // Build and validate the replacement service before persisting anything. + // Persisting is the last fallible step, so the configuration on disk and + // the running service change together or not at all. + commit, err := core.EventBroker.PrepareNotifyConfig(notifyConfig) if err != nil { return nil, err } - err = core.EventBroker.InitService(notifyConfig) - if err != nil { + if err = configs.UpdateNotifyConfig(notifyConfig); err != nil { return nil, err } + commit() return &clientpb.Empty{}, nil } diff --git a/server/rpc/rpc_notify_config_test.go b/server/rpc/rpc_notify_config_test.go new file mode 100644 index 000000000..a7c495c0d --- /dev/null +++ b/server/rpc/rpc_notify_config_test.go @@ -0,0 +1,123 @@ +package rpc + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/chainreactors/IoM-go/proto/client/clientpb" + "github.com/chainreactors/malice-network/server/internal/configs" + "github.com/chainreactors/malice-network/server/internal/core" + config "github.com/gookit/config/v2" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// TestUpdateNotifyConfigTransaction pins the transactional contract of +// UpdateNotifyConfig: the persisted configuration and the running +// notification service change together or not at all. +func TestUpdateNotifyConfigTransaction(t *testing.T) { + configs.InitTestConfigRuntime(t) + + dir := t.TempDir() + path := filepath.Join(dir, "config.yaml") + if err := os.WriteFile(path, []byte("server:\n enable: true\n"), 0o600); err != nil { + t.Fatal(err) + } + // Mirror the persistence hook PrepareConfig installs at server boot: + // every set.* event rewrites the config file. + config.WithOptions(config.WithHookFunc(func(event string, c *config.Config) { + if strings.HasPrefix(event, "set.") { + f, err := os.OpenFile(path, os.O_WRONLY|os.O_TRUNC, 0o600) + if err != nil { + return + } + defer f.Close() + _, _ = config.DumpTo(f, config.Yaml) + } + })) + if err := config.LoadFiles(path); err != nil { + t.Fatal(err) + } + + oldBroker := core.EventBroker + core.EventBroker = core.NewBroker() + t.Cleanup(func() { + core.EventBroker.Stop() + core.EventBroker = oldBroker + }) + + var deliveries atomic.Int64 + receiver := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + deliveries.Add(1) + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"StatusCode":0,"StatusMessage":"success"}`)) + })) + defer receiver.Close() + + waitDeliveries := func(want int64) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if deliveries.Load() >= want { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("deliveries = %d, want >= %d", deliveries.Load(), want) + } + + rpcServer := &Server{} + ctx := context.Background() + + if _, err := rpcServer.UpdateNotifyConfig(ctx, nil); status.Code(err) != codes.InvalidArgument { + t.Fatalf("nil request: got %v, want InvalidArgument", err) + } + + if _, err := rpcServer.UpdateNotifyConfig(ctx, &clientpb.Notify{ + LarkEnable: true, + LarkWebhookUrl: receiver.URL, + }); err != nil { + t.Fatalf("valid update failed: %v", err) + } + if got := configs.GetNotifyConfig(); got == nil || got.Lark == nil || !got.Enable || got.Lark.WebHookUrl != receiver.URL { + t.Fatalf("loaded configuration not updated: %#v", got) + } + persisted, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(persisted), receiver.URL) { + t.Fatalf("persisted configuration missing the new lark webhook url:\n%s", persisted) + } + core.EventBroker.Notify(core.Event{EventType: "test", Op: "op", Message: "message"}) + waitDeliveries(1) + + // A failing update must leave both the persisted configuration and the + // running service untouched. + if _, err := rpcServer.UpdateNotifyConfig(ctx, &clientpb.Notify{ + LarkEnable: true, + LarkWebhookUrl: "not-a-url", + }); err == nil { + t.Fatal("invalid update unexpectedly succeeded") + } + after, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(persisted, after) { + t.Fatal("failed update altered the persisted configuration") + } + if got := configs.GetNotifyConfig(); got == nil || got.Lark == nil || got.Lark.WebHookUrl != receiver.URL { + t.Fatalf("failed update altered the loaded configuration: %#v", got) + } + core.EventBroker.Notify(core.Event{EventType: "test", Op: "op", Message: "message"}) + waitDeliveries(2) +}