Server side filtering

client/handlers.go

@@ -61,7 +61,7 @@ type BotHandlers struct {
 
 	// Store the reactions we have previously sent by messageID,
 	// so we don't accidentally take back our own reactions.
-	reactions   map[int]map[string]interface{}
+	reactions   map[int64]map[string]interface{}
 	reactionsMu sync.Mutex
 
 	// Deadlock detection (deadlock_watch.go): record time of last successful
@@ -82,7 +82,7 @@ func (c *Client) SetupChatbot() error {
 		}),
 		autoGreet:  map[string]time.Time{},
 		messageBuf: []messages.Message{},
-		reactions:  map[int]map[string]interface{}{},
+		reactions:  map[int64]map[string]interface{}{},
 	}
 
 	// Add JavaScript support.
@@ -149,7 +149,7 @@ func (h *BotHandlers) cacheMessage(msg messages.Message) {
 }
 
 // Get a message by ID from the recent message buffer.
-func (h *BotHandlers) getMessageByID(msgID int) (messages.Message, bool) {
+func (h *BotHandlers) getMessageByID(msgID int64) (messages.Message, bool) {
 	h.messageBufMu.RLock()
 	defer h.messageBufMu.RUnlock()
 	for _, msg := range h.messageBuf {

client/rivescript_macros.go

@@ -45,7 +45,7 @@ func (h *BotHandlers) setObjectMacros() {
 					time.Sleep(2500 * time.Millisecond)
 					h.client.Send(messages.Message{
 						Action:    messages.ActionReact,
-						MessageID: msgID,
+						MessageID: int64(msgID),
 						Message:   args[1],
 					})
 				}()
@@ -77,7 +77,7 @@ func (h *BotHandlers) setObjectMacros() {
 				// Take it back.
 				h.client.Send(messages.Message{
 					Action:    messages.ActionTakeback,
-					MessageID: msgID,
+					MessageID: int64(msgID),
 				})
 			} else {
 				return fmt.Sprintf("[takeback: %s]", err)
@@ -94,7 +94,7 @@ func (h *BotHandlers) setObjectMacros() {
 				var comment = strings.Join(args[1:], " ")
 
 				// Look up this message.
-				if msg, ok := h.getMessageByID(msgID); ok {
+				if msg, ok := h.getMessageByID(int64(msgID)); ok {
 					// Report it with the custom comment.
 					h.client.Send(messages.Message{
 						Action:    messages.ActionReport,

docs/Configuration.md

@@ -42,6 +42,20 @@ PreviewImageWidth = 360
   Branding = "<em>VIP Members</em>"
   Icon = "fa fa-circle"
   MutuallySecret = false
+
+[[MessageFilters]]
+  Enabled = true
+  PublicChannels = true
+  PrivateChannels = true
+  KeywordPhrases = [
+    "\\bswear words\\b",
+    "\\b(swearing|cursing)\\b",
+    "suck my ([^\\s]+)"
+  ]
+  CensorMessage = true
+  ForwardMessage = false
+  ReportMessage = false
+  ChatServerResponse = "Watch your language."
 ```
 
 A description of the config directives includes:
@@ -87,3 +101,21 @@ If using JWT authentication, your website can mark some users as VIPs when sendi
 * **Branding** (string): HTML supported, this will appear in webcam sharing modals to "make my cam only visible to fellow VIP users"
 * **Icon** (string): icon CSS name from Font Awesome.
 * **MutuallySecret** (bool): if true, the VIP features are hidden and only visible to people who are, themselves, VIP. For example, the icon on the Who List will only show to VIP users but non-VIP will not see the icon.
+
+## Message Filters
+
+BareRTC supports optional server-side filtering of messages. These can be applied to monitor public channels, Direct Messages, or both; and provide a variety of options how you want to handle filtered messages.
+
+You can configure multiple sets of filters to treat different sets of keywords with different behaviors.
+
+Options for the `[[MessageFilters]]` section include:
+
+* **Enabled** (bool): whether to enable this filter. The default settings.toml has a filter template example by default, but it's not enabled.
+* **PublicChannels** (bool): whether to apply the filter to public channel messages.
+* **PrivateChannels** (bool): whether to apply the filter to private (Direct Message) channels.
+* **KeywordPhrases** ([]string): a listing of regular expression compatible strings to search the user's message again.
+    * Tip: use word-boundary `\b` metacharacters to detect whole words and reduce false positives from partial word matches.
+* **CensorMessage** (bool): if true, the matching keywords will be substituted with asterisks in the user's message when it appears in chat.
+* **ForwardMessage** (bool): whether to repeat the message to the other chatters. If false, the sender will see their own message echo (possibly censored) but other chatters will not get their message at all.
+* **ReportMessage** (bool): if true, report the message along with the recent context (previous 10 messages in that conversation) to your website's report webhook (if configured).
+* **ChatServerResponse** (str): optional - you can have ChatServer send a message to the sender (in the same channel) after the filter has been run. An empty string will not send a ChatServer message.

go.mod

@@ -47,4 +47,3 @@ require (
 	golang.org/x/term v0.12.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 )
-

pkg/config/config.go

@@ -13,7 +13,7 @@ import (
 
 // Version of the config format - when new fields are added, it will attempt
 // to write the settings.toml to disk so new defaults populate.
-var currentVersion = 7
+var currentVersion = 8
 
 // Config for your BareRTC app.
 type Config struct {
@@ -47,6 +47,8 @@ type Config struct {
 	WebhookURLs []WebhookURL
 
 	VIP VIP
+
+	MessageFilters []*MessageFilter
 }
 
 type TurnConfig struct {
@@ -154,6 +156,19 @@ func DefaultConfig() Config {
 			Branding: "<em>VIP Members</em>",
 			Icon:     "fa fa-circle",
 		},
+		MessageFilters: []*MessageFilter{
+			{
+				PublicChannels:  true,
+				PrivateChannels: true,
+				KeywordPhrases: []string{
+					`\bswear words\b`,
+					`\b(swearing|cursing)\b`,
+					`suck my ([^\s]+)`,
+				},
+				CensorMessage:      true,
+				ChatServerResponse: "Watch your language.",
+			},
+		},
 	}
 	c.JWT.Strict = true
 	return c

pkg/config/message_filters.go

@@ -0,0 +1,47 @@
+package config
+
+import (
+	"regexp"
+	"sync"
+
+	"git.kirsle.net/apps/barertc/pkg/log"
+)
+
+// MessageFilter configures censored or auto-flagged messages in chat.
+type MessageFilter struct {
+	Enabled            bool
+	PublicChannels     bool
+	PrivateChannels    bool
+	KeywordPhrases     []string
+	CensorMessage      bool
+	ForwardMessage     bool
+	ReportMessage      bool
+	ChatServerResponse string
+
+	// Private use variables.
+	isRegexpCompiled bool
+	regexps          []*regexp.Regexp
+	regexpMu         sync.Mutex
+}
+
+// IterPhrases returns the keyword phrases as regular expressions.
+func (mf *MessageFilter) IterPhrases() []*regexp.Regexp {
+	if mf.isRegexpCompiled {
+		return mf.regexps
+	}
+
+	// Compile and return the regexps.
+	mf.regexpMu.Lock()
+	defer mf.regexpMu.Unlock()
+	mf.regexps = []*regexp.Regexp{}
+	for _, phrase := range mf.KeywordPhrases {
+		re, err := regexp.Compile(phrase)
+		if err != nil {
+			log.Error("MessageFilter: phrase '%s' did not compile as a regexp: %s", phrase, err)
+			continue
+		}
+		mf.regexps = append(mf.regexps, re)
+	}
+
+	return mf.regexps
+}

pkg/handlers.go

@@ -159,6 +159,40 @@ func (s *Server) OnMessage(sub *Subscriber, msg messages.Message) {
 		MessageID: mid,
 	}
 
+	// Run message filters.
+	if filter, ok := s.filterMessage(sub, msg, &message); ok {
+		// What do we do with the matched filter?
+
+		// If we will not send this message out, do echo it back to
+		// the sender (possibly with censors applied).
+		if !filter.ForwardMessage {
+			s.SendTo(sub.Username, message)
+		}
+
+		// Is ChatServer to say something?
+		if filter.ChatServerResponse != "" {
+			sub.ChatServer(filter.ChatServerResponse)
+		}
+
+		// Are we to report the message to the site admin?
+		if filter.ReportMessage {
+			// If the user is OP, just tell them we would.
+			if sub.IsAdmin() {
+				sub.ChatServer("Your recent chat context would have been reported to your main website.")
+			}
+
+			// Send the report to the main website.
+			if err := s.reportFilteredMessage(sub, msg); err != nil {
+				log.Error("Reporting filtered message: %s", err)
+			}
+		}
+
+		// If we are not forwarding this message, stop here.
+		if !filter.ForwardMessage {
+			return
+		}
+	}
+
 	// Is this a DM?
 	if strings.HasPrefix(msg.Channel, "@") {
 		// Echo the message only to both parties.

pkg/message_filters.go

@@ -0,0 +1,168 @@
+package barertc
+
+import (
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"git.kirsle.net/apps/barertc/pkg/config"
+	"git.kirsle.net/apps/barertc/pkg/messages"
+)
+
+// Functionality for handling server-side message filtering and reporting.
+
+// filterMessage will check an incoming user message against the configured
+// server-side filters and react accordingly. This function also is
+// responsible for collecting the recent contexts (10 messages per channel).
+//
+// Parameters: the rawMsg is their (pre-Markdown-formatted) original message
+// (for the message context); the msg pointer is their post-formatted one, which
+// may be modified to censor their word before returning.
+//
+// Returns the matching message filter (or nil) and a boolean (matched).
+func (s *Server) filterMessage(sub *Subscriber, rawMsg messages.Message, msg *messages.Message) (*config.MessageFilter, bool) {
+	// Collect the recent channel context first.
+	if strings.HasPrefix(msg.Channel, "@") {
+		// DM
+		pushDirectMessageContext(sub, sub.Username, msg.Channel[1:], rawMsg)
+	} else {
+		// Public channel
+		pushMessageContext(sub, msg.Channel, rawMsg)
+	}
+
+	// Check it against the configured filters.
+	var matched bool
+	for _, filter := range config.Current.MessageFilters {
+		if !filter.Enabled {
+			continue
+		}
+
+		for _, phrase := range filter.IterPhrases() {
+			m := phrase.FindAllStringSubmatch(msg.Message, -1)
+			for _, match := range m {
+				// Found a match!
+				matched = true
+
+				// Censor it?
+				if filter.CensorMessage {
+					msg.Message = strings.ReplaceAll(msg.Message, match[0], strings.Repeat("*", len(match[0])))
+				}
+			}
+		}
+
+		if matched {
+			return filter, true
+		}
+	}
+
+	return nil, false
+}
+
+// Report the filtered message along with recent context.
+func (s *Server) reportFilteredMessage(sub *Subscriber, msg messages.Message) error {
+	if !WebhookEnabled(WebhookReport) {
+		return errors.New("report webhook is not enabled on this server")
+	}
+
+	// Prepare the report.
+	var context string
+	if strings.HasPrefix(msg.Channel, "@") {
+		context = getDirectMessageContext(sub.Username, msg.Channel[1:])
+	} else {
+		context = getMessageContext(msg.Channel)
+	}
+
+	if err := PostWebhook(WebhookReport, WebhookRequest{
+		Action: WebhookReport,
+		APIKey: config.Current.AdminAPIKey,
+		Report: WebhookRequestReport{
+			FromUsername:  sub.Username,
+			AboutUsername: sub.Username,
+			Channel:       msg.Channel,
+			Timestamp:     time.Now().Format(time.RFC1123),
+			Reason:        "Server Side Message Filter",
+			Message:       msg.Message,
+			Comment: fmt.Sprintf(
+				"This is an automated report via server side chat filters.\n\n"+
+					"The recent context in this channel included the following conversation:\n\n"+
+					"%s",
+				context,
+			),
+		},
+	}); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Message Context Caching
+//
+// Hold the recent (10) messages for each channel so in case of automated
+// reporting, the context can be delivered in the report.
+var (
+	messageContexts    = map[string][]string{}
+	messageContextMu   sync.RWMutex
+	messageContextSize = 10
+)
+
+// Push a message onto the recent messages context.
+func pushMessageContext(sub *Subscriber, channel string, msg messages.Message) {
+	messageContextMu.Lock()
+	defer messageContextMu.Unlock()
+
+	// Initialize the context for new channel the first time.
+	if _, ok := messageContexts[channel]; !ok {
+		messageContexts[channel] = []string{}
+	}
+
+	// Append this message to it.
+	messageContexts[channel] = append(messageContexts[channel], fmt.Sprintf(
+		"%s [%s] %s",
+		time.Now().Format("2006-01-02 15:04:05"),
+		sub.Username,
+		strings.TrimSpace(msg.Message),
+	))
+
+	fmt.Printf("Context %s:\n%+v\n", channel, messageContexts[channel])
+
+	// Trim the context to recent messages only.
+	if len(messageContexts[channel]) > messageContextSize {
+		messageContexts[channel] = messageContexts[channel][len(messageContexts[channel])-messageContextSize:]
+	}
+}
+
+// Push a message context for DMs. A channel name will be derived consistently
+// based on the sorted pair of usernames.
+func pushDirectMessageContext(sub *Subscriber, username1, username2 string, msg messages.Message) {
+	var names = []string{username1, username2}
+	sort.Strings(names)
+	pushMessageContext(
+		sub,
+		fmt.Sprintf("@%s", strings.Join(names, ":")),
+		msg,
+	)
+}
+
+// Get the recent message context, pretty printed.
+func getMessageContext(channel string) string {
+	messageContextMu.RLock()
+	defer messageContextMu.RUnlock()
+
+	if _, ok := messageContexts[channel]; !ok {
+		return "(No recent message history in this channel)"
+	}
+
+	return strings.Join(messageContexts[channel], "\n\n")
+}
+
+func getDirectMessageContext(username1, username2 string) string {
+	var names = []string{username1, username2}
+	sort.Strings(names)
+	return getMessageContext(
+		fmt.Sprintf("@%s", strings.Join(names, ":")),
+	)
+}

pkg/messages/messages.go

@@ -1,15 +1,18 @@
 package messages
 
-import "sync"
+import (
+	"sync"
+	"time"
+)
 
 // Auto incrementing Message ID for anything pushed out by the server.
 var (
-	messageID int
+	messageID = time.Now().Unix()
 	mu        sync.Mutex
 )
 
 // NextMessageID atomically increments and returns a new MessageID.
-func NextMessageID() int {
+func NextMessageID() int64 {
 	mu.Lock()
 	defer mu.Unlock()
 	messageID++
@@ -42,7 +45,7 @@ type Message struct {
 	DND         bool   `json:"dnd,omitempty"`    // Do Not Disturb, e.g. DMs are closed
 
 	// Message ID to support takebacks/local deletions
-	MessageID int `json:"msgID,omitempty"`
+	MessageID int64 `json:"msgID,omitempty"`
 
 	// Sent on `open` actions along with the (other) Username.
 	OpenSecret string `json:"openSecret,omitempty"`

pkg/websocket.go

@@ -43,7 +43,7 @@ type Subscriber struct {
 
 	// Record which message IDs belong to this user.
 	midMu      sync.Mutex
-	messageIDs map[int]struct{}
+	messageIDs map[int64]struct{}
 }
 
 // ReadLoop spawns a goroutine that reads from the websocket connection.
@@ -141,7 +141,16 @@ func (sub *Subscriber) SendJSON(v interface{}) error {
 		return err
 	}
 	log.Debug("SendJSON(%d=%s): %s", sub.ID, sub.Username, data)
-	return sub.conn.Write(sub.ctx, websocket.MessageText, data)
+
+	// Add the message to the recipient's queue. If the queue is too full,
+	// disconnect the client as they can't keep up.
+	select {
+	case sub.messages <- data:
+	default:
+		go sub.closeSlow()
+	}
+
+	return nil
 }
 
 // SendMe sends the current user state to the client.
@@ -197,7 +206,7 @@ func (s *Server) WebSocket() http.HandlerFunc {
 			booted:     make(map[string]struct{}),
 			muted:      make(map[string]struct{}),
 			blocked:    make(map[string]struct{}),
-			messageIDs: make(map[int]struct{}),
+			messageIDs: make(map[int64]struct{}),
 			ChatStatus: "online",
 		}
 

src/App.vue

@@ -849,7 +849,7 @@ export default {
         },
 
         sendTypingNotification() {
-            // TODO
+            // Send typing indicator for DM threads.
         },
 
         // Emoji reactions