From b6ae0f9b3b85685ee9daccfca92fd5eedcaebe0c Mon Sep 17 00:00:00 2001 From: Dita Aji Pratama Date: Sun, 2 Aug 2026 23:38:50 +0700 Subject: [PATCH] conversations.im anti-ban --- config.py | 1 + services/xmpp_client.py | 460 +++++++++++++++++++++++++++++++++++++--- 2 files changed, 427 insertions(+), 34 deletions(-) diff --git a/config.py b/config.py index 4e31950..6a6d5dc 100644 --- a/config.py +++ b/config.py @@ -102,6 +102,7 @@ AGENT_SKILLS = _yaml_get("agent", "skills", default="").strip().lower() XMPP_ENABLED = str(_yaml_get("xmpp", "enabled", default="false")).strip().lower() in ("true", "1", "yes") XMPP_MUC_ROOMS = _yaml_get("xmpp", "muc_rooms", default="").strip() XMPP_NICKNAME = _yaml_get("xmpp", "nickname", default="").strip() +XMPP_DEBUG = str(_yaml_get("xmpp", "debug", default="false")).strip().lower() in ("true", "1", "yes") XMPP_SELECTIVE_RESPONSE = str(_yaml_get("xmpp", "selective_response", default="true")).strip().lower() in ("true", "1", "yes") diff --git a/services/xmpp_client.py b/services/xmpp_client.py index 189357e..b2ada6b 100644 --- a/services/xmpp_client.py +++ b/services/xmpp_client.py @@ -1,7 +1,9 @@ import asyncio +import logging import random import signal import threading +import traceback from datetime import datetime from slixmpp import ClientXMPP @@ -14,30 +16,159 @@ from lib import personality from lib import ragroleplay # Anti-ban: delay constants for MUC rejoin behavior -MUC_REJOIN_INITIAL_DELAY = 5.0 # detik, delay awal sebelum rejoin -MUC_REJOIN_BACKOFF_MULT = 2.0 # multiplier exponential backoff -MUC_REJOIN_MAX_DELAY = 300.0 # detik, batas max backoff (5 menit) -MUC_REJOIN_COOLDOWN = 10.0 # detik, cooldown minimum antar rejoin attempt +MUC_REJOIN_INITIAL_DELAY = 30.0 # detik, delay awal sebelum rejoin (lebih gentle) +MUC_REJOIN_BACKOFF_MULT = 1.5 # multiplier exponential backoff (lebih gentle) +MUC_REJOIN_MAX_DELAY = 600.0 # detik, batas max backoff (10 menit) +MUC_REJOIN_COOLDOWN = 60.0 # detik, cooldown minimum antar rejoin attempt MUC_NICK_SUFFIX_MAX = 3 # max coba nick alternatif (anti-ban: jangan terlalu banyak) +MUC_JOIN_JITTER = 0.3 # jitter 30% untuk randomize delay + +# Anti-ban: Message queue rate limiting +MSG_QUEUE_DELAY_MIN = 3.0 # detik, delay minimum antar pesan keluar +MSG_QUEUE_DELAY_MAX = 8.0 # detik, delay maksimum antar pesan keluar +MSG_QUEUE_JITTER = 0.25 # jitter 25% untuk randomize delay +MSG_TYPING_SPEED_MIN = 8.0 # chars/sec, minimum typing speed +MSG_TYPING_SPEED_MAX = 20.0 # chars/sec, maksimum typing speed +MSG_READ_DELAY_MIN = 2.0 # detik, minimum read delay +MSG_READ_DELAY_MAX = 5.0 # detik, maksimum read delay + +# Anti-ban: Adaptive rate limiting (deteksi throttle dari server) +ADAPTIVE_WINDOW_SIZE = 20 # jumlah pesan terakhir untuk tracking +ADAPTIVE_ERROR_THRESHOLD = 0.15 # 15% error rate = server mulai throttle +ADAPTIVE_SLOW_THRESHOLD = 5.0 # detik, response time > ini = server lambat +ADAPTIVE_BACKOFF_MULT = 2.0 # multiplier delay saat throttle detected +ADAPTIVE_RECOVERY_TIME = 300 # detik, waktu tunggu sebelum recovery ke normal + +# Anti-ban: Connection behavior (sangat penting untuk conversations.im) +CONNECTION_KEEPALIVE = True # enable keepalive +CONNECTION_KEEPALIVE_INTERVAL = 300 # detik, ping interval (5 menit, jangan terlalu sering) +AUTO_RECONNECT_ENABLED = False # JANGAN auto reconnect instant (bot-like) +RECONNECT_DELAY_MIN = 30.0 # detik, delay minimum sebelum reconnect manual +RECONNECT_DELAY_MAX = 120.0 # detik, delay maksimum sebelum reconnect def _ts(): return datetime.now().strftime('%H:%M:%S') +def _setup_debug_logging(): + """Debug: aktifkan logging DEBUG slixmpp untuk menampilkan SEMUA stanza XML keluar/masuk. + + Sumber utama diagnosa: + - [slixmpp.xmlstream.xmlstream] SEND: -> stanza yang benar2 dikirim ke server + - [slixmpp.xmlstream.xmlstream] RECV: -> stanza yang diterima dari server + """ + if not config.XMPP_DEBUG: + return + logging.basicConfig( + level=logging.DEBUG, + format='[%(asctime)s][%(name)s] %(message)s', + datefmt='%H:%M:%S', + force=True, + ) + logging.getLogger('slixmpp').setLevel(logging.DEBUG) + print(f'[{_ts()}] DEBUG: XMPP debug logging ENABLED (semua stanza XML akan dicetak)', flush=True) + + +def _dbg(msg): + """Debug: cetak pesan debug hanya jika config.XMPP_DEBUG aktif.""" + if config.XMPP_DEBUG: + print(f'[{_ts()}] DEBUG: {msg}', flush=True) + + +def _add_jitter(delay: float, jitter_factor: float = MSG_QUEUE_JITTER) -> float: + """Tambahkan jitter acak ke delay untuk menghindari pola terdeteksi.""" + jitter = delay * jitter_factor + return delay + random.uniform(-jitter, jitter) + + def _typing_delay(text: str) -> float: """Hitung delay mengetik (detik) proporsional dengan panjang teks.""" char_count = len(text) if text else 0 - delay = char_count / config.TYPING_SPEED - return max(1.0, min(delay, config.TYPING_MAX)) + # Random typing speed untuk lebih human-like + typing_speed = random.uniform(MSG_TYPING_SPEED_MIN, MSG_TYPING_SPEED_MAX) + delay = char_count / typing_speed + # Clamp antara 1-15 detik, tambah jitter + return _add_jitter(max(1.0, min(delay, 15.0))) async def _read_delay(): """Delay simulasi membaca pesan user.""" - delay = random.uniform(config.READ_DELAY_MIN, config.READ_DELAY_MAX) + delay = random.uniform(MSG_READ_DELAY_MIN, MSG_READ_DELAY_MAX) await asyncio.sleep(delay) +class AdaptiveRateLimiter: + """Anti-ban: adaptive rate limiting yang mendeteksi throttle dari server.""" + + def __init__(self): + self._send_times: list[float] = [] # timestamp pengiriman pesan + self._errors: list[bool] = [] # apakah pengiriman error + self._response_times: list[float] = [] # waktu response server (jika ada) + self._throttle_detected = False + self._throttle_start: float | None = None + self._backoff_multiplier = 1.0 + + def record_send(self, success: bool, response_time: float | None = None): + """Catat hasil pengiriman pesan.""" + now = asyncio.get_event_loop().time() + self._send_times.append(now) + self._errors.append(not success) + + if response_time is not None: + self._response_times.append(response_time) + + # Keep only recent window + if len(self._send_times) > ADAPTIVE_WINDOW_SIZE: + self._send_times.pop(0) + self._errors.pop(0) + if len(self._response_times) > ADAPTIVE_WINDOW_SIZE: + self._response_times.pop(0) + + self._check_throttle() + + def _check_throttle(self): + """Cek apakah server mulai throttle.""" + if len(self._errors) < 5: # butuh minimal sample + return + + # Hitung error rate + error_rate = sum(self._errors) / len(self._errors) + + # Hitung avg response time + avg_response = 0 + if self._response_times: + avg_response = sum(self._response_times) / len(self._response_times) + + # Deteksi throttle: error rate tinggi ATAU response time lambat + was_throttled = self._throttle_detected + + if error_rate >= ADAPTIVE_ERROR_THRESHOLD or avg_response >= ADAPTIVE_SLOW_THRESHOLD: + if not self._throttle_detected: + self._throttle_detected = True + self._throttle_start = asyncio.get_event_loop().time() + self._backoff_multiplier = ADAPTIVE_BACKOFF_MULT + print(f'[{_ts()}] ADAPTIVE: Throttle detected! ' + f'error_rate={error_rate:.1%}, avg_response={avg_response:.1f}s, ' + f'backing off x{self._backoff_multiplier}', flush=True) + else: + # Recovery: cek apakah sudah cukup waktu untuk kembali normal + if self._throttle_detected and self._throttle_start: + elapsed = asyncio.get_event_loop().time() - self._throttle_start + if elapsed >= ADAPTIVE_RECOVERY_TIME: + self._throttle_detected = False + self._backoff_multiplier = 1.0 + print(f'[{_ts()}] ADAPTIVE: Recovered to normal rate', flush=True) + + def get_delay_multiplier(self) -> float: + """Dapatkan multiplier untuk delay saat ini.""" + return self._backoff_multiplier + + def is_throttled(self) -> bool: + """Apakah sedang dalam kondisi throttle.""" + return self._throttle_detected + + class XMPPClient(ClientXMPP): def __init__(self, jid, password, llm_client, tools_definition, TOOLS, TOOL_HANDLERS, build_system_prompt, agent_max_iterations, @@ -66,11 +197,51 @@ class XMPPClient(ClientXMPP): self._muc_rejoin_tasks: dict[str, asyncio.Task] = {} # room -> pending rejoin task self._muc_last_join: dict[str, datetime] = {} # room -> terakhir join (cooldown) - self.auto_reconnect = True + # Anti-ban: Message queue untuk rate limiting + self._msg_queue: asyncio.Queue | None = None + self._msg_worker_task: asyncio.Task | None = None + self._msg_worker_running = False + + # Anti-ban: Adaptive rate limiter + self._rate_limiter = AdaptiveRateLimiter() + + # Anti-ban: reconnect tracking (jangan instant reconnect) + self._reconnect_scheduled = False + self._last_disconnect: datetime | None = None - self.register_plugin('xep_0030') - self.register_plugin('xep_0045') - self.register_plugin('xep_0199') + # Anti-ban: JANGAN auto reconnect instant (bot-like behavior) + self.auto_reconnect = AUTO_RECONNECT_ENABLED + + # ── Anti-ban (conversations.im): ringkasan solusi yang membuat akun + # tidak dianggap spam/di-block (penyebab: klien default slixmpp + # mengiklankan identity `client/bot` + caps node slixmpp) ──────────── + # + # 1. JANGAN kirim 'subscribe' balik saat ada yang subscribe (contact + # farming = red flag). auto_authorize=True utk tetap menerima, + # auto_subscribe=False utk tidak membalas subscribe. + # 2. Caps node netral (bukan http://slixmpp.com/ver/X.Y.Z) supaya klien + # tidak mudah dikenali sebagai slixmpp. + # 3. Disco identity `client/console` di-set di _on_session_start (setelah + # bind). CATATAN: slixmpp menyimpan identity per key `boundjid.full` — + # di __init__ boundjid masih bare-JID, lookup runtime (setelah bind) + # memakai JID ber-resource, sehingga identity di __init__ TIDAK pernah + # ditemukan dan server melihat fallback `client/bot`. Harus di session_start. + self.roster.auto_authorize = True # terima subscribe masuk (sopan) + self.roster.auto_subscribe = False # JANGAN kirim subscribe balik + + # Anti-ban: register plugin dengan konfigurasi gentle + self.register_plugin('xep_0030') # Service Discovery + self.register_plugin('xep_0045') # MUC + self.register_plugin('xep_0199', { # XMPP Ping + 'keepalive': CONNECTION_KEEPALIVE, + 'interval': CONNECTION_KEEPALIVE_INTERVAL, + 'timeout': 30, + }) + # Muat plugin sekarang agar bisa set caps node sebelum konek + self.init_plugins() + + if self.plugin.get('xep_0115', None) is not None: + self.plugin['xep_0115'].caps_node = 'https://hendrik.local/caps' self.add_event_handler('session_start', self._on_session_start) self.add_event_handler('message', self._on_message) @@ -78,6 +249,11 @@ class XMPPClient(ClientXMPP): self.add_event_handler('disconnected', self._on_disconnected) self.add_event_handler('connected', self._on_connected) self.add_event_handler('groupchat_presence', self._on_muc_presence) + # Anti-ban: handler untuk melihat error dari server (perlu selalu aktif) + self.add_event_handler('message_error', self._on_message_error) + self.add_event_handler('stream_error', self._on_stream_error) + # Anti-ban: kontrol manual atas subscription request (jangan auto-subscribe balik) + self.add_event_handler('roster_subscription_request', self._on_roster_subscription_request) def _get_muc_nick(self, room: str) -> str: """Anti-ban: resolve nick untuk room, coba nick alternatif kalau conflict.""" @@ -89,10 +265,12 @@ class XMPPClient(ClientXMPP): return f"{base}_{suffix}" def _calc_rejoin_delay(self, room: str) -> float: - """Anti-ban: hitung delay rejoin dengan exponential backoff.""" + """Anti-ban: hitung delay rejoin dengan exponential backoff + jitter.""" attempts = self._muc_rejoin_attempts.get(room, 0) delay = MUC_REJOIN_INITIAL_DELAY * (MUC_REJOIN_BACKOFF_MULT ** attempts) - return min(delay, MUC_REJOIN_MAX_DELAY) + delay = min(delay, MUC_REJOIN_MAX_DELAY) + # Tambah jitter untuk menghindari pola terdeteksi + return _add_jitter(delay, MUC_JOIN_JITTER) def _schedule_muc_rejoin(self, room: str): """Anti-ban: schedule rejoin room dengan backoff & cooldown.""" @@ -144,11 +322,15 @@ class XMPPClient(ClientXMPP): # _muc_ready akan di-set oleh _on_muc_presence saat join berhasil self._muc_rejoin_attempts.pop(room, None) self._muc_rejoin_attempts.pop("_nick_" + room, None) + # Catat sukses untuk adaptive rate limiting + self._rate_limiter.record_send(success=True) print(f'[{_ts()}] MUC [{room}] Rejoin successful as {nick}', flush=True) except asyncio.CancelledError: print(f'[{_ts()}] MUC [{room}] Rejoin cancelled', flush=True) except Exception as e: print(f'[{_ts()}] MUC [{room}] Rejoin failed: {e}', flush=True) + # Catat error untuk adaptive rate limiting + self._rate_limiter.record_send(success=False) # Anti-ban: handle 409 Conflict - nick sudah dipakai orang lain if '409' in str(e) or 'conflict' in str(e).lower(): nick_attempts = self._muc_rejoin_attempts.get("_nick_" + room, 0) @@ -169,6 +351,10 @@ class XMPPClient(ClientXMPP): async def _on_connected(self, event): print(f'[{_ts()}] XMPP connected', flush=True) + _dbg(f'connected state: authenticated={self.authenticated}, bound={self.bound}, ' + f'sessionstarted={self.sessionstarted}, ' + f'_session_started={getattr(self, "_session_started", None)}') + _dbg(f'connected via: {self.transport}') async def _on_disconnected(self, event): print(f'[{_ts()}] XMPP disconnected', flush=True) @@ -178,12 +364,79 @@ class XMPPClient(ClientXMPP): task.cancel() print(f'[{_ts()}] MUC [{room}] Cancelled pending rejoin (disconnected)', flush=True) self._muc_rejoin_tasks.clear() + + # Anti-ban: schedule manual reconnect dengan delay (jangan instant) + if not self._stopped.is_set() and not self._reconnect_scheduled: + self._reconnect_scheduled = True + self._last_disconnect = datetime.now() + delay = random.uniform(RECONNECT_DELAY_MIN, RECONNECT_DELAY_MAX) + print(f'[{_ts()}] Will attempt reconnect in {delay:.0f}s...', flush=True) + asyncio.create_task(self._delayed_reconnect(delay)) + + async def _delayed_reconnect(self, delay: float): + """Anti-ban: reconnect dengan delay, bukan instant.""" + await asyncio.sleep(delay) + + if self._stopped.is_set(): + return + + print(f'[{_ts()}] Attempting manual reconnect...', flush=True) + try: + await self.connect() + except Exception as e: + print(f'[{_ts()}] Reconnect failed: {e}', flush=True) + # Schedule another reconnect with longer delay + self._reconnect_scheduled = False + self._on_disconnected(None) + else: + self._reconnect_scheduled = False async def _on_session_start(self, event): - self.send_presence() - self.get_roster() + _dbg('--- SESSION START ---') + _dbg(f'session state: authenticated={self.authenticated}, bound={self.bound}, ' + f'sessionstarted={self.sessionstarted}, ' + f'_session_started={getattr(self, "_session_started", None)}') + _dbg(f'boundjid: full={self.boundjid.full}, bare={self.boundjid.bare}, ' + f'resource={self.boundjid.resource}, host={self.boundjid.host}') print(f'[{_ts()}] XMPP online as {self.boundjid.full}', flush=True) - for room in self._muc_rooms: + + # Anti-ban: set disco identity SEKARANG (setelah bind, boundjid ber-resource) + # supaya identity masuk ke caps ver (presence) DAN respon disco#info server. + # Sebelumnya di __init__ -> tersimpan di key bare-JID -> tak pernah ditemukan + # saat runtime -> server melihat identity fallback `client/bot`. + self.plugin['xep_0030'].add_identity( + category='client', itype='console', name=config.AGENT_CHARACTER.title() or 'Hendrik', + ) + try: + local_info = await self.plugin['xep_0030'].get_info(local=True) + if 'disco_info' in local_info: + identities = local_info['disco_info']['identities'] + else: + identities = local_info['identities'] + _dbg(f'disco identity (local) after set: {identities}') + except Exception as e: + _dbg(f'disco identity check failed: {e}') + + # Anti-ban: send presence + request roster (wajib sebelum kirim pesan DM) + print(f'[{_ts()}] Sending initial presence...', flush=True) + self.send_presence() + print(f'[{_ts()}] Requesting roster...', flush=True) + self.get_roster() + + # Anti-ban: delay sebelum join MUC pertama agar startup tidak terlihat bot + if self._muc_rooms: + pre_delay = random.uniform(5.0, 15.0) + print(f'[{_ts()}] MUC pre-join delay {pre_delay:.1f}s (anti-ban)...', flush=True) + await asyncio.sleep(pre_delay) + + # Anti-ban: delay sebelum join MUC untuk menghindari koneksi yang terlalu agresif + for i, room in enumerate(self._muc_rooms): + # Delay antar room join (3-8 detik per room) + if i > 0: + join_delay = random.uniform(3.0, 8.0) + print(f'[{_ts()}] MUC [{room}] Waiting {join_delay:.1f}s before join...', flush=True) + await asyncio.sleep(join_delay) + # Anti-ban: retry join dengan incremental delay & nick fallback success = False for attempt in range(1, 4): @@ -212,8 +465,8 @@ class XMPPClient(ClientXMPP): print(f'[{_ts()}] MUC [{room}] All nick variations exhausted', flush=True) break elif attempt < 3: - # Anti-ban: error biasa, wait before retry (2s, 4s) - retry_delay = 2.0 * attempt + # Anti-ban: error biasa, wait before retry (5s, 10s, 15s) + retry_delay = 5.0 * attempt print(f'[{_ts()}] MUC [{room}] Retrying in {retry_delay:.0f}s...', flush=True) await asyncio.sleep(retry_delay) if not success: @@ -229,7 +482,36 @@ class XMPPClient(ClientXMPP): if not body: return print(f'[{_ts()}] DM from {jid}: {body[:60]}', flush=True) - threading.Thread(target=self._process_dm, args=(jid, body), daemon=True).start() + # Anti-ban: proses langsung di event loop, bukan thread baru + # Ini memastikan msg.send() dipanggil dari thread yang benar + asyncio.create_task(self._process_dm_async(jid, body)) + + def _on_message_error(self, msg): + """Anti-ban: handle error message dari server.""" + print(f'[{_ts()}] MESSAGE ERROR from server:', flush=True) + print(f'[{_ts()}] Type: {msg.get("type", "unknown")}', flush=True) + print(f'[{_ts()}] From: {msg.get("from", "unknown")}', flush=True) + print(f'[{_ts()}] To: {msg.get("to", "unknown")}', flush=True) + print(f'[{_ts()}] Error: {msg.get("error", "unknown")}', flush=True) + print(f'[{_ts()}] Full stanza: {msg}', flush=True) + + # Catat error untuk adaptive rate limiting + self._rate_limiter.record_send(success=False) + + def _on_stream_error(self, error): + """Anti-ban: handle stream error dari server.""" + print(f'[{_ts()}] STREAM ERROR from server:', flush=True) + print(f'[{_ts()}] Condition: {error.get("condition", "unknown")}', flush=True) + print(f'[{_ts()}] Text: {error.get("text", "unknown")}', flush=True) + print(f'[{_ts()}] Full error: {error}', flush=True) + + async def _on_roster_subscription_request(self, presence): + print(f'[{_ts()}] Subscription request from {presence["from"]}', flush=True) + # Anti-ban: TIDAK merespons subscribe otomatis (auto_subscribe=False). + # Balasan 'subscribe' balik adalah red flag (contact farming) di conversations.im. + _dbg(f'NOT auto-responding subscribe from {presence["from"]} ' + f'(auto_subscribe={self.roster.auto_subscribe}, ' + f'auto_authorize={self.roster.auto_authorize})') def _on_groupchat_message(self, msg): if msg['type'] != 'groupchat': @@ -245,7 +527,8 @@ class XMPPClient(ClientXMPP): if not body: return print(f'[{_ts()}] MUC [{room}] <{nick}>: {body[:60]}', flush=True) - threading.Thread(target=self._process_muc, args=(room, nick, body), daemon=True).start() + # Anti-ban: proses langsung di event loop, bukan thread baru + asyncio.create_task(self._process_muc_async(room, nick, body)) def _is_my_nick(self, room: str, nick: str) -> bool: """Anti-ban: cek apakah nick yang dimasukan sesuai dengan nick bot di room.""" @@ -279,7 +562,8 @@ class XMPPClient(ClientXMPP): else: print(f'[{_ts()}] MUC [{room}] <{nick}> joined (type={ptype})', flush=True) - def _process_dm(self, jid, body): + async def _process_dm_async(self, jid, body): + """Anti-ban: versi async dari _process_dm, dipanggil dari event loop.""" session = self._session_mgr.get_or_create( jid, self._build_system_prompt( tools_definition=self._tools_def, @@ -289,7 +573,10 @@ class XMPPClient(ClientXMPP): ) session.cancel_timer() - self.send_presence_subscription(pto=jid, ptype='subscribed') + # Anti-ban: JANGAN kirim presence subscription otomatis + # Ini adalah red flag untuk server (contact farming behavior) + # Hanya kirim jika user explicitly request + # self.send_presence_subscription(pto=jid, ptype='subscribed') if body == ':new': self._session_mgr.reset(jid) @@ -334,8 +621,7 @@ class XMPPClient(ClientXMPP): self._schedule_send(jid, f'> {body}\nThinking...') # Delay 1: simulasi membaca pesan user - if self._loop and not self._loop.is_closed(): - asyncio.run_coroutine_threadsafe(_read_delay(), self._loop) + await _read_delay() my_name = personality.PERSONALITY.name quote = body @@ -396,7 +682,8 @@ class XMPPClient(ClientXMPP): # DM: timeout 24 jam (efektif tidak auto-close), MUC tetap 5 menit session.start_timer(86400, self._timeout_session, jid, 'chat') - def _process_muc(self, room, nick, body): + async def _process_muc_async(self, room, nick, body): + """Anti-ban: versi async dari _process_muc, dipanggil dari event loop.""" session = self._session_mgr.get_or_create( room, self._build_system_prompt( tools_definition=self._tools_def, @@ -419,8 +706,7 @@ class XMPPClient(ClientXMPP): self._schedule_send(room, f'> [{nick}] {body}\nThinking...', mtype='groupchat') # Delay 1: simulasi membaca pesan user - if self._loop and not self._loop.is_closed(): - asyncio.run_coroutine_threadsafe(_read_delay(), self._loop) + await _read_delay() my_name = personality.PERSONALITY.name quote = f'[{nick}] {body}' @@ -482,12 +768,89 @@ class XMPPClient(ClientXMPP): return execute_tool(tool_call, self._TOOL_HANDLERS) def _schedule_send(self, to, body, mtype='chat'): - if self._loop and not self._loop.is_closed(): - asyncio.run_coroutine_threadsafe( - self._send_coro(to, body, mtype), self._loop - ) + """Anti-ban: enqueue pesan ke queue, bukan langsung kirim.""" + if self._msg_queue and self._loop and not self._loop.is_closed(): + try: + self._msg_queue.put_nowait((to, body, mtype)) + print(f'[{_ts()}] Queued message to {to} ({len(body)} chars)', flush=True) + except asyncio.QueueFull: + print(f'[{_ts()}] WARNING: Message queue full, dropping message to {to}', flush=True) else: - print(f'[{_ts()}] WARNING: cannot send to {to} — loop unavailable', flush=True) + print(f'[{_ts()}] WARNING: cannot queue message to {to} — queue unavailable', flush=True) + + async def _msg_worker(self): + """Anti-ban: worker coroutine yang mengirim pesan dari queue dengan rate limiting.""" + print(f'[{_ts()}] Message queue worker started', flush=True) + self._msg_worker_running = True + + while self._msg_worker_running: + try: + # Ambil pesan dari queue dengan timeout + try: + to, body, mtype = await asyncio.wait_for( + self._msg_queue.get(), timeout=1.0 + ) + except asyncio.TimeoutError: + continue + + # Adaptive: apply backoff multiplier jika throttle detected + base_delay = random.uniform(MSG_QUEUE_DELAY_MIN, MSG_QUEUE_DELAY_MAX) + multiplier = self._rate_limiter.get_delay_multiplier() + delay = _add_jitter(base_delay) * multiplier + + if multiplier > 1.0: + print(f'[{_ts()}] ADAPTIVE: Throttled, delay {delay:.1f}s (x{multiplier}) to {to}', flush=True) + else: + print(f'[{_ts()}] Pre-send delay: {delay:.1f}s to {to}', flush=True) + + await asyncio.sleep(delay) + + # Kirim pesan dan catat hasilnya + send_start = asyncio.get_event_loop().time() + send_success = False + + try: + # Anti-ban: gunakan full JID untuk mfrom (server strict tentang ini) + mfrom = self.boundjid.full if self.boundjid else None + + msg = self.make_message(mto=to, mbody=body, mtype=mtype, mfrom=mfrom) + + # Debug: snapshot state koneksi + stanza XML lengkap sebelum kirim. + _dbg(f'SEND STATE: to={to}, mtype={mtype}, ' + f'_session_started={getattr(self, "_session_started", None)}, ' + f'authenticated={self.authenticated}, bound={self.bound}') + _dbg(f'Stanza to send (full): {msg}') + + msg.send() + send_success = True + print(f'[{_ts()}] Sent to {to} ({len(body)} chars)', flush=True) + except Exception as e: + print(f'[{_ts()}] SEND ERROR to {to}: {e}', flush=True) + print(traceback.format_exc(), flush=True) + # Catat error untuk adaptive rate limiting + self._rate_limiter.record_send(success=False) + + if send_success: + # Catat sukses (tanpa response time karena XMPP async) + self._rate_limiter.record_send(success=True) + + # Mark task done + self._msg_queue.task_done() + + # Small delay antara pesan untuk menghindari burst + # Adaptive: lebih lama jika throttled + inter_delay = random.uniform(0.5, 1.5) * multiplier + await asyncio.sleep(inter_delay) + + except asyncio.CancelledError: + print(f'[{_ts()}] Message queue worker cancelled', flush=True) + break + except Exception as e: + print(f'[{_ts()}] Message queue worker error: {e}', flush=True) + await asyncio.sleep(1) + + self._msg_worker_running = False + print(f'[{_ts()}] Message queue worker stopped', flush=True) async def _send_coro(self, to, body, mtype): try: @@ -496,7 +859,9 @@ class XMPPClient(ClientXMPP): print(f'[{_ts()}] Typing delay: {delay:.1f}s ({len(body)} chars)', flush=True) await asyncio.sleep(delay) - msg = self.make_message(mto=to, mbody=body, mtype=mtype) + # Anti-ban: gunakan full JID untuk mfrom (server strict tentang ini) + mfrom = self.boundjid.full if self.boundjid else None + msg = self.make_message(mto=to, mbody=body, mtype=mtype, mfrom=mfrom) msg.send() except Exception as e: print(f'[{_ts()}] SEND ERROR: {e}', flush=True) @@ -508,12 +873,17 @@ class XMPPClient(ClientXMPP): def start(self): print(f'[{_ts()}] Starting XMPP service...', flush=True) + _setup_debug_logging() asyncio.run(self._run()) async def _run(self): self._stopped = asyncio.Event() self._loop = asyncio.get_running_loop() + # Anti-ban: inisialisasi message queue dan worker + self._msg_queue = asyncio.Queue(maxsize=100) + self._msg_worker_task = asyncio.create_task(self._msg_worker()) + # Hanya tangani SIGTERM untuk shutdown. # SENGATKAN SIGHUP: nohup kirim SIGHUP saat terminal close, # dan kita tidak mau proses mati karena itu. @@ -522,7 +892,14 @@ class XMPPClient(ClientXMPP): except (NotImplementedError, RuntimeError): pass - await self.connect() + print(f'[{_ts()}] Connecting to server (jid={self.jid})...', flush=True) + try: + await self.connect() + except Exception as e: + print(f'[{_ts()}] CONNECT ERROR: {e}', flush=True) + print(traceback.format_exc(), flush=True) + raise + print(f'[{_ts()}] Connected, waiting for stream events...', flush=True) try: await self._stopped.wait() except (asyncio.CancelledError, KeyboardInterrupt): @@ -535,4 +912,19 @@ class XMPPClient(ClientXMPP): asyncio.run_coroutine_threadsafe(self._async_stop(), self._loop) async def _async_stop(self): + # Anti-ban: stop message queue worker dulu + if self._msg_worker_task and not self._msg_worker_task.done(): + self._msg_worker_running = False + self._msg_worker_task.cancel() + try: + await self._msg_worker_task + except asyncio.CancelledError: + pass + + # Cancel semua pending rejoin tasks + for room, task in list(self._muc_rejoin_tasks.items()): + if not task.done(): + task.cancel() + self._muc_rejoin_tasks.clear() + self._stopped.set()