diff --git a/agent/cmd/chat_moderation_test.go b/agent/cmd/chat_moderation_test.go new file mode 100644 index 000000000..9fb7d84b9 --- /dev/null +++ b/agent/cmd/chat_moderation_test.go @@ -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) + } + }) + } +} diff --git a/agent/cmd/chat_twitch.go b/agent/cmd/chat_twitch.go index 6ccc274af..bc1a3996f 100644 --- a/agent/cmd/chat_twitch.go +++ b/agent/cmd/chat_twitch.go @@ -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 { @@ -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 } diff --git a/agent/internal/clients/twitch/chat.go b/agent/internal/clients/twitch/chat.go index 3d5fef566..62aa0ba79 100644 --- a/agent/internal/clients/twitch/chat.go +++ b/agent/internal/clients/twitch/chat.go @@ -3,6 +3,7 @@ package twitch import ( "context" "net/url" + "strings" "time" ) @@ -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 { diff --git a/agent/internal/handler/join_test.go b/agent/internal/handler/join_test.go index bb316bce2..a14a2e1f0 100644 --- a/agent/internal/handler/join_test.go +++ b/agent/internal/handler/join_test.go @@ -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) {