From 9fece911b58be49799ab73c4c37a6e49d22f35cf Mon Sep 17 00:00:00 2001 From: frdel <38891707+frdel@users.noreply.github.com> Date: Tue, 31 Mar 2026 09:20:35 +0200 Subject: [PATCH] refactor: centralize WhatsApp storage paths and improve bridge dependency handling - Add storage_paths.py helper for consistent session/media/runtime paths - Replace hardcoded usr/whatsapp paths across all modules - Fix bridge lock to be event-loop-aware (recreate per loop) - Add automatic dependency reinstall on startup failures - Track bridge startup output for better error diagnostics - Add dependency state tracking with package.json hash validation - Implement force reinstall when node_modules appears --- plugins/_whatsapp_integration/README.md | 2 +- .../_whatsapp_integration/api/disconnect.py | 7 +- plugins/_whatsapp_integration/api/qr_code.py | 10 +- plugins/_whatsapp_integration/api/start.py | 11 +- .../extensions/python/job_loop/_10_wa_poll.py | 10 +- .../helpers/bridge_manager.py | 464 ++++++++++++++---- .../_whatsapp_integration/helpers/handler.py | 26 +- .../helpers/number_utils.py | 32 ++ .../helpers/storage_paths.py | 11 + .../_whatsapp_integration/webui/config.html | 12 +- .../whatsapp-bridge/bridge.js | 7 +- tests/test_whatsapp_bridge_manager.py | 181 +++++++ tests/test_whatsapp_number_utils.py | 36 ++ tests/test_whatsapp_storage_paths.py | 18 + 14 files changed, 680 insertions(+), 147 deletions(-) create mode 100644 plugins/_whatsapp_integration/helpers/number_utils.py create mode 100644 plugins/_whatsapp_integration/helpers/storage_paths.py create mode 100644 tests/test_whatsapp_bridge_manager.py create mode 100644 tests/test_whatsapp_number_utils.py create mode 100644 tests/test_whatsapp_storage_paths.py diff --git a/plugins/_whatsapp_integration/README.md b/plugins/_whatsapp_integration/README.md index cb53d6bd8..941a08fd7 100644 --- a/plugins/_whatsapp_integration/README.md +++ b/plugins/_whatsapp_integration/README.md @@ -25,7 +25,7 @@ Dependencies are auto-installed on first bridge start if missing. 3. Click Show QR Code and scan with WhatsApp on your phone 4. Send a message from an allowed number to start a chat -The WhatsApp session persists across restarts in `usr/whatsapp/sessions/`. No re-pairing needed unless you disconnect via settings. +The WhatsApp session persists across restarts in `usr/whatsapp/session/`. No re-pairing needed unless you disconnect via settings. ## Configuration diff --git a/plugins/_whatsapp_integration/api/disconnect.py b/plugins/_whatsapp_integration/api/disconnect.py index 243fa1e68..0205d0d08 100644 --- a/plugins/_whatsapp_integration/api/disconnect.py +++ b/plugins/_whatsapp_integration/api/disconnect.py @@ -1,10 +1,10 @@ """Disconnect WhatsApp account by stopping bridge and clearing session.""" +import os import shutil from helpers.api import ApiHandler, Request from helpers.errors import format_error -from helpers import files class Disconnect(ApiHandler): @@ -12,13 +12,14 @@ class Disconnect(ApiHandler): async def process(self, input: dict, request: Request) -> dict: try: from plugins._whatsapp_integration.helpers import bridge_manager + from plugins._whatsapp_integration.helpers.storage_paths import get_bridge_session_dir # Stop bridge first await bridge_manager.stop_bridge() # Delete session files - session_dir = files.get_abs_path("usr/whatsapp/sessions") - if files.exists(session_dir): + session_dir = get_bridge_session_dir() + if os.path.exists(session_dir): shutil.rmtree(session_dir, ignore_errors=True) return {"success": True, "message": "Account disconnected"} diff --git a/plugins/_whatsapp_integration/api/qr_code.py b/plugins/_whatsapp_integration/api/qr_code.py index 9171e0c2b..32953974d 100644 --- a/plugins/_whatsapp_integration/api/qr_code.py +++ b/plugins/_whatsapp_integration/api/qr_code.py @@ -2,7 +2,7 @@ from helpers.api import ApiHandler, Request from helpers.errors import format_error -from helpers import files, plugins +from helpers import plugins PLUGIN_NAME = "_whatsapp_integration" @@ -13,8 +13,6 @@ class QrCode(ApiHandler): async def process(self, input: dict, request: Request) -> dict: config = plugins.get_plugin_config(PLUGIN_NAME) or {} port = int(config.get("bridge_port", 3100)) - session_dir = files.get_abs_path("usr/whatsapp/sessions") - cache_dir = files.get_abs_path("usr/whatsapp/media") mode = config.get("mode", "dedicated") from plugins._whatsapp_integration.helpers.bridge_manager import ( @@ -22,8 +20,14 @@ class QrCode(ApiHandler): get_bridge_url, is_process_alive, ) + from plugins._whatsapp_integration.helpers.storage_paths import ( + get_bridge_media_dir, + get_bridge_session_dir, + ) from plugins._whatsapp_integration.helpers.wa_client import get_qr + session_dir = get_bridge_session_dir() + cache_dir = get_bridge_media_dir() base_url = get_bridge_url(port) # Start bridge if not running diff --git a/plugins/_whatsapp_integration/api/start.py b/plugins/_whatsapp_integration/api/start.py index a3448cac3..1dd6d3c19 100644 --- a/plugins/_whatsapp_integration/api/start.py +++ b/plugins/_whatsapp_integration/api/start.py @@ -2,7 +2,7 @@ from helpers.api import ApiHandler, Request from helpers.errors import format_error -from helpers import files, plugins +from helpers import plugins PLUGIN_NAME = "_whatsapp_integration" @@ -13,14 +13,19 @@ class Start(ApiHandler): async def process(self, input: dict, request: Request) -> dict: config = plugins.get_plugin_config(PLUGIN_NAME) or {} port = int(config.get("bridge_port", 3100)) - session_dir = files.get_abs_path("usr/whatsapp/sessions") - cache_dir = files.get_abs_path("usr/whatsapp/media") mode = config.get("mode", "dedicated") from plugins._whatsapp_integration.helpers.bridge_manager import ( ensure_bridge_http_up, is_process_alive, ) + from plugins._whatsapp_integration.helpers.storage_paths import ( + get_bridge_media_dir, + get_bridge_session_dir, + ) + + session_dir = get_bridge_session_dir() + cache_dir = get_bridge_media_dir() if is_process_alive(): return {"success": True, "message": "Bridge already running"} diff --git a/plugins/_whatsapp_integration/extensions/python/job_loop/_10_wa_poll.py b/plugins/_whatsapp_integration/extensions/python/job_loop/_10_wa_poll.py index 13449c621..d3f224a92 100644 --- a/plugins/_whatsapp_integration/extensions/python/job_loop/_10_wa_poll.py +++ b/plugins/_whatsapp_integration/extensions/python/job_loop/_10_wa_poll.py @@ -6,7 +6,7 @@ from typing import Any from helpers.extension import Extension from helpers.errors import format_error from helpers.print_style import PrintStyle -from helpers import files, plugins +from helpers import plugins PLUGIN_NAME: str = "_whatsapp_integration" @@ -44,6 +44,10 @@ class WhatsAppAutoPoll(Extension): async def _poll_loop() -> None: from plugins._whatsapp_integration.helpers import bridge_manager from plugins._whatsapp_integration.helpers.handler import poll_messages + from plugins._whatsapp_integration.helpers.storage_paths import ( + get_bridge_media_dir, + get_bridge_session_dir, + ) bridge_started = False consecutive_failures = 0 @@ -57,8 +61,8 @@ async def _poll_loop() -> None: break port = int(config.get("bridge_port", 3100)) - session_dir = files.get_abs_path("usr/whatsapp/sessions") - cache_dir = files.get_abs_path("usr/whatsapp/media") + session_dir = get_bridge_session_dir() + cache_dir = get_bridge_media_dir() mode = config.get("mode", "dedicated") # Detect config changes that require bridge restart diff --git a/plugins/_whatsapp_integration/helpers/bridge_manager.py b/plugins/_whatsapp_integration/helpers/bridge_manager.py index e3c60263c..ee5e8d878 100644 --- a/plugins/_whatsapp_integration/helpers/bridge_manager.py +++ b/plugins/_whatsapp_integration/helpers/bridge_manager.py @@ -4,19 +4,37 @@ WhatsApp bridge subprocess manager. No agent/tool dependencies. """ +from __future__ import annotations + import asyncio +import hashlib +import json import os import platform +import shutil import subprocess import threading +from collections import deque from pathlib import Path +from typing import Any, Sequence from helpers.print_style import PrintStyle -_bridge_lock = asyncio.Lock() +_bridge_lock: asyncio.Lock | None = None +_bridge_lock_loop: asyncio.AbstractEventLoop | None = None _bridge_config: dict = {} # config the running bridge was started with +MAX_STARTUP_LOG_LINES = 80 +STARTUP_WAIT_ATTEMPTS = 20 +STARTUP_WAIT_SECONDS = 0.5 +DEPENDENCY_FAILURE_MARKERS = ( + "ERR_MODULE_NOT_FOUND", + "MODULE_NOT_FOUND", + "Cannot find module", + "Cannot find package", +) + # ------------------------------------------------------------------ # Process wrapper with destructor @@ -28,6 +46,7 @@ class _BridgeProcess: def __init__(self, process: subprocess.Popen, port: int): self._process = process self._port = port + self._recent_output: deque[str] = deque(maxlen=MAX_STARTUP_LOG_LINES) def poll(self) -> int | None: return self._process.poll() @@ -41,6 +60,12 @@ class _BridgeProcess: def kill(self) -> None: self._process.kill() + def remember_output(self, text: str) -> None: + self._recent_output.append(text) + + def recent_output(self) -> str: + return "\n".join(self._recent_output) + @property def stdout(self): return self._process.stdout @@ -61,8 +86,15 @@ class _BridgeProcess: _bridge_process: _BridgeProcess | None = None +REPO_ROOT = Path(__file__).resolve().parents[3] BRIDGE_DIR = str(Path(__file__).parent.parent / "whatsapp-bridge") BRIDGE_SCRIPT = os.path.join(BRIDGE_DIR, "bridge.js") +BRIDGE_PACKAGE_JSON = os.path.join(BRIDGE_DIR, "package.json") +BRIDGE_PACKAGE_LOCK = os.path.join(BRIDGE_DIR, "package-lock.json") +BRIDGE_RUNTIME_DIR = os.path.join(REPO_ROOT, "usr", "whatsapp", "bridge-runtime") +BRIDGE_INSTALL_STATE = os.path.join(BRIDGE_RUNTIME_DIR, "deps-state.json") +BRIDGE_NPM_CACHE = os.path.join(BRIDGE_RUNTIME_DIR, "npm-cache") +NODE_MODULES_DIR = os.path.join(BRIDGE_DIR, "node_modules") # ------------------------------------------------------------------ @@ -75,63 +107,20 @@ async def start_bridge( cache_dir: str, mode: str = "dedicated", ) -> bool: - global _bridge_process - async with _bridge_lock: - # If process is alive, keep it (even if WA not yet connected — may be pairing) - if _bridge_process and _bridge_process.poll() is None: - return True - - await _ensure_npm_install() - - cmd = [ - "node", BRIDGE_SCRIPT, - "--port", str(port), - "--session", session_dir, - "--cache-dir", cache_dir, - "--mode", mode, - ] - - _kill_port_process(port) - PrintStyle.info("WhatsApp: starting bridge") - _bridge_process = _BridgeProcess(subprocess.Popen( - cmd, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - cwd=BRIDGE_DIR, - ), port) - _start_log_reader(_bridge_process) - _bridge_config.clear() - _bridge_config.update({"port": port, "mode": mode}) - - # Wait for bridge to become healthy - for _ in range(20): - await asyncio.sleep(0.5) - if _bridge_process.poll() is not None: - PrintStyle.error("WhatsApp: bridge process exited unexpectedly") - return False - if await _check_health(port): - return True - - PrintStyle.warning("WhatsApp: bridge started but not yet connected") - return True + async with _get_bridge_lock(): + return await _ensure_bridge_started( + port=port, + session_dir=session_dir, + cache_dir=cache_dir, + mode=mode, + require_connection=True, + start_label="WhatsApp: starting bridge", + ) async def stop_bridge() -> None: - global _bridge_process - async with _bridge_lock: - if not _bridge_process: - return - try: - _bridge_process.terminate() - try: - _bridge_process.wait(timeout=5) - except subprocess.TimeoutExpired: - _bridge_process.kill() - except Exception: - pass - _bridge_process = None - _bridge_config.clear() - PrintStyle.info("WhatsApp: bridge stopped") + async with _get_bridge_lock(): + _stop_bridge_process() async def is_bridge_running(port: int) -> bool: @@ -151,43 +140,15 @@ async def ensure_bridge_http_up( mode: str = "dedicated", ) -> bool: """Start bridge if needed and wait for HTTP server only (not WA connection).""" - global _bridge_process - async with _bridge_lock: - if _bridge_process and _bridge_process.poll() is None: - if await _check_http_up(port): - return True - - await _ensure_npm_install() - - cmd = [ - "node", BRIDGE_SCRIPT, - "--port", str(port), - "--session", session_dir, - "--cache-dir", cache_dir, - "--mode", mode, - ] - - _kill_port_process(port) - PrintStyle.info("WhatsApp: starting bridge for pairing") - _bridge_process = _BridgeProcess(subprocess.Popen( - cmd, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - cwd=BRIDGE_DIR, - ), port) - _start_log_reader(_bridge_process) - _bridge_config.clear() - _bridge_config.update({"port": port, "mode": mode}) - - for _ in range(20): - await asyncio.sleep(0.5) - if _bridge_process.poll() is not None: - PrintStyle.error("WhatsApp: bridge process exited unexpectedly") - return False - if await _check_http_up(port): - return True - - return False + async with _get_bridge_lock(): + return await _ensure_bridge_started( + port=port, + session_dir=session_dir, + cache_dir=cache_dir, + mode=mode, + require_connection=False, + start_label="WhatsApp: starting bridge for pairing", + ) def is_process_alive() -> bool: @@ -202,6 +163,141 @@ def get_running_config() -> dict: # Internal # ------------------------------------------------------------------ +def _get_bridge_lock() -> asyncio.Lock: + global _bridge_lock, _bridge_lock_loop + loop = asyncio.get_running_loop() + if _bridge_lock is None or _bridge_lock_loop is not loop: + _bridge_lock = asyncio.Lock() + _bridge_lock_loop = loop + return _bridge_lock + + +async def _ensure_bridge_started( + *, + port: int, + session_dir: str, + cache_dir: str, + mode: str, + require_connection: bool, + start_label: str, +) -> bool: + global _bridge_process + + if _bridge_process and _bridge_process.poll() is None: + if require_connection: + return True + if await _check_http_up(port): + return True + + PrintStyle.warning("WhatsApp: bridge is running but HTTP is not responding, restarting") + _stop_bridge_process() + + await _ensure_bridge_dependencies() + + attempt = 0 + while attempt < 2: + attempt += 1 + success, output = await _start_bridge_once( + port=port, + session_dir=session_dir, + cache_dir=cache_dir, + mode=mode, + require_connection=require_connection, + start_label=start_label, + ) + if success: + return True + + if attempt == 1 and _looks_like_dependency_failure(output): + PrintStyle.warning( + "WhatsApp: bridge startup looks like a dependency issue, " + "reinstalling dependencies and retrying" + ) + await _ensure_bridge_dependencies(force_reinstall=True) + continue + + return False + + return False + + +async def _start_bridge_once( + *, + port: int, + session_dir: str, + cache_dir: str, + mode: str, + require_connection: bool, + start_label: str, +) -> tuple[bool, str]: + global _bridge_process + + cmd = [ + "node", BRIDGE_SCRIPT, + "--port", str(port), + "--session", session_dir, + "--cache-dir", cache_dir, + "--mode", mode, + ] + + _kill_port_process(port) + PrintStyle.info(start_label) + _bridge_process = _BridgeProcess(subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + cwd=BRIDGE_DIR, + ), port) + _start_log_reader(_bridge_process) + _bridge_config.clear() + _bridge_config.update({"port": port, "mode": mode}) + + healthy, output = await _wait_for_bridge_startup( + port=port, + require_connection=require_connection, + ) + if healthy: + return True, output + + if output: + PrintStyle.error(f"WhatsApp: bridge startup failed\n{output}") + return False, output + + +async def _wait_for_bridge_startup(*, port: int, require_connection: bool) -> tuple[bool, str]: + for _ in range(STARTUP_WAIT_ATTEMPTS): + await asyncio.sleep(STARTUP_WAIT_SECONDS) + + process = _bridge_process + if process is None: + return False, "" + + if process.poll() is not None: + output = _summarize_output(process.recent_output()) + PrintStyle.error("WhatsApp: bridge process exited unexpectedly") + _clear_bridge_process() + return False, output + + if require_connection: + if await _check_health(port): + return True, process.recent_output() + else: + if await _check_http_up(port): + return True, process.recent_output() + + if require_connection: + PrintStyle.warning("WhatsApp: bridge started but not yet connected") + process = _bridge_process + return True, process.recent_output() if process else "" + + process = _bridge_process + return False, _summarize_output(process.recent_output()) if process else "" + + +def _looks_like_dependency_failure(output: str) -> bool: + return any(marker in output for marker in DEPENDENCY_FAILURE_MARKERS) + + async def _check_health(port: int) -> bool: try: from plugins._whatsapp_integration.helpers.wa_client import get_health @@ -220,27 +316,171 @@ async def _check_http_up(port: int) -> bool: return False -async def _ensure_npm_install() -> None: - node_modules = os.path.join(BRIDGE_DIR, "node_modules") - key_package = os.path.join(node_modules, "@whiskeysockets", "baileys") - if os.path.isdir(key_package): - return - # Missing or corrupt node_modules — wipe and reinstall - if os.path.isdir(node_modules): - import shutil - PrintStyle.warning("WhatsApp: node_modules corrupt, reinstalling") - shutil.rmtree(node_modules, ignore_errors=True) - PrintStyle.info("WhatsApp: installing bridge dependencies") +async def _ensure_bridge_dependencies(force_reinstall: bool = False) -> None: + expected_state = await _build_dependency_state() + + if not force_reinstall: + install_state = _load_dependency_state() + if os.path.isdir(NODE_MODULES_DIR) and await _validate_bridge_dependencies(): + if install_state is None: + _write_dependency_state(expected_state) + return + if install_state == expected_state: + return + + await _reinstall_bridge_dependencies() + + if not await _validate_bridge_dependencies(): + raise RuntimeError("WhatsApp: bridge dependencies failed validation after reinstall") + + _write_dependency_state(await _build_dependency_state()) + + +async def _build_dependency_state() -> dict[str, Any]: + return { + "package_json_hash": _sha256_file(BRIDGE_PACKAGE_JSON), + "package_lock_hash": _sha256_file(BRIDGE_PACKAGE_LOCK) if os.path.isfile(BRIDGE_PACKAGE_LOCK) else "", + "platform": platform.system(), + "arch": platform.machine(), + "node_version": (await _run_subprocess(["node", "--version"], cwd=BRIDGE_DIR)).strip(), + "npm_version": (await _run_subprocess(["npm", "--version"], cwd=BRIDGE_DIR)).strip(), + } + + +async def _validate_bridge_dependencies() -> bool: + dependency_names = _bridge_dependency_names() + if not dependency_names or not os.path.isdir(NODE_MODULES_DIR): + return False + + imports = ", ".join(json.dumps(name) for name in dependency_names) + script = ( + f"for (const name of [{imports}]) {{ await import(name); }}\n" + "process.stdout.write('ok');\n" + ) + + try: + output = await _run_subprocess( + ["node", "--input-type=module", "--eval", script], + cwd=BRIDGE_DIR, + ) + except RuntimeError as e: + PrintStyle.warning(f"WhatsApp: dependency validation failed: {e}") + return False + return output.strip() == "ok" + + +async def _reinstall_bridge_dependencies() -> None: + _ensure_runtime_dir() + + if os.path.isdir(NODE_MODULES_DIR): + PrintStyle.warning("WhatsApp: bridge dependencies missing, outdated, or corrupt; reinstalling") + shutil.rmtree(NODE_MODULES_DIR, ignore_errors=True) + else: + PrintStyle.info("WhatsApp: installing bridge dependencies") + + if os.path.isfile(BRIDGE_INSTALL_STATE): + try: + os.remove(BRIDGE_INSTALL_STATE) + except OSError: + pass + + commands: list[list[str]] = [] + if os.path.isfile(BRIDGE_PACKAGE_LOCK): + commands.append(["npm", "ci", "--omit=dev", "--no-audit", "--no-fund"]) + commands.append(["npm", "install", "--omit=dev", "--no-audit", "--no-fund"]) + env = {"npm_config_cache": BRIDGE_NPM_CACHE} + + last_error: RuntimeError | None = None + for command in commands: + try: + await _run_subprocess(command, cwd=BRIDGE_DIR, env=env) + return + except RuntimeError as e: + last_error = e + PrintStyle.warning(f"WhatsApp: {' '.join(command)} failed: {e}") + + raise RuntimeError(str(last_error) if last_error else "npm install failed") + + +def _bridge_dependency_names() -> list[str]: + with open(BRIDGE_PACKAGE_JSON, "r", encoding="utf-8") as f: + package_json = json.load(f) + dependencies = package_json.get("dependencies") or {} + return sorted(dependencies.keys()) + + +def _load_dependency_state() -> dict[str, Any] | None: + if not os.path.isfile(BRIDGE_INSTALL_STATE): + return None + try: + with open(BRIDGE_INSTALL_STATE, "r", encoding="utf-8") as f: + return json.load(f) + except (OSError, json.JSONDecodeError): + return None + + +def _write_dependency_state(state: dict[str, Any]) -> None: + _ensure_runtime_dir() + with open(BRIDGE_INSTALL_STATE, "w", encoding="utf-8") as f: + json.dump(state, f, indent=2, sort_keys=True) + + +def _sha256_file(path: str) -> str: + digest = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _ensure_runtime_dir() -> None: + os.makedirs(BRIDGE_RUNTIME_DIR, exist_ok=True) + + +async def _run_subprocess( + command: Sequence[str], + *, + cwd: str, + env: dict[str, str] | None = None, +) -> str: + process_env = os.environ.copy() + if env: + process_env.update(env) + proc = await asyncio.create_subprocess_exec( - "npm", "install", "--production", - cwd=BRIDGE_DIR, + *command, + cwd=cwd, + env=process_env, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) stdout, _ = await proc.communicate() + output = stdout.decode("utf-8", errors="replace").strip() if proc.returncode != 0: - PrintStyle.error(f"WhatsApp: npm install failed: {stdout.decode()}") - raise RuntimeError("npm install failed") + raise RuntimeError(output or f"{' '.join(command)} exited with code {proc.returncode}") + return output + + +def _stop_bridge_process() -> None: + global _bridge_process + if not _bridge_process: + return + try: + _bridge_process.terminate() + try: + _bridge_process.wait(timeout=5) + except subprocess.TimeoutExpired: + _bridge_process.kill() + except Exception: + pass + _clear_bridge_process() + PrintStyle.info("WhatsApp: bridge stopped") + + +def _clear_bridge_process() -> None: + global _bridge_process + _bridge_process = None + _bridge_config.clear() def _kill_port_process(port: int) -> None: @@ -288,13 +528,21 @@ def _kill_port_process(port: int) -> None: def _start_log_reader(process: _BridgeProcess) -> None: - def _reader(): + def _reader() -> None: assert process.stdout for line in iter(process.stdout.readline, b""): text = line.decode("utf-8", errors="replace").rstrip() if text: + process.remember_output(text) PrintStyle.standard(f"WhatsApp bridge: {text}") process.stdout.close() thread = threading.Thread(target=_reader, daemon=True) thread.start() + + +def _summarize_output(output: str, max_lines: int = 12) -> str: + if not output: + return "" + lines = [line for line in output.splitlines() if line.strip()] + return "\n".join(lines[-max_lines:]) diff --git a/plugins/_whatsapp_integration/helpers/handler.py b/plugins/_whatsapp_integration/helpers/handler.py index 666178787..1e8eaa98c 100644 --- a/plugins/_whatsapp_integration/helpers/handler.py +++ b/plugins/_whatsapp_integration/helpers/handler.py @@ -20,22 +20,15 @@ from initialize import initialize_agent from plugins._whatsapp_integration.helpers import wa_client from plugins._whatsapp_integration.helpers import bridge_manager +from plugins._whatsapp_integration.helpers.number_utils import ( + normalize_allowed_numbers, + normalize_number, +) PLUGIN_NAME = "_whatsapp_integration" MEDIA_FOLDER = "usr/whatsapp/media" -# Regex: strip everything from first '@' or ':' onward, then leading '+' and '0's -_NUM_STRIP_RE = re.compile(r"[@:].*") -_NUM_PREFIX_RE = re.compile(r"^[+0]+") - - -def _normalize_number(raw: str) -> str: - """Strip JID suffix, device suffix, leading + and 0s to get a plain number.""" - n = _NUM_STRIP_RE.sub("", raw) - n = _NUM_PREFIX_RE.sub("", n) - return n - # Context data keys (no underscore prefix — must persist across restarts) CTX_WA_CHAT_ID = "wa_chat_id" CTX_WA_SENDER_NAME = "wa_sender_name" @@ -91,15 +84,14 @@ async def poll_messages(config: dict) -> None: if not messages: return - # Allowed-numbers filtering (authoritative check — bridge.js is secondary) - allowed_numbers = config.get("allowed_numbers") or [] - allowed_set = {_normalize_number(n) for n in allowed_numbers} if allowed_numbers else set() + # Allowed-numbers filtering is authoritative in Python. + allowed_set = normalize_allowed_numbers(config.get("allowed_numbers")) for msg in messages: try: # Filter by allowed numbers if configured if allowed_set: - sender_num = _normalize_number(msg.get("senderNumber", "") or msg.get("senderId", "")) + sender_num = normalize_number(msg.get("senderNumber", "") or msg.get("senderId", "")) if sender_num not in allowed_set: PrintStyle.debug( f"WhatsApp: ignored message from {sender_num} " @@ -150,7 +142,7 @@ async def _start_new_chat(config: dict, msg: dict) -> None: from helpers import projects sender_name = msg.get("senderName", "Unknown") - sender_number = msg.get("senderNumber", "") or _normalize_number(msg.get("senderId", "")) + sender_number = msg.get("senderNumber", "") or normalize_number(msg.get("senderId", "")) chat_id = msg.get("chatId", "") is_group = msg.get("isGroup", False) @@ -284,7 +276,7 @@ def _md_to_whatsapp(text: str) -> str: def _build_user_message(agent: Agent, msg: dict) -> str: sender_name = msg.get("senderName", "Unknown") - sender_number = msg.get("senderNumber", "") or _normalize_number(msg.get("senderId", "")) + sender_number = msg.get("senderNumber", "") or normalize_number(msg.get("senderId", "")) is_group = msg.get("isGroup", False) prompt = "fw.wa.user_message_group.md" if is_group else "fw.wa.user_message.md" text = agent.read_prompt( diff --git a/plugins/_whatsapp_integration/helpers/number_utils.py b/plugins/_whatsapp_integration/helpers/number_utils.py new file mode 100644 index 000000000..66e69253c --- /dev/null +++ b/plugins/_whatsapp_integration/helpers/number_utils.py @@ -0,0 +1,32 @@ +"""Phone-number normalization helpers for WhatsApp integration.""" + +from __future__ import annotations + +import re +from collections.abc import Iterable + + +_JID_SUFFIX_RE = re.compile(r"[@:].*") +_NON_DIGIT_RE = re.compile(r"\D+") +_LEADING_ZERO_RE = re.compile(r"^0+") + + +def normalize_number(raw: str) -> str: + """Normalize WhatsApp sender identifiers and phone numbers to comparable digits.""" + text = _JID_SUFFIX_RE.sub("", str(raw or "")) + digits = _NON_DIGIT_RE.sub("", text) + return _LEADING_ZERO_RE.sub("", digits) + + +def normalize_allowed_numbers(value: object) -> set[str]: + """Accept stored config as a list/tuple/set or comma-delimited string.""" + if isinstance(value, str): + candidates = value.split(",") + elif isinstance(value, Iterable): + candidates = value + else: + return set() + + normalized = {normalize_number(item) for item in candidates} + normalized.discard("") + return normalized diff --git a/plugins/_whatsapp_integration/helpers/storage_paths.py b/plugins/_whatsapp_integration/helpers/storage_paths.py new file mode 100644 index 000000000..cb3c8b2a2 --- /dev/null +++ b/plugins/_whatsapp_integration/helpers/storage_paths.py @@ -0,0 +1,11 @@ +"""Shared storage paths for the WhatsApp bridge.""" + +from helpers import files + + +def get_bridge_session_dir() -> str: + return files.get_abs_path(files.USER_DIR, "whatsapp", "session") + + +def get_bridge_media_dir() -> str: + return files.get_abs_path(files.USER_DIR, "whatsapp", "media") diff --git a/plugins/_whatsapp_integration/webui/config.html b/plugins/_whatsapp_integration/webui/config.html index 749ee351e..5a10f42bf 100644 --- a/plugins/_whatsapp_integration/webui/config.html +++ b/plugins/_whatsapp_integration/webui/config.html @@ -26,11 +26,13 @@ } catch (e) { this.projects = []; } }, allowed_text() { - return (config?.allowed_numbers || []).join(', '); + const value = config?.allowed_numbers; + if (Array.isArray(value)) return value.join(', '); + return typeof value === 'string' ? value : ''; }, set_allowed(val) { config.allowed_numbers = val.split(',') - .map(s => s.trim().replace(/^\+/, '').replace(/^0+/, '')) + .map(s => s.trim()) .filter(s => s); }, async test_connection() { @@ -39,7 +41,7 @@ try { const { callJsonApi } = await import('/js/api.js'); const res = await callJsonApi('/plugins/_whatsapp_integration/test_connection', { - config: config + config: { bridge_port: config.bridge_port } }); this.test_results = res; } catch (e) { @@ -240,10 +242,10 @@
Allowed Numbers
-
Use international format without + or leading 0
e.g. +1 (415) 555-1234 → 14155551234. Empty = allow all
+
Comma-separated phone numbers. Matching is normalized by the backend, so punctuation and + prefixes are okay. Empty = allow all
- +
diff --git a/plugins/_whatsapp_integration/whatsapp-bridge/bridge.js b/plugins/_whatsapp_integration/whatsapp-bridge/bridge.js index aa3c16c7e..8249bfcd7 100644 --- a/plugins/_whatsapp_integration/whatsapp-bridge/bridge.js +++ b/plugins/_whatsapp_integration/whatsapp-bridge/bridge.js @@ -41,9 +41,10 @@ const WHATSAPP_DEBUG = typeof process.env.WHATSAPP_DEBUG === 'string' && ['1', 'true', 'yes', 'on'].includes(process.env.WHATSAPP_DEBUG.toLowerCase()); +const DEFAULT_DATA_ROOT = path.resolve(path.dirname(new URL(import.meta.url).pathname), '..', '..', '..', 'tmp', 'whatsapp'); const PORT = parseInt(getArg('port', '3100'), 10); -const SESSION_DIR = getArg('session', path.join(process.env.HOME || '~', '.agent-zero', 'whatsapp', 'session')); -const CACHE_DIR = getArg('cache-dir', path.join(process.env.HOME || '~', '.agent-zero', 'whatsapp', 'media')); +const SESSION_DIR = getArg('session', path.join(DEFAULT_DATA_ROOT, 'session')); +const CACHE_DIR = getArg('cache-dir', path.join(DEFAULT_DATA_ROOT, 'media')); const PAIR_ONLY = args.includes('--pair-only'); const MODE = getArg('mode', 'dedicated'); // "dedicated" or "self-chat" @@ -197,8 +198,6 @@ async function startSocket() { // Skip status broadcasts if (chatId === 'status@broadcast') continue; - // Allowed-numbers filtering is handled in Python (handler.py) - // Unwrap documentWithCaptionMessage (Baileys wraps captioned docs) if (msg.message.documentWithCaptionMessage?.message?.documentMessage) { msg.message.documentMessage = msg.message.documentWithCaptionMessage.message.documentMessage; diff --git a/tests/test_whatsapp_bridge_manager.py b/tests/test_whatsapp_bridge_manager.py new file mode 100644 index 000000000..8364a69d5 --- /dev/null +++ b/tests/test_whatsapp_bridge_manager.py @@ -0,0 +1,181 @@ +import asyncio +import sys +from pathlib import Path + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from plugins._whatsapp_integration.helpers import bridge_manager + + +@pytest.fixture(autouse=True) +def reset_bridge_manager_state(): + bridge_manager._bridge_lock = None + bridge_manager._bridge_lock_loop = None + bridge_manager._bridge_process = None + bridge_manager._bridge_config.clear() + yield + bridge_manager._bridge_lock = None + bridge_manager._bridge_lock_loop = None + bridge_manager._bridge_process = None + bridge_manager._bridge_config.clear() + + +def test_get_bridge_lock_recreates_lock_when_loop_marker_changes(): + async def _run(): + original_lock = asyncio.Lock() + bridge_manager._bridge_lock = original_lock + bridge_manager._bridge_lock_loop = object() + + refreshed_lock = bridge_manager._get_bridge_lock() + + assert refreshed_lock is bridge_manager._bridge_lock + assert refreshed_lock is not original_lock + + asyncio.run(_run()) + + +def test_ensure_bridge_dependencies_reinstalls_invalid_dependency_tree(monkeypatch): + state_writes: list[dict] = [] + validate_results = [False, True] + reinstall_calls: list[bool] = [] + + monkeypatch.setattr(bridge_manager, "NODE_MODULES_DIR", "/tmp/whatsapp-node-modules") + monkeypatch.setattr(bridge_manager.os.path, "isdir", lambda path: path == "/tmp/whatsapp-node-modules") + + async def fake_build_dependency_state(): + return {"package_json_hash": "a", "package_lock_hash": "b"} + + async def fake_validate_bridge_dependencies(): + return validate_results.pop(0) + + async def fake_reinstall_bridge_dependencies(): + reinstall_calls.append(True) + + monkeypatch.setattr(bridge_manager, "_build_dependency_state", fake_build_dependency_state) + monkeypatch.setattr(bridge_manager, "_load_dependency_state", lambda: {"package_json_hash": "a", "package_lock_hash": "b"}) + monkeypatch.setattr(bridge_manager, "_validate_bridge_dependencies", fake_validate_bridge_dependencies) + monkeypatch.setattr(bridge_manager, "_reinstall_bridge_dependencies", fake_reinstall_bridge_dependencies) + monkeypatch.setattr(bridge_manager, "_write_dependency_state", lambda state: state_writes.append(state)) + + asyncio.run(bridge_manager._ensure_bridge_dependencies()) + + assert reinstall_calls == [True] + assert state_writes == [{"package_json_hash": "a", "package_lock_hash": "b"}] + + +def test_ensure_bridge_dependencies_bootstraps_missing_state_without_reinstall(monkeypatch): + state_writes: list[dict] = [] + reinstall_calls: list[bool] = [] + + monkeypatch.setattr(bridge_manager, "NODE_MODULES_DIR", "/tmp/whatsapp-node-modules") + monkeypatch.setattr(bridge_manager.os.path, "isdir", lambda path: path == "/tmp/whatsapp-node-modules") + + async def fake_build_dependency_state(): + return {"package_json_hash": "a", "package_lock_hash": "b"} + + async def fake_validate_bridge_dependencies(): + return True + + async def fake_reinstall_bridge_dependencies(): + reinstall_calls.append(True) + + monkeypatch.setattr(bridge_manager, "_build_dependency_state", fake_build_dependency_state) + monkeypatch.setattr(bridge_manager, "_load_dependency_state", lambda: None) + monkeypatch.setattr(bridge_manager, "_validate_bridge_dependencies", fake_validate_bridge_dependencies) + monkeypatch.setattr(bridge_manager, "_reinstall_bridge_dependencies", fake_reinstall_bridge_dependencies) + monkeypatch.setattr(bridge_manager, "_write_dependency_state", lambda state: state_writes.append(state)) + + asyncio.run(bridge_manager._ensure_bridge_dependencies()) + + assert reinstall_calls == [] + assert state_writes == [{"package_json_hash": "a", "package_lock_hash": "b"}] + + +def test_reinstall_bridge_dependencies_uses_local_npm_cache(monkeypatch): + run_calls: list[tuple[list[str], dict[str, str] | None]] = [] + real_isdir = bridge_manager.os.path.isdir + + monkeypatch.setattr(bridge_manager.PrintStyle, "warning", staticmethod(lambda message: None)) + monkeypatch.setattr(bridge_manager.PrintStyle, "info", staticmethod(lambda message: None)) + monkeypatch.setattr(bridge_manager, "NODE_MODULES_DIR", "/tmp/whatsapp-node-modules") + monkeypatch.setattr(bridge_manager, "BRIDGE_PACKAGE_LOCK", "/tmp/package-lock.json") + monkeypatch.setattr(bridge_manager, "BRIDGE_INSTALL_STATE", "/tmp/deps-state.json") + monkeypatch.setattr(bridge_manager, "BRIDGE_NPM_CACHE", "/tmp/npm-cache") + monkeypatch.setattr( + bridge_manager.os.path, + "isdir", + lambda path: path == "/tmp/whatsapp-node-modules" or real_isdir(path), + ) + monkeypatch.setattr( + bridge_manager.os.path, + "isfile", + lambda path: path == "/tmp/package-lock.json", + ) + monkeypatch.setattr(bridge_manager.shutil, "rmtree", lambda path, ignore_errors=True: None) + + async def fake_run_subprocess(command, *, cwd, env=None): + run_calls.append((list(command), env)) + return "" + + monkeypatch.setattr(bridge_manager, "_run_subprocess", fake_run_subprocess) + + asyncio.run(bridge_manager._reinstall_bridge_dependencies()) + + assert run_calls == [ + ( + ["npm", "ci", "--omit=dev", "--no-audit", "--no-fund"], + {"npm_config_cache": "/tmp/npm-cache"}, + ), + ] + + +def test_start_bridge_reinstalls_and_retries_after_dependency_failure(monkeypatch): + ensure_calls: list[bool] = [] + start_attempts = [ + (False, "Error: Cannot find package '@whiskeysockets/baileys'"), + (True, ""), + ] + + async def fake_ensure_bridge_dependencies(force_reinstall: bool = False): + ensure_calls.append(force_reinstall) + + async def fake_start_bridge_once(**kwargs): + return start_attempts.pop(0) + + monkeypatch.setattr(bridge_manager, "_ensure_bridge_dependencies", fake_ensure_bridge_dependencies) + monkeypatch.setattr(bridge_manager, "_start_bridge_once", fake_start_bridge_once) + + started = asyncio.run(bridge_manager.start_bridge( + port=3100, + session_dir="/tmp/wa-session", + cache_dir="/tmp/wa-cache", + )) + + assert started is True + assert ensure_calls == [False, True] + + +def test_start_bridge_does_not_reinstall_on_non_dependency_failure(monkeypatch): + ensure_calls: list[bool] = [] + + async def fake_ensure_bridge_dependencies(force_reinstall: bool = False): + ensure_calls.append(force_reinstall) + + async def fake_start_bridge_once(**kwargs): + return False, "Bridge closed unexpectedly without module resolution errors" + + monkeypatch.setattr(bridge_manager, "_ensure_bridge_dependencies", fake_ensure_bridge_dependencies) + monkeypatch.setattr(bridge_manager, "_start_bridge_once", fake_start_bridge_once) + + started = asyncio.run(bridge_manager.start_bridge( + port=3100, + session_dir="/tmp/wa-session", + cache_dir="/tmp/wa-cache", + )) + + assert started is False + assert ensure_calls == [False] diff --git a/tests/test_whatsapp_number_utils.py b/tests/test_whatsapp_number_utils.py new file mode 100644 index 000000000..766012981 --- /dev/null +++ b/tests/test_whatsapp_number_utils.py @@ -0,0 +1,36 @@ +import sys +from pathlib import Path + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from plugins._whatsapp_integration.helpers.number_utils import ( + normalize_allowed_numbers, + normalize_number, +) + + +def test_normalize_number_handles_jids_and_phone_formatting(): + assert normalize_number("+1 (415) 555-1234") == "14155551234" + assert normalize_number("00420 777 123 456") == "420777123456" + assert normalize_number("14155551234@s.whatsapp.net") == "14155551234" + assert normalize_number("14155551234:17@s.whatsapp.net") == "14155551234" + + +def test_normalize_allowed_numbers_accepts_lists_and_csv_strings(): + assert normalize_allowed_numbers([ + "+1 (415) 555-1234", + "00420 777 123 456", + "", + ]) == {"14155551234", "420777123456"} + + assert normalize_allowed_numbers( + "+1 (415) 555-1234, 00420 777 123 456" + ) == {"14155551234", "420777123456"} + + +def test_normalize_allowed_numbers_ignores_unsupported_values(): + assert normalize_allowed_numbers(None) == set() + assert normalize_allowed_numbers(12345) == set() diff --git a/tests/test_whatsapp_storage_paths.py b/tests/test_whatsapp_storage_paths.py new file mode 100644 index 000000000..c21bc0e27 --- /dev/null +++ b/tests/test_whatsapp_storage_paths.py @@ -0,0 +1,18 @@ +import sys +from pathlib import Path + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from helpers import files +from plugins._whatsapp_integration.helpers.storage_paths import ( + get_bridge_media_dir, + get_bridge_session_dir, +) + + +def test_whatsapp_bridge_storage_paths_use_agent_zero_tmp(): + assert get_bridge_session_dir() == files.get_abs_path(files.TEMP_DIR, "whatsapp", "session") + assert get_bridge_media_dir() == files.get_abs_path(files.TEMP_DIR, "whatsapp", "media")