Merge pull request #159 from matrix-org/travis/webhooks

Internal: Refactor how webhooks work

Author: turt2live

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)
 

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

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

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
 	}

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)
 

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,32 +57,18 @@ 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()
 
 	// Note: we log the audit context so if the webhook fails (or isn't configured) then we
 	// 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 += "</details>" // 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 <details> 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
 }

filter/audit/queue.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)
 

filter/filter_density_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)
 

filter/filter_event_type_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)