From ce528843f48c19b21ca217122a7de7d1a3a9afe9 Mon Sep 17 00:00:00 2001 From: Alexey Grigorev Date: Mon, 12 Oct 2026 15:43:13 +0200 Subject: [PATCH] Record deletes and auto-wipe cross-channel copy-paste spam Log every delete/ban with the triggering reaction, admin, and message text (Dynamo 90-day TTL + S3 archive). Route user message events to the automator. Similar text in two or more channels is folded into a Dynamo aggregate; a 5-minute EventBridge scan deletes quiet sessions and DMs Alexey the same deactivate flow as :ban-user: (summary, then the user id as a separate message). --- .grok/skills/delete-audit/SKILL.md | 54 +++ README.md | 15 +- automator/README.md | 9 + automator/scripts/query_delete_audit.py | 112 +++++++ automator/src/audit.py | 204 ++++++++++++ automator/src/config.yaml | 33 ++ automator/src/lambda_function.py | 400 +++++++++++++++++++---- automator/src/spam.py | 257 +++++++++++++++ automator/tests/test_audit.py | 155 +++++++++ automator/tests/test_automator_lambda.py | 222 +++++++++++++ automator/tests/test_reactions_e2e.py | 2 + automator/tests/test_spam.py | 170 ++++++++++ docs/delete-audit.md | 89 +++++ docs/logs.md | 9 + router/src/lambda_function.py | 26 ++ router/tests/test_router_lambda.py | 36 +- 16 files changed, 1725 insertions(+), 68 deletions(-) create mode 100644 .grok/skills/delete-audit/SKILL.md create mode 100644 automator/scripts/query_delete_audit.py create mode 100644 automator/src/audit.py create mode 100644 automator/src/spam.py create mode 100644 automator/tests/test_audit.py create mode 100644 automator/tests/test_spam.py create mode 100644 docs/delete-audit.md diff --git a/.grok/skills/delete-audit/SKILL.md b/.grok/skills/delete-audit/SKILL.md new file mode 100644 index 0000000..646ce39 --- /dev/null +++ b/.grok/skills/delete-audit/SKILL.md @@ -0,0 +1,54 @@ +--- +name: delete-audit +description: > + Query Slack automator delete/ban audit records (who triggered a wipe, which + reaction, which messages). Use when the user asks what was deleted, who was + banned, delete-audit history, or runs /delete-audit. +--- + +# Delete-audit query + +The automator records every delete (`DELETE_MESSAGE`, `BAN_USER`, +`REPOST_TO_THREAD_AND_DELETE`, `REMOVE_BROADCAST`, `CROSS_CHANNEL_SPAM`) with +the trigger, target user, and message text. + +Schema, storage, and cost: `docs/delete-audit.md` in +`/home/alexey/git/au-tomator-lambda`. Do not restate it. + +## Credentials + +Default AWS Gate vends **sandbox** `817685572750` only. The table and bucket +are in **main** `387546586013`. Sandbox and `cmp-alert-investigator` cannot +`dynamodb:Query` this table. + +Need main-account operator credentials (Alexey applying `main/automator`, or a +role that can read the table and bucket). If Gate is closed or the caller is +sandbox-only, say so and stop; do not guess records. + +## Query + +From `/home/alexey/git/au-tomator-lambda/automator`: + +```bash +uv run python scripts/query_delete_audit.py --user U123 +uv run python scripts/query_delete_audit.py --date YYYY-MM-DD +uv run python scripts/query_delete_audit.py --user U123 --s3 +``` + +`--user` is the **target** Slack id (the person whose messages were removed), +not the admin who reacted. `--s3` is the forever copy after the 90-day Dynamo +TTL. + +If the script is missing, use: + +- Dynamo `automator-delete-audit` `eu-west-1` + - `pk = USER#` + - GSI `gsi1` `gsi1pk = DATE#YYYY-MM-DD` +- S3 `dtc-automator-delete-audit-387546586013` prefixes `by-user//` and + `by-date/YYYY/MM/DD/` + +## How to report + +Show `recorded_at`, `action_type`, `reaction`, `triggered_by`, +`target_user_id`, counts, and message text. For a ban/spam wipe, list each +deleted message (channel, ts, kind, text). Do not dump emails unless asked. diff --git a/README.md b/README.md index 3f09c45..74adef3 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,7 @@ Its only job is to look at the incoming event and quickly decide where it should |-------|-----------| | **Reaction added by admin** | Automator | | **App mention** | Automator / FAQ Assistant | -| **Message event** | Disabled for now | +| **Message event** (user posts, not bots/edits) | Automator spam tracker | | **Button click** (e.g. from an alert) | Acknowledged only | The router and automator are separate on purpose: Slack expects a response within about 3 seconds. The router responds immediately and invokes the automator asynchronously, so longer work doesn’t time out. @@ -65,6 +65,13 @@ Examples: The automator can also send a relevant reply to the author of the message you reacted to, so they get clear feedback. +If someone posts the **same or very similar message in multiple channels**, the +automator folds those copies into a DynamoDB aggregate, waits until they go +quiet (~3 minutes, scanned every 5 minutes), deletes them, and DMs Alexey the +same deactivate workflow as `:ban-user:` (summary, then the user id as a +separate message). Config: `spam_tracker` in `automator/src/config.yaml`. +Slack must send `message` events to the router (`message.channels`). + --- ## Deployment @@ -146,6 +153,11 @@ Each entry defines an emoji (reaction name) and what the automator should do whe | `FAQ_ASSISTANT` | Calls the Cloudflare FAQ assistant, then posts the reply in the thread. | | `REMOVE_BROADCAST` | Removes a “also sent to channel” thread reply from the channel (keeps it in the thread). | | `REPOST_TO_THREAD_AND_DELETE` | Reposts the message to the thread with a custom message, then deletes it from the channel. | +| `BAN_USER` | Deletes the user's recent messages (and thread replies), DMs innocent participants, and DMs the moderator a cleanup summary. | + +Every delete (`DELETE_MESSAGE`, `BAN_USER`, `REPOST_TO_THREAD_AND_DELETE`, +`REMOVE_BROADCAST`) is recorded with the triggering reaction, the admin who +reacted, and the deleted message text. See [docs/delete-audit.md](docs/delete-audit.md). #### Multiple handlers @@ -180,6 +192,7 @@ If a placeholder is channel-specific and the channel isn’t in the map, the `de | `shameless-rules` | DELETE_MESSAGE | Enforce shameless-promo rules; DM author. | | `jobs-rules` | DELETE_MESSAGE | Enforce job-posting rules; DM author. | | `ask-ai` | ASK_AI | Generate and post an AI reply in the thread. | +| `ban-user` | BAN_USER | Wipe the author's recent messages and DM the moderator a cleanup summary. | To change behavior, edit `automator/src/config.yaml` and keep the same structure so the application stays compatible. diff --git a/automator/README.md b/automator/README.md index 2f682b7..0f7b337 100644 --- a/automator/README.md +++ b/automator/README.md @@ -14,6 +14,15 @@ mentions. The router invokes it asynchronously: See the top-level [README](../README.md) for the configuration reference (`src/config.yaml`) and the FAQ assistant setup. +Delete / ban-user actions are recorded (trigger, target user, message text) +in DynamoDB + S3. See [docs/delete-audit.md](../docs/delete-audit.md). Query +them with `/delete-audit`. + +Cross-channel copy-paste spam is pre-aggregated in DynamoDB (`spam.py`). A +5-minute EventBridge scan deletes quiet sessions and DMs Alexey the same +deactivate flow as `:ban-user:`. Requires `message.channels` on the Slack app +and the Dynamo/EventBridge resources in `aws-infra/main/automator`. + ## Layout ``` diff --git a/automator/scripts/query_delete_audit.py b/automator/scripts/query_delete_audit.py new file mode 100644 index 0000000..2a43f5a --- /dev/null +++ b/automator/scripts/query_delete_audit.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +"""Query recorded Slack delete actions. + +Examples: + python scripts/query_delete_audit.py --user U123 + python scripts/query_delete_audit.py --date 2026-04-08 + python scripts/query_delete_audit.py --user U123 --s3 +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys + +import boto3 + + +TABLE = os.getenv('AUDIT_TABLE', 'automator-delete-audit') +BUCKET = os.getenv('AUDIT_BUCKET', 'dtc-automator-delete-audit-387546586013') +REGION = os.getenv('AWS_REGION', 'eu-west-1') + + +def _print(records): + print(json.dumps(records, indent=2, default=str)) + print(f'{len(records)} record(s)', file=sys.stderr) + + +def query_user(user_id): + dynamodb = boto3.resource('dynamodb', region_name=REGION) + table = dynamodb.Table(TABLE) + items = [] + kwargs = { + 'KeyConditionExpression': boto3.dynamodb.conditions.Key('pk').eq( + f'USER#{user_id}' + ) + } + while True: + resp = table.query(**kwargs) + items.extend(resp.get('Items', [])) + if 'LastEvaluatedKey' not in resp: + break + kwargs['ExclusiveStartKey'] = resp['LastEvaluatedKey'] + return items + + +def query_date(day): + dynamodb = boto3.resource('dynamodb', region_name=REGION) + table = dynamodb.Table(TABLE) + items = [] + kwargs = { + 'IndexName': 'gsi1', + 'KeyConditionExpression': boto3.dynamodb.conditions.Key('gsi1pk').eq( + f'DATE#{day}' + ), + } + while True: + resp = table.query(**kwargs) + items.extend(resp.get('Items', [])) + if 'LastEvaluatedKey' not in resp: + break + kwargs['ExclusiveStartKey'] = resp['LastEvaluatedKey'] + return items + + +def list_s3(user_id=None, day=None): + s3 = boto3.client('s3', region_name=REGION) + if user_id: + prefix = f'by-user/{user_id}/' + if day: + year, month, rest = day.split('-', 2) + prefix = f'by-user/{user_id}/{year}/{month}/{rest}/' + elif day: + year, month, rest = day.split('-', 2) + prefix = f'by-date/{year}/{month}/{rest}/' + else: + raise SystemExit('--s3 requires --user and/or --date') + + records = [] + paginator = s3.get_paginator('list_objects_v2') + for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix): + for obj in page.get('Contents') or []: + body = s3.get_object(Bucket=BUCKET, Key=obj['Key'])['Body'].read() + records.append(json.loads(body)) + return records + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--user', help='Slack user id the action targeted') + parser.add_argument('--date', help='UTC calendar day YYYY-MM-DD') + parser.add_argument( + '--s3', action='store_true', + help='Read the forever S3 copy instead of DynamoDB', + ) + args = parser.parse_args() + + if args.s3: + _print(list_s3(user_id=args.user, day=args.date)) + return + if args.user: + _print(query_user(args.user)) + return + if args.date: + _print(query_date(args.date)) + return + parser.error('pass --user, --date, or --s3 with one of those') + + +if __name__ == '__main__': + main() diff --git a/automator/src/audit.py b/automator/src/audit.py new file mode 100644 index 0000000..b0c867f --- /dev/null +++ b/automator/src/audit.py @@ -0,0 +1,204 @@ +"""Record Slack delete actions for later review. + +Each delete action (single-message delete, thread wipe, broadcast removal, +or ban-user cleanup) is stored three ways: + +1. A structured CloudWatch log line (`DELETE_AUDIT ...`) — already paying + for the Lambda log group; useful for the last couple of weeks. +2. DynamoDB (`automator-delete-audit`) with a TTL — query by target user or + by date while the item is hot. +3. S3 (`dtc-automator-delete-audit-`) — the cheap forever copy. + Dual-written at event time so there is no extra dump Lambda. + +DynamoDB is PAY_PER_REQUEST and S3 Standard. Glacier / Standard-IA would +cost more here: they round each object up to 128 KB, and these records are +a few KB. Leave the objects in Standard. + +Writes never raise to the caller. A failed audit must not block a delete. +""" + +import json +import os +import time +from datetime import datetime, timezone + +from logs import logger + + +AUDIT_TABLE = os.getenv('AUDIT_TABLE', 'automator-delete-audit') +AUDIT_BUCKET = os.getenv( + 'AUDIT_BUCKET', 'dtc-automator-delete-audit-387546586013' +) +AUDIT_TTL_DAYS = int(os.getenv('AUDIT_TTL_DAYS', '90')) +TEXT_LIMIT = 4000 + + +def _utcnow(): + return datetime.now(timezone.utc) + + +def _clip(text): + text = text or '' + if len(text) <= TEXT_LIMIT: + return text + return text[:TEXT_LIMIT] + '…' + + +def _drop_none(value): + if isinstance(value, dict): + return {k: _drop_none(v) for k, v in value.items() if v is not None} + if isinstance(value, list): + return [_drop_none(v) for v in value] + return value + + +def _action_id(event): + item = event.get('item') or {} + parts = [ + item.get('channel') or 'no-channel', + item.get('ts') or 'no-ts', + event.get('reaction') or 'no-reaction', + event.get('event_ts') or event.get('ts') or 'no-event-ts', + ] + return ':'.join(str(p) for p in parts) + + +def build_audit_record( + event, + action_type, + deleted_messages, + *, + target_user_id=None, + target_user=None, + channel_name=None, + extra=None, + fake_delete=False, + now=None, +): + """Build the audit document (no I/O).""" + now = now or _utcnow() + item = event.get('item') or {} + recorded_at = now.strftime('%Y-%m-%dT%H:%M:%SZ') + day = now.strftime('%Y-%m-%d') + action_id = _action_id(event) + + messages = [] + for msg in deleted_messages or []: + messages.append({ + 'channel': msg.get('channel'), + 'ts': msg.get('ts'), + 'user': msg.get('user'), + 'text': _clip(msg.get('text')), + 'kind': msg.get('kind') or 'parent', + 'ok': bool(msg.get('ok', True)), + 'fake': bool(msg.get('fake', fake_delete)), + }) + + target = target_user_id or (target_user or {}).get('id') + record = { + 'action_id': action_id, + 'recorded_at': recorded_at, + 'action_type': action_type, + 'reaction': event.get('reaction'), + 'triggered_by': event.get('user'), + 'event_ts': event.get('event_ts') or event.get('ts'), + 'channel': item.get('channel'), + 'channel_name': channel_name, + 'message_ts': item.get('ts'), + 'item_user': event.get('item_user'), + 'target_user_id': target, + 'target_user': target_user, + 'fake_delete': bool(fake_delete), + 'deleted_count': len(messages), + 'deleted_ok_count': sum(1 for m in messages if m['ok']), + 'deleted_messages': messages, + 'extra': extra or {}, + 'pk': f"USER#{target or 'unknown'}", + 'sk': f"{recorded_at}#{action_id}", + 'gsi1pk': f"DATE#{day}", + 'gsi1sk': f"{recorded_at}#{action_id}", + 'expires_at': int(now.timestamp()) + AUDIT_TTL_DAYS * 86400, + } + return _drop_none(record) + + +def s3_keys(record): + recorded_at = record['recorded_at'] + year, month, day = recorded_at[:10].split('-') + action_id = record['action_id'].replace('/', '_') + keys = [f"by-date/{year}/{month}/{day}/{action_id}.json"] + user_id = record.get('target_user_id') + if user_id: + keys.append( + f"by-user/{user_id}/{year}/{month}/{day}/{action_id}.json" + ) + return keys + + +def _put_dynamo(item): + import boto3 + + table = boto3.resource('dynamodb').Table(AUDIT_TABLE) + table.put_item(Item=item) + + +def _put_s3(key, body): + import boto3 + + boto3.client('s3').put_object( + Bucket=AUDIT_BUCKET, + Key=key, + Body=body.encode('utf-8'), + ContentType='application/json', + ) + + +def _aws_enabled(): + return os.getenv('AUDIT_ENABLED', '1') == '1' + + +def record_delete_action( + event, + action_type, + deleted_messages, + *, + target_user_id=None, + target_user=None, + channel_name=None, + extra=None, + fake_delete=False, +): + """Log the action and persist it to DynamoDB + S3. Never raises.""" + try: + record = build_audit_record( + event, + action_type, + deleted_messages, + target_user_id=target_user_id, + target_user=target_user, + channel_name=channel_name, + extra=extra, + fake_delete=fake_delete, + ) + except Exception as exc: + logger.info(f"DELETE_AUDIT build failed: {exc}") + return None + + logger.info('DELETE_AUDIT ' + json.dumps(record, default=str)) + + if not _aws_enabled(): + return record + + try: + _put_dynamo(record) + except Exception as exc: + logger.info(f"DELETE_AUDIT dynamodb write failed: {exc}") + + body = json.dumps(record, default=str, indent=2) + for key in s3_keys(record): + try: + _put_s3(key, body) + except Exception as exc: + logger.info(f"DELETE_AUDIT s3 write failed key={key}: {exc}") + + return record diff --git a/automator/src/config.yaml b/automator/src/config.yaml index e815170..b9d5c17 100644 --- a/automator/src/config.yaml +++ b/automator/src/config.yaml @@ -4,6 +4,39 @@ admins: workspace: admin_users_url: https://datatalks-club.slack.com/admin +# Cross-channel copy-paste spam: wait until they go quiet, then wipe and DM +# admins the same way :ban-user: does (summary + user id as a separate message). +spam_tracker: + enabled: true + quiet_seconds: 180 + min_channels: 2 + min_chars: 40 + similarity: 0.88 + lookback_hours: 24 + admin_users: + - U01AXE0P5M3 + admin_message: | + 🚨 Cross-channel spam cleanup + + *User:* <@{target_user}> + *User ID:* `{target_user_id}` + *Display name:* {display_name} + *Real name:* {real_name} + *Email:* {email} + *Profile last updated:* {updated} + + *Channels:* {channel_count} — {channel_list} + *Sample:* + > {sample_text} + + *Messages deleted:* {deleted_count} + *Thread replies deleted:* {thread_deleted_count} + + *Next step — deactivate manually:* + 1. Open <{admin_url}|admin members page> + 2. Search for this exact user ID: `{target_user_id}` + 3. Open the matching member and deactivate. + channels: C02R98X7DS9: "course-mlops-zoomcamp" C01FABYF2RG: "course-data-engineering" diff --git a/automator/src/lambda_function.py b/automator/src/lambda_function.py index 031e022..c4a2487 100644 --- a/automator/src/lambda_function.py +++ b/automator/src/lambda_function.py @@ -10,6 +10,8 @@ import util import groqu import slack +import audit +import spam from logs import logger @@ -265,11 +267,17 @@ def handle_remove_broadcast(event, reaction_config): if is_broadcasted_reply: # Remove the broadcasted message from the channel # (it will still remain in the thread) - if FAKE_DELETE: - logger.info(f"FAKE_DELETE broadcasted message for {channel} {ts}") - else: - slack.remove_message(channel, ts) + ok = _delete_message(channel, ts) logger.info(f"Removed broadcasted message from channel {channel} (kept in thread)") + _record_delete_action( + event, + 'REMOVE_BROADCAST', + [_deleted_entry( + channel, ts, message_details.get('user'), + message_details.get('text', ''), 'broadcast', ok, + )], + target_user_id=message_details.get('user'), + ) else: logger.info(f"Message {ts} is not a broadcasted reply, skipping removal") @@ -287,14 +295,16 @@ def handle_delete_message(event, reaction_config): return user = message_details.get('user') - original_message = message_details.get('text', '') + original_message_raw = message_details.get('text', '') # Escape curly braces in user message to avoid format string issues - original_message = original_message.replace('{', '{{').replace('}', '}}') + original_message = original_message_raw.replace('{', '{{').replace('}', '}}') # Skip if there's no user (e.g., bot message or deleted message) if not user: logger.info(f"Message has no user for {channel} {ts}") return + + deleted_messages = [] # Handle thread messages if thread_message is configured thread_message_pattern = reaction_config.get('thread_message') @@ -322,11 +332,11 @@ def handle_delete_message(event, reaction_config): if thread_dm: slack.send_dm(reply_user, thread_dm) - # Delete the thread reply - if FAKE_DELETE: - logger.info(f"FAKE_DELETE thread reply for {channel} {reply_ts}") - else: - slack.remove_message(channel, reply_ts) + ok = _delete_message(channel, reply_ts) + deleted_messages.append(_deleted_entry( + channel, reply_ts, reply_user, reply.get('text', ''), + 'reply', ok, + )) # Now handle the parent message message_pattern = reaction_config['message'] @@ -345,11 +355,16 @@ def handle_delete_message(event, reaction_config): if message_dm: slack.send_dm(user, message_dm) - # Delete the parent message - if FAKE_DELETE: - logger.info(f"FAKE_DELETE for {channel} {ts}") - else: - slack.remove_message(channel, ts) + ok = _delete_message(channel, ts) + deleted_messages.append(_deleted_entry( + channel, ts, user, original_message_raw, 'parent', ok, + )) + _record_delete_action( + event, + 'DELETE_MESSAGE', + deleted_messages, + target_user_id=user, + ) def handle_ask_ai(event, reaction_config): @@ -423,12 +438,40 @@ def handle_repost_to_thread_and_delete(event, reaction_config): slack.post_message_thread(event, thread_message) logger.info(f"Reposted message to thread for {channel} {ts}") - # Delete the original message from the channel - if FAKE_DELETE: - logger.info(f"FAKE_DELETE for {channel} {ts}") - else: - slack.remove_message(channel, ts) + ok = _delete_message(channel, ts) logger.info(f"Deleted original message from channel {channel} (kept in thread)") + _record_delete_action( + event, + 'REPOST_TO_THREAD_AND_DELETE', + [_deleted_entry(channel, ts, user, original_message, 'parent', ok)], + target_user_id=user, + ) + + +def _deleted_entry(channel, ts, user=None, text='', kind='parent', ok=True): + return { + 'channel': channel, + 'ts': ts, + 'user': user, + 'text': text, + 'kind': kind, + 'ok': ok, + 'fake': FAKE_DELETE, + } + + +def _record_delete_action(event, action_type, deleted_messages, **kwargs): + try: + audit.record_delete_action( + event, + action_type, + deleted_messages, + channel_name=get_channel_name((event.get('item') or {}).get('channel')), + fake_delete=FAKE_DELETE, + **kwargs, + ) + except Exception as exc: + logger.info(f"delete audit failed: {exc}") def _delete_message(channel, ts): @@ -480,7 +523,7 @@ def _format_reply_notification(template, reply_user, channel, reply_text): def _format_admin_summary(template, target_user_id, user_info, parent_deleted, - replies_deleted, admin_url): + replies_deleted, admin_url, extra_values=None): profile = (user_info or {}).get('profile', {}) updated_ts = (user_info or {}).get('updated') updated_str = datetime.utcfromtimestamp(updated_ts).strftime('%Y-%m-%d') if updated_ts else 'unknown' @@ -496,6 +539,8 @@ def _format_admin_summary(template, target_user_id, user_info, parent_deleted, 'thread_deleted_count': replies_deleted, 'admin_url': admin_url, } + if extra_values: + values.update(extra_values) return template.format(**values) @@ -530,6 +575,214 @@ def _build_admin_summary_blocks(summary, target_user_id, admin_url): return blocks +def _cleanup_user_message_matches(matches, target_user_id, reply_template): + deleted = set() + parent_deleted = 0 + replies_deleted = 0 + deleted_messages = [] + + for msg in matches: + msg_channel = (msg.get('channel') or {}).get('id') + msg_ts = msg.get('ts') + if not msg_channel or not msg_ts: + continue + + thread_ts = msg.get('thread_ts') + is_parent = (not thread_ts) or (thread_ts == msg_ts) + + if is_parent: + for reply in slack.get_thread_replies(msg_channel, msg_ts): + reply_ts = reply['ts'] + reply_user = reply.get('user') + key = (msg_channel, reply_ts) + if key in deleted: + continue + + if reply_user and reply_user != target_user_id and reply_template: + notification = _format_reply_notification( + reply_template, reply_user, msg_channel, + reply.get('text', '') + ) + slack.send_dm(reply_user, notification) + + ok = _delete_message(msg_channel, reply_ts) + deleted_messages.append(_deleted_entry( + msg_channel, reply_ts, reply_user, reply.get('text', ''), + 'reply', ok, + )) + if ok: + deleted.add(key) + replies_deleted += 1 + + key = (msg_channel, msg_ts) + if key in deleted: + continue + kind = 'parent' if is_parent else 'reply' + ok = _delete_message(msg_channel, msg_ts) + deleted_messages.append(_deleted_entry( + msg_channel, msg_ts, target_user_id, msg.get('text', ''), + kind, ok, + )) + if ok: + deleted.add(key) + parent_deleted += 1 + + return parent_deleted, replies_deleted, deleted_messages + + +def _notify_admins_of_cleanup(admin_ids, admin_template, target_user_id, + user_info, parent_deleted, replies_deleted, + extra_values=None): + if not admin_template or not admin_ids: + return + admin_url = (config.get('workspace') or {}).get('admin_users_url', '') + summary = _format_admin_summary( + admin_template, target_user_id, user_info, + parent_deleted, replies_deleted, admin_url, extra_values=extra_values, + ) + blocks = _build_admin_summary_blocks(summary, target_user_id, admin_url) + for admin_id in admin_ids: + slack.send_dm_blocks(admin_id, summary, blocks) + slack.send_dm(admin_id, target_user_id) + + +def _tracked_to_search_match(entry): + match = { + 'channel': {'id': entry.get('channel')}, + 'ts': entry.get('ts'), + 'text': entry.get('text') or '', + } + thread_ts = entry.get('thread_ts') + if thread_ts: + match['thread_ts'] = thread_ts + return match + + +def flush_cross_channel_spam(payload, *, now=None, store=None): + """Wipe a quiet cross-channel spam session and DM admins like :ban-user:.""" + user_id = payload.get('user_id') + fp = payload.get('fingerprint') + if not user_id or not fp: + return + + hooks = store or {} + load = hooks.get('load_session') or spam.load_session + delete = hooks.get('delete_session') or spam.delete_session + + try: + item = load(user_id, fp) + except Exception as exc: + logger.info(f"spam flush load failed: {exc}") + return + + if not item: + logger.info(f"spam flush: no session user={user_id} fp={fp}") + return + if item.get('status') == 'done': + return + + now = now if now is not None else time.time() + quiet = spam.quiet_seconds(config) + last_seen = float(item.get('last_seen') or 0) + if last_seen and (now - last_seen) < quiet - 1: + logger.info(f"spam flush: still posting user={user_id}") + return + + if user_id in set(config.get('admins', [])): + logger.info(f"spam flush: refusing admin {user_id}") + return + + spam_cfg = spam.tracker_config(config) + ban_cfg = reaction_configs.get('ban-user') or {} + reply_template = spam_cfg.get('reply_message') or ban_cfg.get('reply_message') + admin_template = spam_cfg.get('admin_message') or ban_cfg.get('admin_message') + lookback_hours = int(spam_cfg.get('lookback_hours') or ban_cfg.get('lookback_hours') or 24) + threshold = spam.similarity_threshold(config) + canonical = item.get('canonical_text') or '' + + user_info = slack.get_user_info(user_id) or {} + username = user_info.get('name') or (user_info.get('profile') or {}).get('display_name') + + matches = [] + seen = set() + for entry in item.get('messages') or []: + match = _tracked_to_search_match(entry) + key = _message_key(match) + if None in key or key in seen: + continue + matches.append(match) + seen.add(key) + + if username: + pad_days = lookback_hours // 24 + 1 + after_date = (datetime.utcnow() - timedelta(days=pad_days)).strftime('%Y-%m-%d') + cutoff_ts = now - lookback_hours * 3600 + for found in slack.search_user_messages(username, after_date): + if float(found.get('ts') or 0) < cutoff_ts: + continue + if not spam.is_similar(found.get('text') or '', canonical, threshold): + continue + key = _message_key(found) + if key in seen: + continue + matches.append(found) + seen.add(key) + + parent_deleted, replies_deleted, deleted_messages = ( + _cleanup_user_message_matches(matches, user_id, reply_template) + ) + + first = (item.get('messages') or [{}])[0] + audit_event = { + 'user': 'spam-tracker', + 'reaction': 'cross-channel-spam', + 'event_ts': str(now), + 'item': { + 'channel': first.get('channel'), + 'ts': first.get('ts'), + }, + 'item_user': user_id, + } + channels = sorted(spam.unique_channels(item.get('messages') or [])) + _record_delete_action( + audit_event, + 'CROSS_CHANNEL_SPAM', + deleted_messages, + target_user_id=user_id, + target_user={ + 'id': user_id, + 'name': username, + 'display_name': (user_info.get('profile') or {}).get('display_name') or '', + 'real_name': (user_info.get('profile') or {}).get('real_name') or '', + 'email': (user_info.get('profile') or {}).get('email') or '', + }, + extra={ + 'lookback_hours': lookback_hours, + 'parent_deleted': parent_deleted, + 'replies_deleted': replies_deleted, + 'channel_count': len(channels), + 'fingerprint': fp, + }, + ) + + admin_ids = spam_cfg.get('admin_users') or config.get('admins') or [] + channel_list = ' '.join(f'<#{cid}>' for cid in channels) + _notify_admins_of_cleanup( + admin_ids, admin_template, user_id, user_info, + parent_deleted, replies_deleted, + extra_values={ + 'channel_count': len(channels), + 'channel_list': channel_list, + 'sample_text': canonical[:500], + }, + ) + + try: + delete(user_id, fp) + except Exception as exc: + logger.info(f"spam flush delete session failed: {exc}") + + def handle_ban_user(event, reaction_config): """Delete recent messages from a spammer (no DM to them), notify innocent thread participants, and DM the reacting moderator @@ -586,55 +839,36 @@ def handle_ban_user(event, reaction_config): seen.add(key) reply_template = reaction_config.get('reply_message') - deleted = set() - parent_deleted = 0 - replies_deleted = 0 - - for msg in matches: - msg_channel = (msg.get('channel') or {}).get('id') - msg_ts = msg.get('ts') - if not msg_channel or not msg_ts: - continue - - thread_ts = msg.get('thread_ts') - is_parent = (not thread_ts) or (thread_ts == msg_ts) - - if is_parent: - for reply in slack.get_thread_replies(msg_channel, msg_ts): - reply_ts = reply['ts'] - reply_user = reply.get('user') - key = (msg_channel, reply_ts) - if key in deleted: - continue - - if reply_user and reply_user != target_user_id and reply_template: - notification = _format_reply_notification( - reply_template, reply_user, msg_channel, - reply.get('text', '') - ) - slack.send_dm(reply_user, notification) - - if _delete_message(msg_channel, reply_ts): - deleted.add(key) - replies_deleted += 1 + parent_deleted, replies_deleted, deleted_messages = ( + _cleanup_user_message_matches(matches, target_user_id, reply_template) + ) - key = (msg_channel, msg_ts) - if key in deleted: - continue - if _delete_message(msg_channel, msg_ts): - deleted.add(key) - parent_deleted += 1 + _record_delete_action( + event, + 'BAN_USER', + deleted_messages, + target_user_id=target_user_id, + target_user={ + 'id': target_user_id, + 'name': username, + 'display_name': (user_info.get('profile') or {}).get('display_name') or '', + 'real_name': (user_info.get('profile') or {}).get('real_name') or '', + 'email': (user_info.get('profile') or {}).get('email') or '', + }, + extra={ + 'lookback_hours': lookback_hours, + 'parent_deleted': parent_deleted, + 'replies_deleted': replies_deleted, + 'search_match_count': len(search_matches), + }, + ) admin_template = reaction_config.get('admin_message') if admin_template and moderator: - admin_url = (config.get('workspace') or {}).get('admin_users_url', '') - summary = _format_admin_summary( - admin_template, target_user_id, user_info, - parent_deleted, replies_deleted, admin_url + _notify_admins_of_cleanup( + [moderator], admin_template, target_user_id, user_info, + parent_deleted, replies_deleted, ) - blocks = _build_admin_summary_blocks(summary, target_user_id, admin_url) - slack.send_dm_blocks(moderator, summary, blocks) - slack.send_dm(moderator, target_user_id) action_handlers = { @@ -691,10 +925,44 @@ def run(body): process_reaction(body, event) return + if event_type == 'message': + spam.ingest_message(event, config) + return + logger.info(f'unsupported event type: {event_type}') +def flush_quiet_spam_sessions(*, now=None, store=None): + """Scheduled scan: flush ready aggregates that have been quiet long enough.""" + now = now if now is not None else time.time() + cutoff = int(now) - spam.quiet_seconds(config) + hooks = store or {} + list_ready = hooks.get('list_ready_sessions') or spam.list_ready_sessions + try: + items = list_ready(cutoff) + except Exception as exc: + logger.info(f"spam scan load failed: {exc}") + return + logger.info(f"spam scan: {len(items)} quiet aggregate(s) cutoff={cutoff}") + for item in items: + flush_cross_channel_spam( + { + 'user_id': item.get('user_id'), + 'fingerprint': item.get('fingerprint'), + }, + now=now, + store=store, + ) + + def lambda_handler(event, context): + if isinstance(event, dict) and event.get('action') == 'spam_quiet_scan': + flush_quiet_spam_sessions() + return { + 'statusCode': 200, + 'body': "done" + } + run(event) return { 'statusCode': 200, diff --git a/automator/src/spam.py b/automator/src/spam.py new file mode 100644 index 0000000..7b027bf --- /dev/null +++ b/automator/src/spam.py @@ -0,0 +1,257 @@ +"""Cross-channel similar-message tracker. + +Each qualifying post updates a DynamoDB aggregate per (user, fingerprint): +channel set, message list, last_seen. Sessions that have reached two +channels sit on a sparse GSI (`ready` / last_seen). EventBridge every 5 +minutes flushes aggregates whose last_seen is older than quiet_seconds. +""" + +import hashlib +import os +import re +import time +from difflib import SequenceMatcher + +from logs import logger + + +SPAM_TABLE = os.getenv('SPAM_TABLE', 'automator-spam-sessions') +TTL_SECONDS = 24 * 3600 +MAX_TRACKED_MESSAGES = 200 +READY_PK = 'ready' + +_SKIP_SUBTYPES = { + 'bot_message', + 'message_changed', + 'message_deleted', + 'channel_join', + 'channel_leave', + 'channel_topic', + 'channel_purpose', + 'message_replied', +} + + +def tracker_config(config): + return (config or {}).get('spam_tracker') or {} + + +def enabled(config): + if os.getenv('SPAM_TRACKER_ENABLED', '1') != '1': + return False + cfg = tracker_config(config) + return cfg.get('enabled', True) + + +def quiet_seconds(config): + seconds = int(tracker_config(config).get('quiet_seconds', 180)) + return max(30, min(seconds, 3600)) + + +def min_channels(config): + return int(tracker_config(config).get('min_channels', 2)) + + +def min_chars(config): + return int(tracker_config(config).get('min_chars', 40)) + + +def similarity_threshold(config): + return float(tracker_config(config).get('similarity', 0.88)) + + +def normalize_text(text): + text = text or '' + text = re.sub(r'<(https?://[^|>]+)(?:\|[^>]*)?>', r'\1', text) + text = re.sub(r'<#[A-Z0-9]+\|([^>]+)>', r'#\1', text) + text = re.sub(r'<@[A-Z0-9]+>', ' ', text) + text = text.lower() + text = re.sub(r'\s+', ' ', text).strip() + return text + + +def fingerprint(text): + normalized = normalize_text(text) + digest = hashlib.sha256(normalized.encode('utf-8')).hexdigest() + return digest[:16] + + +def is_similar(left, right, threshold=0.88): + a = normalize_text(left) + b = normalize_text(right) + if not a or not b: + return False + if a == b: + return True + return SequenceMatcher(None, a, b).ratio() >= threshold + + +def should_ingest(event, config): + if not enabled(config): + return False + if (event or {}).get('type') != 'message': + return False + if event.get('bot_id') or event.get('subtype') in _SKIP_SUBTYPES: + return False + user = event.get('user') + if not user: + return False + if user in set((config or {}).get('admins') or []): + return False + text = event.get('text') or '' + if len(normalize_text(text)) < min_chars(config): + return False + if not event.get('channel') or not event.get('ts'): + return False + return True + + +def session_key(user_id, fp): + return {'pk': f'USER#{user_id}', 'sk': f'FP#{fp}'} + + +def _message_entry(event): + return { + 'channel': event.get('channel'), + 'ts': event.get('ts'), + 'text': event.get('text') or '', + 'thread_ts': event.get('thread_ts'), + } + + +def unique_channels(messages): + return {m.get('channel') for m in messages if m.get('channel')} + + +def mark_ready(item, channel_count, now, threshold): + """Sparse GSI: only multi-channel sessions are scanable.""" + item['channel_count'] = channel_count + item['last_seen'] = int(now) + if channel_count >= threshold: + item['gsi1pk'] = READY_PK + item['gsi1sk'] = int(now) + else: + item.pop('gsi1pk', None) + item.pop('gsi1sk', None) + return item + + +def _table(): + import boto3 + return boto3.resource('dynamodb').Table(SPAM_TABLE) + + +def load_sessions(user_id): + from boto3.dynamodb.conditions import Key + + table = _table() + items = [] + kwargs = { + 'KeyConditionExpression': Key('pk').eq(f'USER#{user_id}'), + } + while True: + resp = table.query(**kwargs) + items.extend(resp.get('Items') or []) + if 'LastEvaluatedKey' not in resp: + break + kwargs['ExclusiveStartKey'] = resp['LastEvaluatedKey'] + return items + + +def load_session(user_id, fp): + table = _table() + resp = table.get_item(Key=session_key(user_id, fp)) + return resp.get('Item') + + +def save_session(item): + _table().put_item(Item=item) + + +def delete_session(user_id, fp): + _table().delete_item(Key=session_key(user_id, fp)) + + +def list_ready_sessions(cutoff): + """Sessions on the ready GSI whose last_seen is at or before cutoff.""" + from boto3.dynamodb.conditions import Key + + table = _table() + items = [] + kwargs = { + 'IndexName': 'gsi1', + 'KeyConditionExpression': ( + Key('gsi1pk').eq(READY_PK) & Key('gsi1sk').lte(int(cutoff)) + ), + } + while True: + resp = table.query(**kwargs) + items.extend(resp.get('Items') or []) + if 'LastEvaluatedKey' not in resp: + break + kwargs['ExclusiveStartKey'] = resp['LastEvaluatedKey'] + return items + + +def find_matching_session(sessions, text, threshold): + for item in sessions: + canonical = item.get('canonical_text') or '' + if is_similar(text, canonical, threshold): + return item + return None + + +def ingest_message(event, config, *, now=None, store=None): + """Fold a message into the (user, fingerprint) aggregate. No I/O besides Dynamo.""" + if not should_ingest(event, config): + return None + + now = now if now is not None else time.time() + user_id = event['user'] + text = event.get('text') or '' + fp = fingerprint(text) + threshold = similarity_threshold(config) + hooks = store or {} + load = hooks.get('load_sessions') or load_sessions + save = hooks.get('save_session') or save_session + + try: + sessions = load(user_id) + except Exception as exc: + logger.info(f"spam tracker load failed: {exc}") + return None + + item = find_matching_session(sessions, text, threshold) + if item is None: + item = { + 'pk': f'USER#{user_id}', + 'sk': f'FP#{fp}', + 'user_id': user_id, + 'fingerprint': fp, + 'canonical_text': text, + 'messages': [], + 'status': 'collecting', + } + + messages = list(item.get('messages') or []) + key = (event.get('channel'), event.get('ts')) + if not any((m.get('channel'), m.get('ts')) == key for m in messages): + if len(messages) < MAX_TRACKED_MESSAGES: + messages.append(_message_entry(event)) + item['messages'] = messages + item['expires_at'] = int(now) + TTL_SECONDS + item['status'] = 'collecting' + mark_ready(item, len(unique_channels(messages)), now, min_channels(config)) + + try: + save(item) + except Exception as extra: + logger.info(f"spam tracker save failed: {extra}") + return item + + if item.get('gsi1pk') == READY_PK: + logger.info( + f"spam tracker aggregate user={user_id} fp={item['fingerprint']} " + f"channels={item.get('channel_count')} last_seen={item.get('last_seen')}" + ) + return item diff --git a/automator/tests/test_audit.py b/automator/tests/test_audit.py new file mode 100644 index 0000000..311b9dd --- /dev/null +++ b/automator/tests/test_audit.py @@ -0,0 +1,155 @@ +import os +import sys +import unittest +from datetime import datetime, timezone +from unittest.mock import patch + +os.environ.setdefault('FAKE_DELETE', '1') +os.environ['AUDIT_ENABLED'] = '0' +os.environ['SPAM_TRACKER_ENABLED'] = '0' +src_dir = os.path.join(os.path.dirname(__file__), '..', 'src') +sys.path.insert(0, src_dir) + +import audit # noqa: E402 + + +class TestBuildAuditRecord(unittest.TestCase): + def setUp(self): + self.now = datetime(2026, 4, 8, 12, 30, 0, tzinfo=timezone.utc) + self.event = { + 'user': 'UMOD', + 'reaction': 'ban-user', + 'event_ts': '1775646600.000001', + 'item': {'channel': 'C_AAA', 'ts': '1775646500.000100'}, + 'item_user': 'U_SPAMMER', + } + + def test_keys_and_trigger(self): + record = audit.build_audit_record( + self.event, + 'BAN_USER', + [{ + 'channel': 'C_AAA', + 'ts': '1775646500.000100', + 'user': 'U_SPAMMER', + 'text': 'buy followers cheap', + 'kind': 'parent', + 'ok': True, + }], + target_user_id='U_SPAMMER', + channel_name='course-ml-zoomcamp', + extra={'lookback_hours': 24}, + now=self.now, + ) + + self.assertEqual(record['triggered_by'], 'UMOD') + self.assertEqual(record['reaction'], 'ban-user') + self.assertEqual(record['action_type'], 'BAN_USER') + self.assertEqual(record['target_user_id'], 'U_SPAMMER') + self.assertEqual(record['channel_name'], 'course-ml-zoomcamp') + self.assertEqual(record['deleted_messages'][0]['text'], 'buy followers cheap') + self.assertEqual(record['pk'], 'USER#U_SPAMMER') + self.assertTrue(record['sk'].startswith('2026-04-08T12:30:00Z#')) + self.assertEqual(record['gsi1pk'], 'DATE#2026-04-08') + self.assertEqual(record['expires_at'], int(self.now.timestamp()) + 90 * 86400) + self.assertEqual( + record['action_id'], + 'C_AAA:1775646500.000100:ban-user:1775646600.000001', + ) + + def test_clips_long_message_text(self): + record = audit.build_audit_record( + self.event, + 'DELETE_MESSAGE', + [{'channel': 'C_AAA', 'ts': '1', 'text': 'x' * 9000, 'ok': True}], + now=self.now, + ) + self.assertEqual(len(record['deleted_messages'][0]['text']), 4001) + self.assertTrue(record['deleted_messages'][0]['text'].endswith('…')) + + def test_unknown_user_pk_when_target_missing(self): + record = audit.build_audit_record( + self.event, 'DELETE_MESSAGE', [], now=self.now, + ) + self.assertEqual(record['pk'], 'USER#unknown') + + def test_s3_keys_include_date_and_user_prefixes(self): + record = audit.build_audit_record( + self.event, + 'BAN_USER', + [], + target_user_id='U_SPAMMER', + now=self.now, + ) + keys = audit.s3_keys(record) + self.assertEqual( + keys, + [ + 'by-date/2026/04/08/C_AAA:1775646500.000100:ban-user:1775646600.000001.json', + 'by-user/U_SPAMMER/2026/04/08/C_AAA:1775646500.000100:ban-user:1775646600.000001.json', + ], + ) + + def test_drops_none_fields(self): + record = audit.build_audit_record( + {'item': {}}, 'DELETE_MESSAGE', [], now=self.now, + ) + self.assertNotIn('triggered_by', record) + self.assertNotIn('target_user_id', record) + + +class TestRecordDeleteAction(unittest.TestCase): + def test_skips_aws_when_disabled(self): + event = { + 'user': 'UMOD', + 'reaction': 'delete', + 'item': {'channel': 'C1', 'ts': '1.0'}, + } + with patch.dict(os.environ, {'AUDIT_ENABLED': '0'}), \ + patch('audit._put_dynamo') as put_d, \ + patch('audit._put_s3') as put_s: + record = audit.record_delete_action( + event, 'DELETE_MESSAGE', + [{'channel': 'C1', 'ts': '1.0', 'text': 'hi', 'ok': True}], + target_user_id='U1', + ) + put_d.assert_not_called() + put_s.assert_not_called() + self.assertEqual(record['action_type'], 'DELETE_MESSAGE') + + def test_writes_dynamo_and_s3_when_enabled(self): + event = { + 'user': 'UMOD', + 'reaction': 'delete', + 'item': {'channel': 'C1', 'ts': '1.0'}, + } + with patch.dict(os.environ, {'AUDIT_ENABLED': '1'}), \ + patch('audit._put_dynamo') as put_d, \ + patch('audit._put_s3') as put_s: + record = audit.record_delete_action( + event, 'DELETE_MESSAGE', + [{'channel': 'C1', 'ts': '1.0', 'text': 'hi', 'ok': True}], + target_user_id='U1', + ) + put_d.assert_called_once() + self.assertEqual(put_d.call_args.args[0]['pk'], 'USER#U1') + self.assertEqual(put_s.call_count, 2) + s3_keys = [c.args[0] for c in put_s.call_args_list] + self.assertTrue(any(k.startswith('by-date/') for k in s3_keys)) + self.assertTrue(any(k.startswith('by-user/U1/') for k in s3_keys)) + self.assertEqual(record['triggered_by'], 'UMOD') + + def test_aws_errors_do_not_raise(self): + event = { + 'user': 'UMOD', + 'reaction': 'delete', + 'item': {'channel': 'C1', 'ts': '1.0'}, + } + with patch.dict(os.environ, {'AUDIT_ENABLED': '1'}), \ + patch('audit._put_dynamo', side_effect=RuntimeError('no table')), \ + patch('audit._put_s3', side_effect=RuntimeError('no bucket')): + record = audit.record_delete_action( + event, 'DELETE_MESSAGE', + [{'channel': 'C1', 'ts': '1.0', 'text': 'hi', 'ok': True}], + ) + self.assertIsNotNone(record) diff --git a/automator/tests/test_automator_lambda.py b/automator/tests/test_automator_lambda.py index 927fac0..1db1def 100644 --- a/automator/tests/test_automator_lambda.py +++ b/automator/tests/test_automator_lambda.py @@ -6,6 +6,8 @@ # Set environment before importing lambda_function os.environ['FAKE_DELETE'] = '1' +os.environ['AUDIT_ENABLED'] = '0' +os.environ['SPAM_TRACKER_ENABLED'] = '0' config_path = os.path.join(os.path.dirname(__file__), '..', 'src', 'config.yaml') os.environ['CONFIG_FILE'] = config_path @@ -977,6 +979,226 @@ def test_action_handlers_has_ban_user(self): ) +class TestDeleteAudit(unittest.TestCase): + """Delete handlers record what was removed and what triggered it.""" + + @patch('automator_lambda_function.audit.record_delete_action') + @patch('automator_lambda_function.slack') + @patch('automator_lambda_function.util') + def test_delete_message_records_parent_and_replies( + self, mock_util, mock_slack, mock_audit + ): + mock_slack.get_message_content.return_value = { + 'user': 'U_PARENT', + 'text': 'spam parent that broke the rules', + } + mock_slack.get_thread_replies.return_value = [ + {'user': 'U_REPLY', 'ts': '1234567890.123457', 'text': 'a reply'}, + ] + mock_util.format_message.return_value = 'dm' + event = { + 'user': 'UMOD', + 'item': {'channel': 'C123456', 'ts': '1234567890.123456'}, + 'reaction': 'delete', + 'event_ts': '1234567891.000000', + } + reaction_config = { + 'message': 'removed', + 'thread_message': 'reply removed', + 'type': 'DELETE_MESSAGE', + } + + lambda_function.handle_delete_message(event, reaction_config) + + mock_audit.assert_called_once() + args, kwargs = mock_audit.call_args + self.assertEqual(args[0], event) + self.assertEqual(args[1], 'DELETE_MESSAGE') + deleted = args[2] + self.assertEqual( + {(m['channel'], m['ts'], m['user'], m['kind']) for m in deleted}, + { + ('C123456', '1234567890.123457', 'U_REPLY', 'reply'), + ('C123456', '1234567890.123456', 'U_PARENT', 'parent'), + }, + ) + parent = next(m for m in deleted if m['kind'] == 'parent') + self.assertEqual(parent['text'], 'spam parent that broke the rules') + self.assertEqual(kwargs['target_user_id'], 'U_PARENT') + + @patch('automator_lambda_function.audit.record_delete_action') + @patch('automator_lambda_function.slack') + def test_ban_user_records_trigger_and_every_deleted_message( + self, mock_slack, mock_audit + ): + mock_slack.get_message_content.return_value = { + 'user': 'U_SPAMMER', + 'text': 'spam message', + } + mock_slack.get_user_info.return_value = { + 'id': 'U_SPAMMER', + 'name': 'spammer', + 'profile': { + 'display_name': 'Spammer', + 'real_name': 'Spam McSpamface', + 'email': 'spam@example.com', + }, + } + mock_slack.search_user_messages.return_value = [ + { + 'channel': {'id': 'C_AAA'}, + 'ts': '1234567890.000100', + 'text': 'spam parent', + }, + { + 'channel': {'id': 'C_BBB'}, + 'ts': '1234567890.000200', + 'thread_ts': '1234567890.000150', + 'text': 'spam reply', + }, + ] + mock_slack.get_thread_replies.return_value = [ + {'user': 'U_VICTIM', 'ts': '1234567890.000101', 'text': 'innocent reply'}, + {'user': 'U_SPAMMER', 'ts': '1234567890.000102', 'text': 'spammer follow-up'}, + ] + event = { + 'user': 'UMOD', + 'item': {'channel': 'C_AAA', 'ts': '1234567890.000100'}, + 'reaction': 'ban-user', + 'event_ts': '1234567891.000000', + } + + with patch('automator_lambda_function.time.time', return_value=1234567890.0 + 1000), \ + patch('automator_lambda_function.FAKE_DELETE', False): + lambda_function.handle_ban_user(event, self._ban_config()) + + mock_audit.assert_called_once() + args, kwargs = mock_audit.call_args + self.assertEqual(args[0]['user'], 'UMOD') + self.assertEqual(args[0]['reaction'], 'ban-user') + self.assertEqual(args[1], 'BAN_USER') + deleted = args[2] + self.assertEqual( + {(m['channel'], m['ts'], m['text']) for m in deleted}, + { + ('C_AAA', '1234567890.000101', 'innocent reply'), + ('C_AAA', '1234567890.000102', 'spammer follow-up'), + ('C_AAA', '1234567890.000100', 'spam message'), + ('C_BBB', '1234567890.000200', 'spam reply'), + }, + ) + self.assertEqual(kwargs['target_user_id'], 'U_SPAMMER') + self.assertEqual(kwargs['target_user']['email'], 'spam@example.com') + self.assertEqual(kwargs['extra']['parent_deleted'], 2) + self.assertEqual(kwargs['extra']['replies_deleted'], 2) + + @patch('automator_lambda_function.audit.record_delete_action') + @patch('automator_lambda_function.slack') + def test_ban_user_does_not_audit_when_refusing_admin( + self, mock_slack, mock_audit + ): + mock_slack.get_message_content.return_value = { + 'user': 'U01AXE0P5M3', + 'text': 'whatever', + } + event = { + 'user': 'UMOD', + 'item': {'channel': 'C_AAA', 'ts': '1234567890.000100'}, + 'reaction': 'ban-user', + } + lambda_function.handle_ban_user(event, self._ban_config()) + mock_audit.assert_not_called() + + def _ban_config(self): + return { + 'lookback_hours': 24, + 'reply_message': 'Hi <@{user}>: {user_message}', + 'admin_message': ( + "User: <@{target_user}>\n" + "ID: {target_user_id}\n" + "Deleted: {deleted_count} parents, {thread_deleted_count} replies\n" + "Admin: {admin_url}" + ), + } + + +class TestCrossChannelSpam(unittest.TestCase): + def _session(self, last_seen=1000): + text = ( + 'Get certified in data engineering today, limited offer, ' + 'click https://spam.example/join now!!!' + ) + return { + 'pk': 'USER#U_SPAMMER', + 'sk': 'FP#abc', + 'user_id': 'U_SPAMMER', + 'fingerprint': 'abc', + 'canonical_text': text, + 'last_seen': last_seen, + 'gsi1pk': 'ready', + 'gsi1sk': last_seen, + 'channel_count': 2, + 'status': 'collecting', + 'messages': [ + {'channel': 'C1', 'ts': '1.0', 'text': text}, + {'channel': 'C2', 'ts': '2.0', 'text': text}, + ], + } + + @patch('automator_lambda_function.audit.record_delete_action') + @patch('automator_lambda_function.slack') + def test_flush_deletes_and_dms_admin_id_separately(self, mock_slack, mock_audit): + session = self._session() + store = { + 'load_session': lambda user_id, fp: session, + 'delete_session': lambda user_id, fp: store.setdefault('deleted', (user_id, fp)), + } + mock_slack.get_user_info.return_value = { + 'id': 'U_SPAMMER', + 'name': 'spammer', + 'profile': { + 'display_name': 'Spammer', + 'real_name': 'Spam Person', + 'email': 'spam@example.com', + }, + } + mock_slack.search_user_messages.return_value = [] + mock_slack.get_thread_replies.return_value = [] + + with patch('automator_lambda_function.FAKE_DELETE', False): + lambda_function.flush_cross_channel_spam( + {'user_id': 'U_SPAMMER', 'fingerprint': 'abc'}, + now=1000 + 180, + store=store, + ) + + deleted_targets = { + (c.args[0], c.args[1]) for c in mock_slack.remove_message.call_args_list + } + self.assertEqual(deleted_targets, {('C1', '1.0'), ('C2', '2.0')}) + mock_slack.send_dm_blocks.assert_called_once() + self.assertEqual(mock_slack.send_dm_blocks.call_args.args[0], 'U01AXE0P5M3') + id_dms = [ + c for c in mock_slack.send_dm.call_args_list + if c.args == ('U01AXE0P5M3', 'U_SPAMMER') + ] + self.assertEqual(len(id_dms), 1) + self.assertEqual(store['deleted'], ('U_SPAMMER', 'abc')) + self.assertEqual(mock_audit.call_args.args[1], 'CROSS_CHANNEL_SPAM') + + @patch('automator_lambda_function.slack') + def test_scan_skips_aggregates_that_are_still_posting(self, mock_slack): + session = self._session(last_seen=1000 + 170) + store = { + 'list_ready_sessions': lambda cutoff: [session], + 'load_session': lambda user_id, fp: session, + 'delete_session': lambda *args: None, + } + lambda_function.flush_quiet_spam_sessions(now=1000 + 180, store=store) + mock_slack.remove_message.assert_not_called() + mock_slack.send_dm.assert_not_called() + + class TestFaqAssistant(unittest.TestCase): """Test Cloudflare FAQ assistant integration.""" diff --git a/automator/tests/test_reactions_e2e.py b/automator/tests/test_reactions_e2e.py index 5e193b9..ade69ec 100644 --- a/automator/tests/test_reactions_e2e.py +++ b/automator/tests/test_reactions_e2e.py @@ -24,6 +24,8 @@ lambda_function = sys.modules[MODNAME] else: os.environ['FAKE_DELETE'] = '1' + os.environ['AUDIT_ENABLED'] = '0' + os.environ['SPAM_TRACKER_ENABLED'] = '0' src_dir = os.path.join(os.path.dirname(__file__), '..', 'src') os.environ['CONFIG_FILE'] = os.path.join(src_dir, 'config.yaml') spec = importlib.util.spec_from_file_location( diff --git a/automator/tests/test_spam.py b/automator/tests/test_spam.py new file mode 100644 index 0000000..f239fe3 --- /dev/null +++ b/automator/tests/test_spam.py @@ -0,0 +1,170 @@ +import os +import sys +import unittest +from unittest.mock import patch + +os.environ.setdefault('FAKE_DELETE', '1') +os.environ.setdefault('AUDIT_ENABLED', '0') +os.environ.setdefault('SPAM_TRACKER_ENABLED', '0') +src_dir = os.path.join(os.path.dirname(__file__), '..', 'src') +os.environ['CONFIG_FILE'] = os.path.join(src_dir, 'config.yaml') +sys.path.insert(0, src_dir) + +import spam # noqa: E402 + + +CONFIG = { + 'admins': ['UADMIN'], + 'spam_tracker': { + 'enabled': True, + 'quiet_seconds': 180, + 'min_channels': 2, + 'min_chars': 40, + 'similarity': 0.88, + }, +} + +SPAM_TEXT = ( + 'Get certified in data engineering today, limited offer, ' + 'click https://spam.example/join now!!!' +) + + +def _event(channel, ts, text=SPAM_TEXT, user='U_SPAMMER'): + return { + 'type': 'message', + 'user': user, + 'channel': channel, + 'ts': ts, + 'text': text, + } + + +class MemoryStore: + def __init__(self): + self.items = {} + + def load_sessions(self, user_id): + prefix = f'USER#{user_id}' + return [v for k, v in self.items.items() if k[0] == prefix] + + def load_session(self, user_id, fp): + return self.items.get((f'USER#{user_id}', f'FP#{fp}')) + + def save_session(self, item): + self.items[(item['pk'], item['sk'])] = dict(item) + + def delete_session(self, user_id, fp): + self.items.pop((f'USER#{user_id}', f'FP#{fp}'), None) + + def list_ready_sessions(self, cutoff): + return [ + v for v in self.items.values() + if v.get('gsi1pk') == 'ready' and int(v.get('gsi1sk') or 0) <= cutoff + ] + + def hooks(self): + return { + 'load_sessions': self.load_sessions, + 'save_session': self.save_session, + 'load_session': self.load_session, + 'delete_session': self.delete_session, + 'list_ready_sessions': self.list_ready_sessions, + } + + +class TestSimilarity(unittest.TestCase): + def test_normalize_collapses_slack_markup(self): + raw = 'Hello <@U123> see NOW' + self.assertEqual(spam.normalize_text(raw), 'hello see https://x.example now') + + def test_same_text_is_similar(self): + self.assertTrue(spam.is_similar(SPAM_TEXT, SPAM_TEXT.upper())) + + def test_small_edit_is_similar(self): + other = SPAM_TEXT.replace('today', 'today!') + self.assertTrue(spam.is_similar(SPAM_TEXT, other, 0.88)) + + def test_unrelated_is_not_similar(self): + self.assertFalse(spam.is_similar(SPAM_TEXT, 'anyone know the homework deadline?')) + + +class TestShouldIngest(unittest.TestCase): + def setUp(self): + self.env = patch.dict(os.environ, {'SPAM_TRACKER_ENABLED': '1'}) + self.env.start() + + def tearDown(self): + self.env.stop() + + def test_skips_short_and_admin_and_bots(self): + self.assertFalse(spam.should_ingest( + _event('C1', '1.0', text='hi'), CONFIG, + )) + self.assertFalse(spam.should_ingest( + _event('C1', '1.0', user='UADMIN'), CONFIG, + )) + bot = _event('C1', '1.0') + bot['bot_id'] = 'B1' + self.assertFalse(spam.should_ingest(bot, CONFIG)) + changed = _event('C1', '1.0') + changed['subtype'] = 'message_changed' + self.assertFalse(spam.should_ingest(changed, CONFIG)) + + def test_accepts_long_user_message(self): + self.assertTrue(spam.should_ingest(_event('C1', '1.0'), CONFIG)) + + +class TestIngest(unittest.TestCase): + def setUp(self): + self.env = patch.dict(os.environ, {'SPAM_TRACKER_ENABLED': '1'}) + self.env.start() + + def tearDown(self): + self.env.stop() + + def test_one_channel_stays_off_ready_index(self): + store = MemoryStore() + item = spam.ingest_message( + _event('C1', '1.0'), CONFIG, now=1000, store=store.hooks(), + ) + self.assertEqual(item['channel_count'], 1) + self.assertNotIn('gsi1pk', item) + self.assertEqual(store.list_ready_sessions(1000), []) + + def test_second_channel_marks_ready_with_last_seen(self): + store = MemoryStore() + hooks = store.hooks() + spam.ingest_message(_event('C1', '1.0'), CONFIG, now=1000, store=hooks) + item = spam.ingest_message( + _event('C2', '2.0'), CONFIG, now=1010, store=hooks, + ) + self.assertEqual(item['channel_count'], 2) + self.assertEqual(item['gsi1pk'], 'ready') + self.assertEqual(item['gsi1sk'], 1010) + self.assertEqual(store.list_ready_sessions(1009), []) + self.assertEqual(len(store.list_ready_sessions(1010)), 1) + + def test_later_copy_moves_last_seen(self): + store = MemoryStore() + hooks = store.hooks() + spam.ingest_message(_event('C1', '1.0'), CONFIG, now=1000, store=hooks) + spam.ingest_message(_event('C2', '2.0'), CONFIG, now=1010, store=hooks) + item = spam.ingest_message( + _event('C3', '3.0'), CONFIG, now=1020, store=hooks, + ) + self.assertEqual(item['gsi1sk'], 1020) + self.assertEqual(item['channel_count'], 3) + self.assertEqual(store.list_ready_sessions(1010), []) + self.assertEqual(len(store.list_ready_sessions(1020)), 1) + + def test_near_duplicate_joins_same_session(self): + store = MemoryStore() + hooks = store.hooks() + spam.ingest_message(_event('C1', '1.0'), CONFIG, now=1000, store=hooks) + variant = SPAM_TEXT.replace('today', 'today!') + item = spam.ingest_message( + _event('C2', '2.0', text=variant), CONFIG, now=1010, store=hooks, + ) + self.assertEqual(len(store.items), 1) + self.assertEqual(len(item['messages']), 2) diff --git a/docs/delete-audit.md b/docs/delete-audit.md new file mode 100644 index 0000000..9a9234a --- /dev/null +++ b/docs/delete-audit.md @@ -0,0 +1,89 @@ +# Delete-action audit + +When an admin reaction deletes Slack messages, the automator records **what +was deleted**, **which reaction triggered it**, and **who triggered it**. + +This is especially important for `:ban-user:`, which can wipe a user's recent +messages plus thread replies. + +## What is stored + +Each action is one record: + +- `action_type` — `DELETE_MESSAGE`, `BAN_USER`, `REPOST_TO_THREAD_AND_DELETE`, `REMOVE_BROADCAST`, or `CROSS_CHANNEL_SPAM` +- `reaction` — emoji that triggered it (`delete`, `ban-user`, `shameless-rules`, …) +- `triggered_by` — Slack user id of the reacting admin +- `target_user_id` / `target_user` — author being acted on (id, name, email when known) +- `channel` / `channel_name` / `message_ts` — the reacted message +- `deleted_messages[]` — every message removed: channel, ts, user, text, kind (`parent` / `reply` / `broadcast`), whether Slack accepted the delete +- `extra` — for bans: lookback window and parent/reply counts + +Message text is clipped at 4,000 characters. + +## Where it lives (cheap) + +Dual-written at event time. No extra dump job. + +| Store | Retention | Cost | Use | +| --- | --- | --- | --- | +| CloudWatch log line `DELETE_AUDIT {json}` | whatever the Lambda log group keeps | already paying | last couple of weeks, Insights | +| DynamoDB `automator-delete-audit` (on-demand, TTL) | 90 days | ~$0 at this volume | query by user or by day | +| S3 Standard `dtc-automator-delete-audit-387546586013` | forever | a few KB per action | archive after Dynamo expires | + +S3 stays in Standard on purpose. Glacier and Standard-IA bill a 128 KB +minimum per object; these files are a few KB, so Standard is cheaper. + +Infra: `aws-infra/main/automator/audit.tf`. Apply that stack before expecting +Dynamo/S3 writes. Until then the Lambda still logs `DELETE_AUDIT` and +swallows storage errors so deletes keep working. + +## Query + +From `automator/`: + +```bash +# Recent actions against one user (Dynamo, last 90 days) +uv run python scripts/query_delete_audit.py --user U123 + +# Everything on a calendar day +uv run python scripts/query_delete_audit.py --date 2026-04-08 + +# Forever copy in S3 (also works after Dynamo TTL) +uv run python scripts/query_delete_audit.py --user U123 --s3 +``` + +## Last two weeks from existing logs + +Current production logs **do not include deleted message bodies**. Each +invocation prints the Slack event (`print(json.dumps(body))`), which has: + +- who reacted (`event.user`) +- which emoji (`event.reaction`) +- which channel / message ts (`event.item`) + +It does **not** have the message text. After a delete, Slack no longer has +that text either. Historical recovery of *what was said* is not possible from +logs written before this audit existed. + +Once AWS Gate is open on this host, search the last 14 days of triggers: + +```bash +START_MS=$(python3 -c 'import time; print(int((time.time()-14*86400)*1000))') +aws logs filter-log-events \ + --log-group-name /aws/lambda/automator-process-reaction \ + --start-time "$START_MS" \ + --filter-pattern '"ban-user"' \ + --region eu-west-1 +``` + +Useful filter patterns: `"ban-user"`, `"reaction": "delete"`, `shameless-rules`, +`jobs-rules`, `DELETE_AUDIT`. + +## Production access for agents + +AWS Gate vends the **sandbox** role (`817685572750`) only. The audit table and +bucket are in **main** (`387546586013`). `cmp-alert-investigator` cannot query +this table. + +Query with main-account operator credentials, or `/delete-audit`. See +`.grok/skills/delete-audit/SKILL.md`. diff --git a/docs/logs.md b/docs/logs.md index d32eb9c..68cccad 100644 --- a/docs/logs.md +++ b/docs/logs.md @@ -11,6 +11,15 @@ The automator lambda logs are stored in: ### Using the "my" profile +The default credentials on this host go through AWS Gate. If Gate is closed +(`HTTP 403`), open it (phone pairing) or use a profile that can read +CloudWatch. A `my` profile is documented below but may not exist on every +machine. + +Delete actions (who reacted, which emoji, which channel/ts, and — after the +audit landed — the deleted message text) are documented in +[delete-audit.md](delete-audit.md). + The default credentials only allow lambda deployment. Use the `my` profile for logs: ```bash diff --git a/router/src/lambda_function.py b/router/src/lambda_function.py index fdd6f6f..6451aa7 100644 --- a/router/src/lambda_function.py +++ b/router/src/lambda_function.py @@ -9,6 +9,25 @@ admins = {'U01AXE0P5M3'} +_SKIP_MESSAGE_SUBTYPES = { + 'bot_message', + 'message_changed', + 'message_deleted', + 'channel_join', + 'channel_leave', + 'channel_topic', + 'channel_purpose', + 'message_replied', +} + + +def _is_user_message(event): + if event.get('bot_id'): + return False + if event.get('subtype') in _SKIP_MESSAGE_SUBTYPES: + return False + return bool(event.get('user') and event.get('channel')) + def extract_body(event): if 'body' not in event: @@ -75,6 +94,13 @@ def run(body): InvocationType='Event', Payload=json.dumps(body) ) + + if event_type == 'message' and _is_user_message(event): + lambda_client.invoke( + FunctionName='automator-process-reaction', + InvocationType='Event', + Payload=json.dumps(body) + ) return { 'statusCode': 200, diff --git a/router/tests/test_router_lambda.py b/router/tests/test_router_lambda.py index c623c1f..68c6f33 100644 --- a/router/tests/test_router_lambda.py +++ b/router/tests/test_router_lambda.py @@ -86,13 +86,47 @@ def test_app_mention_routes_to_automator_for_any_user(self): body, ) - def test_message_event_is_ignored(self): + def test_user_message_routes_to_automator(self): body = { 'event': { 'type': 'message', 'user': 'U_NOT_ADMIN', 'channel': 'C06TEGTGM3J', 'text': 'hello', + 'ts': '1.0', + } + } + + lambda_function.run(body) + + self.assertEqual(len(fake_lambda_client.invocations), 1) + self.assertEqual( + fake_lambda_client.invocations[0]['FunctionName'], + 'automator-process-reaction', + ) + + def test_bot_message_is_ignored(self): + body = { + 'event': { + 'type': 'message', + 'subtype': 'bot_message', + 'bot_id': 'B123', + 'channel': 'C06TEGTGM3J', + 'text': 'hello', + } + } + + lambda_function.run(body) + + self.assertEqual(fake_lambda_client.invocations, []) + + def test_message_changed_is_ignored(self): + body = { + 'event': { + 'type': 'message', + 'subtype': 'message_changed', + 'user': 'U_NOT_ADMIN', + 'channel': 'C06TEGTGM3J', } }