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) +}