Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 99 additions & 0 deletions agent/cmd/chat_moderation_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
package main

import (
"encoding/json"
"net/http"
"strings"
"testing"

"github.com/muxable/rtchat/agent/internal/observability"
"go.uber.org/zap"
"go.uber.org/zap/zaptest/observer"
)

func TestTwitchModerationRejectionReporting(t *testing.T) {
for _, tc := range []struct {
name, action, message string
providerStatus, wantStatus int
wantCode, wantMessage string
}{
{"already banned", "ban", "user is already banned", 400, 400, "FAILED_PRECONDITION", "This user is already banned."},
{"banned user timeout", "timeout", "user is already banned", 400, 400, "FAILED_PRECONDITION", "This user is already banned."},
{"protected user", "ban", "user may not be banned", 400, 400, "FAILED_PRECONDITION", "Twitch does not allow banning this user."},
{"protected timeout", "timeout", "user may not be put in a timeout", 400, 400, "FAILED_PRECONDITION", "Twitch does not allow timing out this user."},
{"already unbanned", "unban", "user is not banned", 400, 400, "FAILED_PRECONDITION", "This user is not banned."},
{"documented wording", "ban", "The user specified in the user_id field is already banned.", 400, 400, "FAILED_PRECONDITION", "This user is already banned."},
{"missing provider parameter", "ban", "user_id is required", 400, 500, "INTERNAL", "INTERNAL"},
{"unexpected response", "ban", "private-provider-detail", 400, 500, "INTERNAL", "INTERNAL"},
{"not a matching reason", "ban", "user is already banned private-provider-detail", 400, 500, "INTERNAL", "INTERNAL"},
{"forbidden", "ban", "private-provider-detail", 403, 500, "INTERNAL", "INTERNAL"},
{"server failure with known text", "ban", "user is already banned", 503, 500, "INTERNAL", "INTERNAL"},
{"rate limited", "ban", "private-provider-detail", 429, 429, "RESOURCE_EXHAUSTED", "too many requests, try again shortly"},
} {
t.Run(tc.name, func(t *testing.T) {
core, logs := observer.New(zap.InfoLevel)
defer zap.ReplaceGlobals(zap.New(observability.Wrap(core)))()
requests := 0
s, _ := newMock(func(r *http.Request) (*http.Response, error) {
requests++
if r.URL.Path != "/helix/moderation/bans" {
t.Fatalf("unexpected endpoint: %s", r.URL.Path)
}
payload, err := json.Marshal(map[string]any{"message": tc.message, "access_token": "private-provider-detail"})
if err != nil {
t.Fatal(err)
}
return respond(tc.providerStatus, string(payload)), nil
})
code, body := call(t, s, tc.action, map[string]any{"provider": "twitch", "channelId": "broadcaster", "userId": "target", "length": 60, "reason": "moderation"}, true)
e := object(body["error"])
if requests != 1 || code != tc.wantStatus || e["status"] != tc.wantCode || e["message"] != tc.wantMessage {
t.Fatalf("requests=%d code=%d response=%v", requests, code, body)
}
entries := logs.FilterMessage("http request").All()
if len(entries) != 1 {
t.Fatal(entries)
}
fields := entries[0].ContextMap()
if fields["status"] != int64(tc.wantStatus) {
t.Fatal(fields)
}
if tc.wantStatus < 500 && (entries[0].Level != zap.InfoLevel || fields["@type"] != nil) {
t.Fatalf("user rejection reported as an agent issue: %v", fields)
}
if tc.wantStatus >= 500 && fields["@type"] == nil {
t.Fatalf("unexpected failure no longer reported: %v", fields)
}
logged, err := json.Marshal(fields)
if err != nil {
t.Fatal(err)
}
encoded, err := json.Marshal(body)
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(logged)+string(encoded), "private-provider-detail") {
t.Fatal("provider details leaked")
}
})
}
}

