#!/usr/bin/env python3 """ Signal Spammer Auto-Removal System Monitors a honeypot Signal group chat, detects spammers who message there, and automatically removes them from all groups you admin. Usage: python spam_remover.py # Daemon mode (SSE listener) python spam_remover.py --ban NUM # Manually ban a number python spam_remover.py --dry-run # Preview without changes python spam_remover.py --once # Single check and exit """ import argparse import json import logging import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional import requests import sseclient from pysignalclijsonrpc import SignalCliJSONRPCApi class SpammerRemover: def __init__(self, config_path: str = "config.json"): self.config = self._load_config(config_path) self.banned_list_path = Path(self.config["banned_list_path"]) self.banned_list = self._load_banned_list() self._setup_logging() self.api = SignalCliJSONRPCApi( endpoint=self.config["signal_cli_endpoint"], account=self.config["account"], ) self.logger.info( f"Initialized. Honeypot: {self.config['honeypot_group_id'][:12]}..." ) def _load_config(self, config_path: str) -> dict[str, Any]: path = Path(config_path) if not path.exists(): print(f"Error: Config file not found: {config_path}") print("Create config.json from the template first.") sys.exit(1) with open(path) as f: return json.load(f) def _setup_logging(self): log_file = self.config.get("log_file", "spam_remover.log") logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(log_file), logging.StreamHandler(sys.stdout), ], ) self.logger = logging.getLogger(__name__) def _load_banned_list(self) -> dict[str, Any]: if self.banned_list_path.exists(): with open(self.banned_list_path) as f: return json.load(f) return {} def _save_banned_list(self): with open(self.banned_list_path, "w") as f: json.dump(self.banned_list, f, indent=2) def is_banned(self, number: str) -> bool: return number in self.banned_list def ban_number(self, number: str, reason: str = "honeypot_message"): if self.is_banned(number): self.logger.info(f"Already banned: {number}") return self.banned_list[number] = { "banned_at": datetime.now(timezone.utc).isoformat(), "reason": reason, "removed_from": [], } self._save_banned_list() self.logger.info(f"Banned: {number}") def get_groups(self) -> list[dict[str, Any]]: try: return self.api.list_groups() except Exception as e: self.logger.error(f"Failed to list groups: {e}") return [] def get_admin_groups(self) -> list[dict[str, Any]]: groups = self.get_groups() admin_groups = [] for group in groups: if not group.get("isMember", False): continue admins = group.get("admins", []) members = group.get("members", []) account = self.config["account"] if account in admins or account in members: if account in admins: admin_groups.append(group) return admin_groups def remove_from_group(self, group_id: str, group_name: str, number: str) -> bool: try: group = self.api.get_group(group_id) members = group.get("members", []) member_numbers = [] for m in members: if isinstance(m, dict): member_numbers.append(m.get("number", m.get("uuid", ""))) else: member_numbers.append(str(m)) if number not in member_numbers: self.logger.info(f"{number} not in group {group_name}, skipping") return True updated_members = [m for m in member_numbers if m != number] self.api.update_group( name=group.get("name", ""), members=updated_members, admins=group.get("admins", []), description=group.get("description", ""), group_link=group.get("groupInviteLink", "disabled"), ) self.logger.info(f"Removed {number} from {group_name}") return True except Exception as e: self.logger.error(f"Failed to remove from {group_name}: {e}") return False def send_notification(self, message: str): try: self.api.send_message( recipients=[self.config["notify_self_number"]], message=message, ) self.logger.info(f"Notification sent: {message[:50]}...") except Exception as e: self.logger.error(f"Failed to send notification: {e}") def process_spammer(self, number: str, dry_run: bool = False) -> dict[str, Any]: result = { "number": number, "already_banned": self.is_banned(number), "groups_removed": [], "groups_failed": [], "dry_run": dry_run, } if self.is_banned(number): self.logger.info(f"Already banned: {number}") return result if dry_run: self.logger.info(f"[DRY RUN] Would ban: {number}") admin_groups = self.get_admin_groups() for group in admin_groups: gid = group.get("id", "") gname = group.get("name", "Unknown") if gid == self.config["honeypot_group_id"]: continue if gid in self.config.get("excluded_groups", []): continue result["groups_removed"].append(gname) return result self.ban_number(number) admin_groups = self.get_admin_groups() removed_count = 0 for group in admin_groups: gid = group.get("id", "") gname = group.get("name", "Unknown") if gid == self.config["honeypot_group_id"]: continue if gid in self.config.get("excluded_groups", []): continue if self.remove_from_group(gid, gname, number): result["groups_removed"].append(gname) removed_count += 1 else: result["groups_failed"].append(gname) if removed_count > 0: self.banned_list[number]["removed_from"] = result["groups_removed"] self._save_banned_list() notification = ( f"Spammer removed: {number}\n" f"Removed from {removed_count} group(s): {', '.join(result['groups_removed'])}" ) self.send_notification(notification) if result["groups_failed"]: self.logger.warning( f"Failed to remove from {len(result['groups_failed'])} groups" ) return result def handle_message(self, envelope: dict): source = envelope.get("sourceNumber") or envelope.get("source", "") if not source: return group_info = envelope.get("groupInfo", {}) group_id = group_info.get("groupId", "") if group_id != self.config["honeypot_group_id"]: return if self.is_banned(source): return self.logger.info(f"Honeypot message from: {source}") result = self.process_spammer(source) self.logger.info( f"Processed {source}: removed from {len(result['groups_removed'])} groups" ) def daemon_mode(self): self.logger.info("Starting daemon mode (SSE listener)...") endpoint = self.config["signal_cli_endpoint"] url = f"{endpoint}/api/v1/events" while True: try: self.logger.info(f"Connecting to SSE: {url}") response = requests.get(url, stream=True, timeout=60) response.raise_for_status() client = sseclient.SSEClient(response) for event in client.events(): if event.event == "message": try: data = json.loads(event.data) envelope = data.get("envelope", {}) self.handle_message(envelope) except json.JSONDecodeError: self.logger.warning(f"Invalid JSON: {event.data[:100]}") except Exception as e: self.logger.error(f"Error processing event: {e}") except requests.exceptions.ConnectionError: self.logger.warning("Connection lost. Reconnecting in 5s...") time.sleep(5) except requests.exceptions.Timeout: self.logger.debug("SSE timeout, reconnecting...") except Exception as e: self.logger.error(f"SSE error: {e}. Reconnecting in 5s...") time.sleep(5) def once_mode(self): self.logger.info("Running single check...") admin_groups = self.get_admin_groups() self.logger.info(f"Found {len(admin_groups)} admin'd groups") for group in admin_groups: gid = group.get("id", "") gname = group.get("name", "Unknown") members = group.get("members", []) member_count = len(members) if isinstance(members, list) else 0 self.logger.info(f" {gname}: {member_count} members") def manual_mode(self, number: str): self.logger.info(f"Manual ban: {number}") result = self.process_spammer(number) if result["already_banned"]: print(f"Already banned: {number}") else: print(f"Banned: {number}") if result["groups_removed"]: print(f"Removed from: {', '.join(result['groups_removed'])}") if result["groups_failed"]: print(f"Failed to remove from: {', '.join(result['groups_failed'])}") def list_banned(self): if not self.banned_list: print("No banned numbers.") return print(f"Banned numbers ({len(self.banned_list)}):") for number, info in self.banned_list.items(): banned_at = info.get("banned_at", "unknown") removed = info.get("removed_from", []) print(f" {number} (banned {banned_at}) - removed from {len(removed)} groups") def main(): parser = argparse.ArgumentParser( description="Signal Spammer Auto-Removal System" ) parser.add_argument( "--config", default="config.json", help="Path to config file (default: config.json)", ) parser.add_argument( "--ban", metavar="NUMBER", help="Manually ban a phone number", ) parser.add_argument( "--dry-run", action="store_true", help="Preview what would be done without making changes", ) parser.add_argument( "--once", action="store_true", help="Single check and exit", ) parser.add_argument( "--list-banned", action="store_true", help="List all banned numbers", ) args = parser.parse_args() remover = SpammerRemover(config_path=args.config) if args.list_banned: remover.list_banned() elif args.ban: remover.manual_mode(args.ban) elif args.once: remover.once_mode() else: remover.daemon_mode() if __name__ == "__main__": main()