diff --git a/api/api_test.go b/api/api_test.go index 9aec272..410b8eb 100644 --- a/api/api_test.go +++ b/api/api_test.go @@ -30,7 +30,7 @@ func makeApi(t *testing.T) *Api { pubsub := test.NewMemoryPubsub(t) assert.NotNil(t, pubsub) - communityManager, err := community.NewManager(cnf, db, pubsub, test.MustMakeAuditQueue(5)) + communityManager, err := community.NewManager(cnf, db, pubsub, test.NewMatrixNotifier(t)) assert.NoError(t, err) assert.NotNil(t, communityManager) diff --git a/cmd/app/main.go b/cmd/app/main.go index c424d96..6d5f67c 100644 --- a/cmd/app/main.go +++ b/cmd/app/main.go @@ -15,9 +15,9 @@ import ( "github.com/go-co-op/gocron/v2" "github.com/matrix-org/policyserv/config" - "github.com/matrix-org/policyserv/filter/audit" "github.com/matrix-org/policyserv/homeserver" "github.com/matrix-org/policyserv/logging" // import this for side effects if this isn't needed directly anymore + "github.com/matrix-org/policyserv/notifiers" "github.com/matrix-org/policyserv/pubsub" "github.com/matrix-org/policyserv/redaction" "github.com/matrix-org/policyserv/storage" @@ -52,12 +52,12 @@ func main() { defer db.Close() defer pubsubClient.Close() - auditQueue, err := audit.NewQueue(instanceConfig.WebhookPoolSize) + notifier, err := notifiers.NewWebhookMatrixNotifier(db, instanceConfig.WebhookPoolSize, instanceConfig.AllowedWebhookDomains) if err != nil { log.Fatal(err) } - communityManager, err := setupCommunityManager(instanceConfig, db, pubsubClient, auditQueue) + communityManager, err := setupCommunityManager(instanceConfig, db, pubsubClient, notifier) if err != nil { log.Fatal(err) } diff --git a/cmd/app/setup_community_manager.go b/cmd/app/setup_community_manager.go index 86b0610..7af8266 100644 --- a/cmd/app/setup_community_manager.go +++ b/cmd/app/setup_community_manager.go @@ -3,11 +3,11 @@ package main import ( "github.com/matrix-org/policyserv/community" "github.com/matrix-org/policyserv/config" - "github.com/matrix-org/policyserv/filter/audit" + "github.com/matrix-org/policyserv/notifiers" "github.com/matrix-org/policyserv/pubsub" "github.com/matrix-org/policyserv/storage" ) -func setupCommunityManager(instanceConfig *config.InstanceConfig, storage storage.PersistentStorage, pubsub pubsub.Client, auditQueue *audit.Queue) (*community.Manager, error) { - return community.NewManager(instanceConfig, storage, pubsub, auditQueue) +func setupCommunityManager(instanceConfig *config.InstanceConfig, storage storage.PersistentStorage, pubsub pubsub.Client, notifier notifiers.MatrixNotifier) (*community.Manager, error) { + return community.NewManager(instanceConfig, storage, pubsub, notifier) } diff --git a/community/manager.go b/community/manager.go index 0dcb342..e981e19 100644 --- a/community/manager.go +++ b/community/manager.go @@ -11,8 +11,8 @@ import ( "github.com/matrix-org/policyserv/config" "github.com/matrix-org/policyserv/content" "github.com/matrix-org/policyserv/filter" - "github.com/matrix-org/policyserv/filter/audit" "github.com/matrix-org/policyserv/internal" + "github.com/matrix-org/policyserv/notifiers" "github.com/matrix-org/policyserv/pubsub" "github.com/matrix-org/policyserv/storage" ) @@ -23,10 +23,10 @@ type Manager struct { roomToCommunityCache *cache.Cache[string, string] // room ID -> community ID instanceConfig *config.InstanceConfig pubsubClient pubsub.Client - auditQueue *audit.Queue + notifier notifiers.MatrixNotifier } -func NewManager(instanceConfig *config.InstanceConfig, storage storage.PersistentStorage, pubsubClient pubsub.Client, auditQueue *audit.Queue) (*Manager, error) { +func NewManager(instanceConfig *config.InstanceConfig, storage storage.PersistentStorage, pubsubClient pubsub.Client, notifier notifiers.MatrixNotifier) (*Manager, error) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) defer cancel() @@ -58,7 +58,7 @@ func NewManager(instanceConfig *config.InstanceConfig, storage storage.Persisten roomToCommunityCache: roomIdCache, instanceConfig: instanceConfig, pubsubClient: pubsubClient, - auditQueue: auditQueue, + notifier: notifier, }, nil } @@ -210,7 +210,7 @@ func (m *Manager) GetFilterSetForCommunityId(ctx context.Context, communityId st MaximumSpamVectorValue: 1.0, }}, } - filterSet, err := filter.NewSet(setConfig, m.storage, m.pubsubClient, m.auditQueue, scanner) + filterSet, err := filter.NewSet(setConfig, m.storage, m.pubsubClient, m.notifier, scanner) if err != nil { return nil, err } diff --git a/community/manager_test.go b/community/manager_test.go index 26008e3..be7ef73 100644 --- a/community/manager_test.go +++ b/community/manager_test.go @@ -24,7 +24,7 @@ func makeManager(t *testing.T) *Manager { assert.NoError(t, err) assert.NotNil(t, cnf) - manager, err := NewManager(cnf, db, pubsubClient, test.MustMakeAuditQueue(5)) + manager, err := NewManager(cnf, db, pubsubClient, test.NewMatrixNotifier(t)) assert.NoError(t, err) assert.NotNil(t, manager) diff --git a/filter/audit.go b/filter/audit.go index 5a63ea6..fb34bd5 100644 --- a/filter/audit.go +++ b/filter/audit.go @@ -6,17 +6,14 @@ import ( "fmt" "html" "log" - "net/url" "slices" "sync" - "testing" "time" - "github.com/matrix-org/policyserv/config" - "github.com/matrix-org/policyserv/filter/audit" + "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/policyserv/filter/classification" "github.com/matrix-org/policyserv/filter/confidence" - "github.com/matrix-org/gomatrixserverlib" + "github.com/matrix-org/policyserv/notifiers" ) type auditContext struct { @@ -25,26 +22,26 @@ type auditContext struct { FinalVectors confidence.Vectors IncrementalVectors []confidence.Vectors FilterResponses map[string][]classification.Classification - WebhookUrl string + CommunityId string - lock sync.Mutex // use a lock instead of a sync.Map because sync.Map doesn't support generics (and library support appears lacking in quality) - instanceConfig *config.InstanceConfig + lock sync.Mutex // use a lock instead of a sync.Map because sync.Map doesn't support generics (and library support appears lacking in quality) + notifier notifiers.MatrixNotifier } -func newAuditContext(instanceConfig *config.InstanceConfig, event gomatrixserverlib.PDU, webhookUrl string) (*auditContext, error) { +func newAuditContext(notifier notifiers.MatrixNotifier, communityId string, event gomatrixserverlib.PDU) (*auditContext, error) { return &auditContext{ Event: event, FilterResponses: make(map[string][]classification.Classification), IncrementalVectors: make([]confidence.Vectors, 0), - WebhookUrl: webhookUrl, + CommunityId: communityId, // Populated later IsSpam: false, FinalVectors: nil, // Internal - lock: sync.Mutex{}, - instanceConfig: instanceConfig, + lock: sync.Mutex{}, + notifier: notifier, }, nil } @@ -60,7 +57,7 @@ func (c *auditContext) AppendSetGroupVectors(vectors confidence.Vectors) { c.IncrementalVectors = append(c.IncrementalVectors, vectors) } -func (c *auditContext) Publish(workQueue *audit.Queue) error { +func (c *auditContext) Publish() error { c.lock.Lock() defer c.lock.Unlock() @@ -68,24 +65,10 @@ func (c *auditContext) Publish(workQueue *audit.Queue) error { // have an idea of what happened. log.Printf("[%s | %s | %s] Audit publish: %#v", c.Event.EventID(), c.Event.RoomID(), c.Event.SenderID(), c) - if c.WebhookUrl == "" || !c.IsSpam { + if !c.IsSpam { return nil // nothing to publish } - // Validate URL - whUrl, err := url.Parse(c.WebhookUrl) - if err != nil { - return err - } - if !testing.Testing() { - if whUrl.Scheme != "https" { - return fmt.Errorf("webhook URL must be HTTPS") - } - } - if !slices.Contains(c.instanceConfig.AllowedWebhookDomains, whUrl.Host) { - return fmt.Errorf("webhook URL host not allowed") - } - respsJson, err := json.MarshalIndent(c.FilterResponses, "", " ") if err != nil { return err // "should never happen" @@ -116,5 +99,15 @@ func (c *auditContext) Publish(workQueue *audit.Queue) error { htmlAudit += "" // close the details block from earlier } - return workQueue.Submit(c.Event.EventID(), htmlAudit, whUrl.String()) + // we don't html2text this because long events can cause hookshot to only show text versions, making all + // of our work to contain the spam to a
block useless. We still put some sort of message here + // though so clients which don't support HTML can still see something useful. + textAudit := "This event requires HTML." + + msgId, err := c.notifier.Send(c.CommunityId, textAudit, htmlAudit) + if err != nil { + return fmt.Errorf("failed to send audit message: %w", err) + } + log.Printf("[%s | %s | %s] Audit message sent: %s", c.Event.EventID(), c.Event.RoomID(), c.Event.SenderID(), msgId) + return nil } diff --git a/filter/audit/queue.go b/filter/audit/queue.go deleted file mode 100644 index e5bcef3..0000000 --- a/filter/audit/queue.go +++ /dev/null @@ -1,76 +0,0 @@ -package audit - -import ( - "bytes" - "encoding/json" - "log" - "net/http" - "time" - - "github.com/panjf2000/ants/v2" -) - -type Queue struct { - pool *ants.Pool -} - -func NewQueue(size int) (*Queue, error) { - pool, err := ants.NewPool(size, ants.WithOptions(ants.Options{ - // Same options as the queue.Pool setup - ExpiryDuration: 1 * time.Minute, - PreAlloc: false, - MaxBlockingTasks: 0, // no limit on submissions - Nonblocking: false, - // If we don't supply a panic handler then ants will print a stack trace for us - //PanicHandler: func(err interface{}) { - // log.Println("Panic in pool:", err) - //}, - Logger: log.Default(), - DisablePurge: false, - })) - if err != nil { - return nil, err - } - return &Queue{ - pool: pool, - }, nil -} - -func (q *Queue) Submit(eventId string, html string, webhookUrl string) error { - workFn := func() { - // Hookshot / Slack format body - reqBody := make(map[string]any) - reqBody["html"] = html - - // we don't html2text this because long events can cause hookshot to only show text versions, making all - // of our work to contain the spam to a
block useless. We still put some sort of message here - // though so clients which don't support HTML can still see something useful. - reqBody["text"] = "This event requires HTML." - - buf := bytes.NewBuffer(nil) - encoder := json.NewEncoder(buf) - encoder.SetEscapeHTML(false) // we get XSS protection from elsewhere, so avoid making the JSON hard to read by humans/tests - err := encoder.Encode(reqBody) - if err != nil { - log.Printf("[%s] Failed to encode JSON: %s", eventId, err) - return - } - - req, err := http.NewRequest("POST", webhookUrl, buf) - if err != nil { - log.Printf("[%s] Failed to create request: %s", eventId, err) - return - } - req.Header.Set("Content-Type", "application/json") - req.Header.Set("User-Agent", "policyserv") - - res, err := http.DefaultClient.Do(req) - if err != nil { - log.Printf("[%s] Failed to send request: %s", eventId, err) - return - } - defer res.Body.Close() - log.Printf("[%s] Audit webhook response: %s", eventId, res.Status) - } - return q.pool.Submit(workFn) -} diff --git a/filter/filter_density_test.go b/filter/filter_density_test.go index a817638..7b2cd46 100644 --- a/filter/filter_density_test.go +++ b/filter/filter_density_test.go @@ -29,7 +29,7 @@ func TestDensityFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_event_type_test.go b/filter/filter_event_type_test.go index 09a7ae5..b657acb 100644 --- a/filter/filter_event_type_test.go +++ b/filter/filter_event_type_test.go @@ -27,7 +27,7 @@ func TestEventTypeFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_frequency_test.go b/filter/filter_frequency_test.go index 275ce3e..cb11aa9 100644 --- a/filter/filter_frequency_test.go +++ b/filter/filter_frequency_test.go @@ -32,7 +32,7 @@ func TestFrequencyFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_hellban_test.go b/filter/filter_hellban_test.go index d43d935..0cd04e1 100644 --- a/filter/filter_hellban_test.go +++ b/filter/filter_hellban_test.go @@ -53,7 +53,7 @@ func TestHellbanPrefilterDoesntEternallyExtend(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -107,7 +107,7 @@ func TestHellbanPrefilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -184,7 +184,7 @@ func TestHellbanPostfilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -284,7 +284,7 @@ func TestHellbanFiltersCombined(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_inline_emoji_size_test.go b/filter/filter_inline_emoji_size_test.go index c8efeb9..10c02f4 100644 --- a/filter/filter_inline_emoji_size_test.go +++ b/filter/filter_inline_emoji_size_test.go @@ -221,7 +221,7 @@ func TestInlineEmojiSizeFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -267,7 +267,7 @@ func TestInlineEmojiSizeFilterRejectsInvalidHtml(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_keyword_template_test.go b/filter/filter_keyword_template_test.go index 9f5641c..a683046 100644 --- a/filter/filter_keyword_template_test.go +++ b/filter/filter_keyword_template_test.go @@ -61,7 +61,7 @@ func TestKeywordTemplateFilter(t *testing.T) { }) assert.NoError(t, err) - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -161,7 +161,7 @@ func TestKeywordTemplateFilterWithFullEvent(t *testing.T) { }) assert.NoError(t, err) - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_keywords_test.go b/filter/filter_keywords_test.go index 339be9d..544781e 100644 --- a/filter/filter_keywords_test.go +++ b/filter/filter_keywords_test.go @@ -30,7 +30,7 @@ func TestKeywordsFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -106,7 +106,7 @@ func TestKeywordsFilterWithFullEvent(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_length_test.go b/filter/filter_length_test.go index 5caa3b0..9967d4c 100644 --- a/filter/filter_length_test.go +++ b/filter/filter_length_test.go @@ -31,7 +31,7 @@ func TestLengthFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_links_test.go b/filter/filter_links_test.go index 26c1e98..777d26b 100644 --- a/filter/filter_links_test.go +++ b/filter/filter_links_test.go @@ -30,7 +30,7 @@ func TestLinkFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -148,7 +148,7 @@ func TestLinkFilterDenyListOnly(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_many_ats_test.go b/filter/filter_many_ats_test.go index 25ea149..749edce 100644 --- a/filter/filter_many_ats_test.go +++ b/filter/filter_many_ats_test.go @@ -27,7 +27,7 @@ func TestManyAtsFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_media_scanning_test.go b/filter/filter_media_scanning_test.go index fb05862..9e55c50 100644 --- a/filter/filter_media_scanning_test.go +++ b/filter/filter_media_scanning_test.go @@ -32,7 +32,7 @@ func TestMediaScanningFilter(t *testing.T) { ps := test.NewMemoryPubsub(t) defer ps.Close() scanner := test.NewMemoryContentScanner(t) - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), scanner) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), scanner) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_media_test.go b/filter/filter_media_test.go index 37abbca..bc13f60 100644 --- a/filter/filter_media_test.go +++ b/filter/filter_media_test.go @@ -26,7 +26,7 @@ func TestMediaFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_mentions_frequency_test.go b/filter/filter_mentions_frequency_test.go index b8c6bdb..639d9ff 100644 --- a/filter/filter_mentions_frequency_test.go +++ b/filter/filter_mentions_frequency_test.go @@ -32,7 +32,7 @@ func TestMentionsFrequencyFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_mentions_test.go b/filter/filter_mentions_test.go index e7e3b40..c3dd0e5 100644 --- a/filter/filter_mentions_test.go +++ b/filter/filter_mentions_test.go @@ -48,7 +48,7 @@ func TestMentionsFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_mjolnir_test.go b/filter/filter_mjolnir_test.go index cdae581..abd6e91 100644 --- a/filter/filter_mjolnir_test.go +++ b/filter/filter_mjolnir_test.go @@ -33,7 +33,7 @@ func TestMjolnirFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_openai_omni_test.go b/filter/filter_openai_omni_test.go index a6ccba3..165ab9f 100644 --- a/filter/filter_openai_omni_test.go +++ b/filter/filter_openai_omni_test.go @@ -39,7 +39,7 @@ func TestOpenAIOmniFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_protect_local_user_test.go b/filter/filter_protect_local_user_test.go index 06bcef5..11272c9 100644 --- a/filter/filter_protect_local_user_test.go +++ b/filter/filter_protect_local_user_test.go @@ -32,7 +32,7 @@ func TestProtectLocalUserFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_sender_test.go b/filter/filter_sender_test.go index 343ace9..fadd795 100644 --- a/filter/filter_sender_test.go +++ b/filter/filter_sender_test.go @@ -26,7 +26,7 @@ func TestSenderFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_sticky_events_test.go b/filter/filter_sticky_events_test.go index 70ffcab..eecfd4f 100644 --- a/filter/filter_sticky_events_test.go +++ b/filter/filter_sticky_events_test.go @@ -29,7 +29,7 @@ func TestStickyEventsFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_trim_length_test.go b/filter/filter_trim_length_test.go index 31beb95..224e2ce 100644 --- a/filter/filter_trim_length_test.go +++ b/filter/filter_trim_length_test.go @@ -28,7 +28,7 @@ func TestTrimLengthFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_unsafe_signing_key_test.go b/filter/filter_unsafe_signing_key_test.go index 55ff60c..f873345 100644 --- a/filter/filter_unsafe_signing_key_test.go +++ b/filter/filter_unsafe_signing_key_test.go @@ -6,10 +6,10 @@ import ( "math/rand" "testing" + "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/policyserv/config" "github.com/matrix-org/policyserv/filter/classification" "github.com/matrix-org/policyserv/test" - "github.com/matrix-org/gomatrixserverlib" "github.com/stretchr/testify/assert" ) @@ -28,7 +28,7 @@ func TestUnsafeSigningKeyFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_untrusted_media_test.go b/filter/filter_untrusted_media_test.go index b013163..6d751db 100644 --- a/filter/filter_untrusted_media_test.go +++ b/filter/filter_untrusted_media_test.go @@ -32,7 +32,7 @@ func TestUntrustedMediaFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_user_id_contains_words_test.go b/filter/filter_user_id_contains_words_test.go index 7150d27..e2779af 100644 --- a/filter/filter_user_id_contains_words_test.go +++ b/filter/filter_user_id_contains_words_test.go @@ -27,7 +27,7 @@ func TestUserIdContainsWordsFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/filter_user_id_length_test.go b/filter/filter_user_id_length_test.go index 963e7ba..0c64984 100644 --- a/filter/filter_user_id_length_test.go +++ b/filter/filter_user_id_length_test.go @@ -27,7 +27,7 @@ func TestUserIdLengthFilter(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/filter/foundation_test.go b/filter/foundation_test.go index 07fb6cc..caf3d3a 100644 --- a/filter/foundation_test.go +++ b/filter/foundation_test.go @@ -60,7 +60,7 @@ func TestMatrixFoundationIntents(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) assert.NoError(t, memStorage.SetListBanRules(context.Background(), mjolnirRoomId, map[string]string{ diff --git a/filter/set.go b/filter/set.go index cf1cf25..f48c38c 100644 --- a/filter/set.go +++ b/filter/set.go @@ -11,11 +11,11 @@ import ( "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/policyserv/config" "github.com/matrix-org/policyserv/content" - "github.com/matrix-org/policyserv/filter/audit" "github.com/matrix-org/policyserv/filter/classification" "github.com/matrix-org/policyserv/filter/confidence" "github.com/matrix-org/policyserv/internal" "github.com/matrix-org/policyserv/media" + "github.com/matrix-org/policyserv/notifiers" "github.com/matrix-org/policyserv/pubsub" "github.com/matrix-org/policyserv/storage" ) @@ -31,7 +31,7 @@ type SetConfig struct { type Set struct { storage storage.PersistentStorage pubsub pubsub.Client - queue *audit.Queue + notifier notifiers.MatrixNotifier contentScanner content.Scanner // may be nil if scanning not possible groups []*setGroup communityConfig *config.CommunityConfig @@ -39,11 +39,14 @@ type Set struct { communityId string } -func NewSet(config *SetConfig, storage storage.PersistentStorage, pubsub pubsub.Client, queue *audit.Queue, contentScanner content.Scanner) (*Set, error) { +func NewSet(config *SetConfig, storage storage.PersistentStorage, pubsub pubsub.Client, notifier notifiers.MatrixNotifier, contentScanner content.Scanner) (*Set, error) { + if config.CommunityId == "" { + config.CommunityId = "default" + } set := &Set{ storage: storage, pubsub: pubsub, - queue: queue, + notifier: notifier, contentScanner: contentScanner, groups: make([]*setGroup, len(config.Groups)), communityConfig: config.CommunityConfig, @@ -91,7 +94,7 @@ func (s *Set) CheckEvent(ctx context.Context, event gomatrixserverlib.PDU, media vecs := confidence.NewConfidenceVectors() vecs.SetVector(classification.Spam, 0.5) // per docs elsewhere, start by assuming 50% likelihood of spam - auditCtx, err := newAuditContext(s.instanceConfig, event, internal.Dereference(s.communityConfig.WebhookUrl)) + auditCtx, err := newAuditContext(s.notifier, s.communityId, event) if err != nil { return nil, err } @@ -140,7 +143,7 @@ func (s *Set) CheckEvent(ctx context.Context, event gomatrixserverlib.PDU, media auditCtx.FinalVectors = vecs auditCtx.IsSpam = s.IsSpamResponse(ctx, vecs) go func(auditCtx *auditContext, s *Set) { // run the audit publishing async to avoid blocking the hot path any more than required - err := auditCtx.Publish(s.queue) + err := auditCtx.Publish() if err != nil { log.Printf("[%s | %s] Non-fatal error publishing audit: %s", auditCtx.Event.EventID(), auditCtx.Event.RoomID().String(), err) } diff --git a/filter/set_group_test.go b/filter/set_group_test.go index f5d5362..beb9143 100644 --- a/filter/set_group_test.go +++ b/filter/set_group_test.go @@ -5,7 +5,6 @@ import ( "slices" "testing" - "github.com/matrix-org/policyserv/config" "github.com/matrix-org/policyserv/filter/classification" "github.com/matrix-org/policyserv/filter/confidence" "github.com/matrix-org/policyserv/test" @@ -19,7 +18,7 @@ func TestSetGroupCheck(t *testing.T) { Type: "m.room.message", Content: make(map[string]any), }) - auditCtx, err := newAuditContext(&config.InstanceConfig{}, event, "") + auditCtx, err := newAuditContext(test.NewMatrixNotifier(t), "default", event) assert.NoError(t, err) assert.NotNil(t, auditCtx) input := &EventInput{ diff --git a/filter/set_test.go b/filter/set_test.go index e7649e1..8fbb8ff 100644 --- a/filter/set_test.go +++ b/filter/set_test.go @@ -7,6 +7,7 @@ import ( "io" "net/http" "net/http/httptest" + "net/url" "strings" "testing" "time" @@ -16,6 +17,8 @@ import ( "github.com/matrix-org/policyserv/filter/confidence" "github.com/matrix-org/policyserv/internal" "github.com/matrix-org/policyserv/media" + "github.com/matrix-org/policyserv/notifiers" + "github.com/matrix-org/policyserv/storage" "github.com/matrix-org/policyserv/test" "github.com/stretchr/testify/assert" ) @@ -33,7 +36,7 @@ func TestNewSet(t *testing.T) { ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -61,7 +64,7 @@ func TestNewSetUnknownFilter(t *testing.T) { ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.ErrorContains(t, err, "error finding filter name") assert.Nil(t, set) } @@ -79,7 +82,7 @@ func TestNewSetErrorMaking(t *testing.T) { ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.ErrorContains(t, err, "error making filter") assert.Nil(t, set) } @@ -102,7 +105,7 @@ func TestSetCheckEvent(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -167,7 +170,7 @@ func TestCheckEventWithErrorInGroup(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -229,7 +232,7 @@ func TestSetCheckText(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -283,7 +286,7 @@ func TestSetCheckTextWithErrorInGroup(t *testing.T) { defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -320,7 +323,7 @@ func TestSetSpamThreshold(t *testing.T) { ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) @@ -367,6 +370,8 @@ func TestCallsWebhook(t *testing.T) { }) server := httptest.NewServer(handler) defer server.Close() + parsedUrl, err := url.Parse(server.URL) + assert.NoError(t, err) cnf := &SetConfig{ CommunityConfig: &config.CommunityConfig{ @@ -382,19 +387,26 @@ func TestCallsWebhook(t *testing.T) { MinimumSpamVectorValue: 0.0, MaximumSpamVectorValue: 1.0, }}, - InstanceConfig: &config.InstanceConfig{ - AllowedWebhookDomains: []string{server.Listener.Addr().String()}, - }, } memStorage := test.NewMemoryStorage(t) defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + notifier, err := notifiers.NewWebhookMatrixNotifier(memStorage, 5, []string{parsedUrl.Host}) + assert.NoError(t, err) + assert.NotNil(t, notifier) + set, err := NewSet(cnf, memStorage, ps, notifier, nil) assert.NoError(t, err) assert.NotNil(t, set) + // Insert the community so the notifier works + err = memStorage.UpsertCommunity(context.Background(), &storage.StoredCommunity{ + CommunityId: set.communityId, + Config: set.communityConfig, + }) + assert.NoError(t, err) + event := test.MustMakePDU(&test.BaseClientEvent{ RoomId: "!foo:example.org", EventId: "$test", @@ -452,6 +464,8 @@ func TestCallsWebhookErrorNonFatal(t *testing.T) { }) server := httptest.NewServer(handler) defer server.Close() + parsedUrl, err := url.Parse(server.URL) + assert.NoError(t, err) cnf := &SetConfig{ CommunityConfig: &config.CommunityConfig{ @@ -463,19 +477,26 @@ func TestCallsWebhookErrorNonFatal(t *testing.T) { MinimumSpamVectorValue: 0.0, MaximumSpamVectorValue: 1.0, }}, - InstanceConfig: &config.InstanceConfig{ - AllowedWebhookDomains: []string{server.Listener.Addr().String()}, - }, } memStorage := test.NewMemoryStorage(t) defer memStorage.Close() ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + notifier, err := notifiers.NewWebhookMatrixNotifier(memStorage, 5, []string{parsedUrl.Host}) + assert.NoError(t, err) + assert.NotNil(t, notifier) + set, err := NewSet(cnf, memStorage, ps, notifier, nil) assert.NoError(t, err) assert.NotNil(t, set) + // Insert the community so the notifier works + err = memStorage.UpsertCommunity(context.Background(), &storage.StoredCommunity{ + CommunityId: set.communityId, + Config: set.communityConfig, + }) + assert.NoError(t, err) + event := test.MustMakePDU(&test.BaseClientEvent{ RoomId: "!foo:example.org", EventId: "$test", @@ -526,7 +547,7 @@ func TestExtractsMedia(t *testing.T) { ps := test.NewMemoryPubsub(t) defer ps.Close() - set, err := NewSet(cnf, memStorage, ps, test.MustMakeAuditQueue(5), nil) + set, err := NewSet(cnf, memStorage, ps, test.NewMatrixNotifier(t), nil) assert.NoError(t, err) assert.NotNil(t, set) diff --git a/homeserver/test_utils.go b/homeserver/test_utils.go index 807003e..ea6131b 100644 --- a/homeserver/test_utils.go +++ b/homeserver/test_utils.go @@ -48,7 +48,7 @@ func NewMockServerForTest(t *testing.T, storage storage.PersistentStorage, confi assert.NoError(t, err) assert.NotNil(t, instanceCnf) pubsub := test.NewMemoryPubsub(t) - communityManager, err := community.NewManager(instanceCnf, storage, pubsub, test.MustMakeAuditQueue(5)) + communityManager, err := community.NewManager(instanceCnf, storage, pubsub, test.NewMatrixNotifier(t)) assert.NoError(t, err) assert.NotNil(t, communityManager) pool, err := queue.NewPool(&queue.PoolConfig{ diff --git a/notifiers/interface.go b/notifiers/interface.go new file mode 100644 index 0000000..3c4f64b --- /dev/null +++ b/notifiers/interface.go @@ -0,0 +1,10 @@ +package notifiers + +// MatrixNotifier - Used to send notifications of activity to Matrix. Note that this accepts a "target" +// which might not be a room ID depending on the implementation. +type MatrixNotifier interface { + // Send - Sends an `m.notice` message to the given community ID, if possible. The implementation might queue the + // message for later delivery - the returned error represents a queue failure in this case rather than a delivery + // failure. Returns a "message ID" for logging purposes. + Send(communityId string, plainText string, htmlText string) (string, error) +} diff --git a/notifiers/webhooks.go b/notifiers/webhooks.go new file mode 100644 index 0000000..994d05f --- /dev/null +++ b/notifiers/webhooks.go @@ -0,0 +1,121 @@ +package notifiers + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "log" + "net/http" + "net/url" + "slices" + "testing" + "time" + + "github.com/matrix-org/policyserv/internal" + "github.com/matrix-org/policyserv/storage" + "github.com/panjf2000/ants/v2" +) + +var ErrWebhookNotHttps = errors.New("webhook must be https") +var ErrWebhookDomainNotAllowed = errors.New("webhook domain not allowed") + +type WebhookMatrixNotifier struct { + MatrixNotifier + + storage storage.PersistentStorage + pool *ants.Pool + allowedDomains []string +} + +func NewWebhookMatrixNotifier(storage storage.PersistentStorage, poolSize int, allowedDomains []string) (*WebhookMatrixNotifier, error) { + pool, err := ants.NewPool(poolSize, ants.WithOptions(ants.Options{ + // Same options as the queue.Pool setup + ExpiryDuration: 1 * time.Minute, + PreAlloc: false, + MaxBlockingTasks: 0, // no limit on submissions + Nonblocking: false, + // If we don't supply a panic handler then ants will print a stack trace for us + //PanicHandler: func(err interface{}) { + // log.Println("Panic in pool:", err) + //}, + Logger: log.Default(), + DisablePurge: false, + })) + if err != nil { + return nil, err + } + return &WebhookMatrixNotifier{ + storage: storage, + pool: pool, + allowedDomains: allowedDomains, + }, nil +} + +func (w *WebhookMatrixNotifier) Send(communityId string, plainText string, htmlText string) (string, error) { + // This context only covers database calls and queue setup - it's not used for actual delivery + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + community, err := w.storage.GetCommunity(ctx, communityId) + if err != nil { + return "", err + } + target := internal.Dereference(community.Config.WebhookUrl) + if target == "" { + msgId := storage.NextId() + log.Printf("[%s] No webhook configured for community %s", msgId, communityId) + return msgId, nil + } + + whUrl, err := url.Parse(target) + if err != nil { + return "", err + } + if !testing.Testing() { + if whUrl.Scheme != "https" { + return "", ErrWebhookNotHttps + } + } + if !slices.Contains(w.allowedDomains, whUrl.Host) { + return "", ErrWebhookDomainNotAllowed + } + + msgId := storage.NextId() + err = w.pool.Submit(func() { + // We override the context here to ensure we don't spend forever trying to send a message + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + // Hookshot / Slack format body + reqBody := make(map[string]any) + reqBody["text"] = plainText + reqBody["html"] = htmlText + + buf := bytes.NewBuffer(nil) + encoder := json.NewEncoder(buf) + encoder.SetEscapeHTML(false) // we expect XSS protection from elsewhere, so avoid making the JSON hard to read by humans/tests + err := encoder.Encode(reqBody) + if err != nil { + log.Printf("[%s] Failed to encode JSON: %s", msgId, err) + return + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, whUrl.String(), buf) + if err != nil { + log.Printf("[%s] Failed to create request: %s", msgId, err) + return + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "policyserv") + + res, err := http.DefaultClient.Do(req) + if err != nil { + log.Printf("[%s] Failed to send request: %s", msgId, err) + return + } + defer res.Body.Close() + log.Printf("[%s] Webhook response: %s", msgId, res.Status) + }) + return msgId, err +} diff --git a/queue/pool_test.go b/queue/pool_test.go index 4a1a7c7..163036b 100644 --- a/queue/pool_test.go +++ b/queue/pool_test.go @@ -24,7 +24,7 @@ func TestPool(t *testing.T) { pubsub := test.NewMemoryPubsub(t) defer pubsub.Close() - manager, err := community.NewManager(cnf, db, pubsub, test.MustMakeAuditQueue(5)) + manager, err := community.NewManager(cnf, db, pubsub, test.NewMatrixNotifier(t)) assert.NoError(t, err) assert.NotNil(t, manager) @@ -79,7 +79,7 @@ func TestPoolHandlesErrors(t *testing.T) { pubsub := test.NewMemoryPubsub(t) defer pubsub.Close() - manager, err := community.NewManager(cnf, db, pubsub, test.MustMakeAuditQueue(5)) + manager, err := community.NewManager(cnf, db, pubsub, test.NewMatrixNotifier(t)) assert.NoError(t, err) assert.NotNil(t, manager) diff --git a/test/audit_queue.go b/test/audit_queue.go deleted file mode 100644 index a0d4fba..0000000 --- a/test/audit_queue.go +++ /dev/null @@ -1,13 +0,0 @@ -package test - -import ( - "github.com/matrix-org/policyserv/filter/audit" -) - -func MustMakeAuditQueue(size int) *audit.Queue { - queue, err := audit.NewQueue(size) - if err != nil { - panic(err) - } - return queue -} diff --git a/test/notifier.go b/test/notifier.go new file mode 100644 index 0000000..15e9745 --- /dev/null +++ b/test/notifier.go @@ -0,0 +1,28 @@ +package test + +import ( + "testing" + + "github.com/matrix-org/policyserv/notifiers" + "github.com/matrix-org/policyserv/storage" + "github.com/stretchr/testify/assert" +) + +type MatrixNotifier struct { + notifiers.MatrixNotifier + + t *testing.T +} + +func NewMatrixNotifier(t *testing.T) *MatrixNotifier { + return &MatrixNotifier{ + t: t, + } +} + +func (n *MatrixNotifier) Send(communityId string, plainText string, htmlText string) (string, error) { + assert.NotEmpty(n.t, communityId, "communityId is required") + assert.NotEmpty(n.t, plainText, "plainText is required") + assert.NotEmpty(n.t, htmlText, "htmlText is required") + return storage.NextId(), nil +}