Files
velox-bot/internal/security/dm_test.go
T
FernandoJVideiraandClaude Opus 5.5 3f87e68a2c Add anti-raid and anti-nuke protection
Join Gate (account age, avatar, advertising and username filters, bot
checks), raid detection with invite pausing, anti-nuke quarantine of
admins who mass delete, manual and automatic lockdowns, a decaying heat
score against spam, role and channel snapshots with restore, and alert
DMs to the owner and trusted users. Everything that weakens protection
is limited to the owner and the trusted list.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-20 21:59:52 +01:00

93 lines
2.6 KiB
Go

package security
import (
"fmt"
"strings"
"testing"
"velox-bot/internal/db/repos/securityrepo"
"github.com/bwmarrin/discordgo"
)
func nuke(e *Engine) {
e.OnAuditLogEntry(ctx, guild, entry(discordgo.AuditLogActionMemberPrune, "50", ""))
}
func TestCriticalAlertsAreDMed(t *testing.T) {
e, d, st := nukeEngine(t)
d.guildName = "Velox **HQ** @everyone"
// Owner (1) is also on the list as a user; a trusted role must not be expanded.
st.trusted[100] = []securityrepo.Trusted{{TargetID: 1, Kind: "user"}, {TargetID: 60, Kind: "user"}, {TargetID: 201, Kind: "role"}, {TargetID: 999, Kind: "user"}}
refresh(e)
nuke(e)
got := strings.Join(d.dmsTo(), ",")
if got != "1,60" {
t.Fatalf("DMs went to %q, want the owner and the trusted user once each (no role members, not the bot)", got)
}
var dm *discordgo.MessageSend
for i, ch := range d.sentTo {
if ch == "dm-1" {
dm = d.sent[i]
}
}
if !strings.Contains(dm.Embeds[0].Title, "Nuke attempt stopped") {
t.Errorf("DM should carry the alert: %+v", dm.Embeds[0])
}
if strings.Contains(dm.Content, "@everyone") || strings.Contains(dm.Content, "**HQ**") || dm.AllowedMentions == nil {
t.Errorf("the server name must be escaped and nothing pinged: %q", dm.Content)
}
if len(dm.Components) != 0 {
t.Error("no action buttons in DMs")
}
}
func TestDMsWorkWithoutAnAlertChannel(t *testing.T) {
e, d, st := nukeEngine(t)
st.set(100).AlertChannelID = 0
refresh(e)
nuke(e)
if strings.Join(d.dmsTo(), ",") != "1" {
t.Errorf("the owner should still be told: %v", d.dmsTo())
}
}
func TestDMsCanBeTurnedOff(t *testing.T) {
e, d, st := nukeEngine(t)
st.set(100).AlertDM = false
refresh(e)
nuke(e)
if len(d.dmsTo()) != 0 {
t.Errorf("DMs are off: %v", d.dmsTo())
}
}
func TestRoutineAlertsAreNotDMed(t *testing.T) {
e, d, st, now := testEngine(t)
s := st.set(100)
s.JoinGateEnabled, s.NoAvatar = true, securityrepo.ActionKick
refresh(e)
e.OnMemberJoin(ctx, guild, &discordgo.Member{User: &discordgo.User{ID: snowflakeAt(now.AddDate(-1, 0, 0), 1)}})
if len(d.dmsTo()) != 0 {
t.Error("Join Gate actions only go to the alert channel")
}
}
func TestClosedDMsAreSkippedAndListIsCapped(t *testing.T) {
e, d, st := nukeEngine(t)
d.closedDMs = map[string]bool{"1": true}
for i := 0; i < 15; i++ {
st.trusted[100] = append(st.trusted[100], securityrepo.Trusted{TargetID: int64(700 + i), Kind: "user"})
}
refresh(e)
nuke(e)
got := d.dmsTo()
if len(got) != maxAlertDMs-1 {
t.Errorf("owner's DMs are closed and the list is capped at %d: got %d (%v)", maxAlertDMs, len(got), got)
}
if got[0] != fmt.Sprint(700) {
t.Errorf("the others still get it: %v", got)
}
}