func TestTwitchModerationMissingUserDoesNotSendInvalidBan(t *testing.T) {
for _, action := range []string{"ban", "unban", "timeout"} {
t.Run(action, func(t *testing.T) {
requests := 0
s, _ := newMock(func(r *http.Request) (*http.Response, error) {
requests++
if r.URL.Path != "/helix/users" {
t.Fatalf("moderation request for a missing user: %s", r.URL.Path)
}
return respond(200, `{"data":[]}`), nil
})
code, body := call(t, s, action, map[string]any{"provider": "twitch", "channelId": "broadcaster", "username": "missing", "length": 60, "reason": "moderation"}, true)
if code != 404 || object(body["error"])["status"] != "NOT_FOUND" || requests != 1 {
t.Fatal(code, body, requests)
}
})
}
}
7 changes: 7 additions & 0 deletions agent/cmd/chat_twitch.go
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,9 @@ func (s *chatRoutes) twitch(ctx context.Context, uid, token, action string, r ch
if err != nil {
return nil, err
}
if user.ID == "" {
return nil, &callable.Error{Code: "not-found", Message: "Twitch user not found"}
}
target = user.ID
}
switch action {
Expand Down Expand Up @@ -116,5 +119,9 @@ func (s *chatRoutes) twitch(ctx context.Context, uid, token, action string, r ch
zap.L().Warn("failed to record raid cancel")
}
}
var rejected *twitch.ModerationRejected
if errors.As(err, &rejected) {
return nil, &callable.Error{Code: "failed-precondition", Message: rejected.Message}
}
return nil, err
}
36 changes: 34 additions & 2 deletions agent/internal/clients/twitch/chat.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package twitch
import (
"context"
"net/url"
"strings"
"time"
)

Expand Down Expand Up @@ -66,11 +67,42 @@ func (c *Client) Ban(ctx context.Context, channel, moderator, user string, durat
if duration != 0 {
data["duration"], data["reason"] = duration, reason
}
return c.write(ctx, "POST", moderationPath("/moderation/bans", channel, moderator, nil), map[string]any{"data": data})
return c.writeBan(ctx, "POST", moderationPath("/moderation/bans", channel, moderator, nil), map[string]any{"data": data})
}

func (c *Client) Unban(ctx context.Context, channel, moderator, user string) error {
return c.write(ctx, "DELETE", moderationPath("/moderation/bans", channel, moderator, url.Values{"user_id": {user}}), nil)
return c.writeBan(ctx, "DELETE", moderationPath("/moderation/bans", channel, moderator, url.Values{"user_id": {user}}), nil)
}

// ModerationRejected is an expected refusal based on the target's ban state.
// Other provider failures retain their redacted HTTP error for reporting.
type ModerationRejected struct{ Message string }

func (e *ModerationRejected) Error() string { return e.Message }

func (c *Client) writeBan(ctx context.Context, method, path string, body any) error {
response, err := c.Do(ctx, method, path, body)
if err != nil {
return err
}
if response.StatusCode == 400 {
var result struct{ Message string }
if response.Decode(&result) == nil {
// Recognize only ban-state rejections; preserve other errors so
// malformed provider requests still trigger diagnostics.
switch strings.TrimSuffix(strings.ToLower(strings.TrimSpace(result.Message)), ".") {
case "user is already banned", "the user specified in the user_id field is already banned":
return &ModerationRejected{Message: "This user is already banned."}
case "user is not banned", "the user specified in the user_id query parameter is not banned":
return &ModerationRejected{Message: "This user is not banned."}
case "user may not be banned", "the user specified in the user_id field may not be banned":
return &ModerationRejected{Message: "Twitch does not allow banning this user."}
case "user may not be put in a timeout", "the user specified in the user_id field may not be put in a timeout":
return &ModerationRejected{Message: "Twitch does not allow timing out this user."}
}
}
}
return response.Err()
}

func (c *Client) DeleteMessages(ctx context.Context, channel, moderator, message string) error {
Expand Down
6 changes: 5 additions & 1 deletion agent/internal/handler/join_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -223,12 +223,16 @@ func TestAnonymousJoinDoesNotUseProfilesOrTokens(t *testing.T) {
func TestJoinWaitsForRequestedRoomState(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
releaseConnection := make(chan struct{})
defer close(releaseConnection)
c := &fakeChat{}
c.connect = func() error {
c.onConnect()
c.onRoomState(twitch.RoomStateMessage{Channel: "another-room"})
cancel()
<-ctx.Done()
// Keep the connection open so cancellation, rather than its exit,
// ends the join. Otherwise both select cases can become ready at once.
<-releaseConnection
return twitch.ErrClientDisconnected
}
if err := joinWithContext(ctx, c, "room", time.Second); !errors.Is(err, context.Canceled) {
Expand Down
Loading