350 lines
12 KiB
Python
Executable File
350 lines
12 KiB
Python
Executable File
#!/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()
|