fix(dashboard): allow safe WhatsApp re-pairing after logout (#142439)

* fix(dashboard): allow safe WhatsApp re-pairing after logout

* fix(whatsapp): retain pairing reservations through credential flush
This commit is contained in:
Gille authored and GitHub committed 2026-10-21 10:08:33 -04:00
1 parent 7dcb1b4cfd
commit 13282fdac4
13 files changed
+892 -97

No files matched your search

+4 -21
View File
@@ -1513,27 +1513,6 @@ def remove_pid_file() -> None:
_clear_running_pid_cache()
def _scoped_lock_record_is_stale(existing: dict[str, Any], existing_pid: Optional[int]) -> bool:
"""True when a foreign scoped-lock record no longer names a live gateway: PID missing/dead,
start time changed (PID reuse), or the live process is not a gateway -- a readable cmdline says
so (also catches boot-time PID+start_time collisions; systemd spawns deterministically);
cmdline unreadable AND start_time unknown on either side => the lock record's own argv is the
only signal left. Stopped (SIGTSTP) processes look alive to _pid_exists; stale so --replace
works."""
if existing_pid is None or not _pid_exists(existing_pid):
return True
recorded_start = existing.get("start_time")
current_start = _get_process_start_time(existing_pid)
if _start_times_conflict(recorded_start, current_start):
return True
if not _looks_like_gateway_process(existing_pid):
if _read_process_cmdline(existing_pid) is not None:
return True
if None in (recorded_start, current_start) and not _record_looks_like_gateway(existing):
return True
return _process_is_stopped(existing_pid)
def _process_is_stopped(pid: int) -> bool:
"""True for a stopped / tracing-stop state (T/t) in ``/proc/<pid>/status``."""
with contextlib.suppress(OSError):
@@ -1548,6 +1527,8 @@ def acquire_scoped_lock(
) -> tuple[bool, Optional[dict[str, Any]]]:
"""Acquire a machine-local lock keyed by scope + identity (one Telegram token across homes).
``metadata={"owner_kind": "process"}`` allows a non-gateway owner such as
a setup process. Its PID/start-time identity still gates stale recovery.
Raises :class:`gateway.status_lock_dir.ScopedLockDirUnusable` (logged once per dir with the repair) when this token may
not use the lock dir at all."""
lock_path = _get_scope_lock_path(scope, identity)
@@ -1562,6 +1543,8 @@ def acquire_scoped_lock(
def _acquire_scoped_lock_at(
lock_path: Path, scope: str, identity: str, metadata: Optional[dict[str, Any]]
) -> tuple[bool, Optional[dict[str, Any]]]:
from gateway.status_scoped_locks import _scoped_lock_record_is_stale
make_user_dir(lock_path.parent, parents=True, exist_ok=True)
# An existing lock this token may not read (left by an elevated gateway) must raise here:
# ``_read_json_file`` would call it empty and O_EXCL "in use", a conflict with no holder.
+28 -4
View File
@@ -1,15 +1,39 @@
"""Scoped-lock release, split from :mod:`gateway.status` (acquire stays there). One-way: this
module reads the status helpers through the module at call time (so tests that patch
``gateway.status`` keep steering them); :mod:`gateway.status` never imports it."""
"""Scoped-lock ownership and release, split from :mod:`gateway.status` (acquire stays there).
Helpers read status through the module at call time, preserving its patch seams.
The acquire path imports the ownership check lazily to avoid an import cycle.
"""
from __future__ import annotations
import contextlib
import os
from typing import Optional
from typing import Any, Optional
from gateway import status as _status
def _scoped_lock_record_is_stale(existing: dict[str, Any], existing_pid: Optional[int]) -> bool:
"""Recover dead/reused/stopped owners. Gateway locks also validate gateway argv.
An explicitly process-owned reservation can belong to a dashboard or setup
worker. Unknown identity stays reserved instead of guessing that it is dead.
"""
if existing_pid is None or not _status._pid_exists(existing_pid):
return True
recorded_start = existing.get("start_time")
current_start = _status._get_process_start_time(existing_pid)
if _status._start_times_conflict(recorded_start, current_start):
return True
metadata = existing.get("metadata")
process_owner = isinstance(metadata, dict) and metadata.get("owner_kind") == "process"
if not process_owner and not _status._looks_like_gateway_process(existing_pid):
if _status._read_process_cmdline(existing_pid) is not None:
return True
if None in (recorded_start, current_start) and not _status._record_looks_like_gateway(existing):
return True
return _status._process_is_stopped(existing_pid)
def release_scoped_lock(scope: str, identity: str) -> None:
"""Release a scope lock owned by this PID. No start_time equality check: on-disk null vs a live
fingerprint would wedge reconnects."""
+1
View File
@@ -76,6 +76,7 @@ class WhatsAppOnboardingStart(BaseModel):
mode: Optional[str] = "bot"
allowed_users: Optional[str] = ""
profile: Optional[str] = None
reset_session: bool = False
class WhatsAppOnboardingApply(BaseModel):
mode: Optional[str] = None
+145 -33
View File
@@ -27,6 +27,7 @@ from hermes_cli.config import OPTIONAL_ENV_VARS, get_env_path
from hermes_constants import get_process_hermes_home
from hermes_cli.web_deps import LateState, late
from hermes_cli.web_server_gateway import _restart_gateway_after
from hermes_cli.web_whatsapp_session import _backup_whatsapp_session
from hermes_cli.web_server_messaging import (
_TelegramOnboardingPairing, _WhatsAppOnboardingSession, _messaging_platform_catalog, _telegram_onboarding_error_message, _telegram_onboarding_lock, _telegram_onboarding_pairings, _whatsapp_onboarding_payload, _whatsapp_onboarding_sessions,
)
@@ -435,6 +436,7 @@ def _terminate_whatsapp_pairing(proc: subprocess.Popen | None) -> None:
except Exception:
with contextlib.suppress(Exception):
proc.kill()
proc.wait(timeout=3)
def _fail_whatsapp_pairing(pairing_id: str, error: str, *, proc=None, unless=_WHATSAPP_ONBOARDING_TERMINAL_STATUSES) -> None:
@@ -485,7 +487,8 @@ def _watch_whatsapp_pairing(pairing_id: str, proc: subprocess.Popen) -> None:
record = _whatsapp_onboarding_sessions.get(pairing_id)
if not record or record.proc is not proc:
return
_apply_pairing_event(record, payload)
if record.status not in {"cancelled", "expired"}:
_apply_pairing_event(record, payload)
returncode = proc.wait()
except Exception as exc:
_fail_whatsapp_pairing(pairing_id, str(exc), proc=proc)
@@ -501,40 +504,71 @@ def _run_whatsapp_pairing(pairing_id: str, session_path: Path, mode: str) -> Non
with _whatsapp_onboarding_lock:
record = _whatsapp_onboarding_sessions.get(pairing_id)
if not record or record.status in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES:
if record:
_release_whatsapp_reservation(record)
return
record.status = "installing"
record.starting_process = True
try:
proc = _spawn_whatsapp_pairing_process(session_path, mode)
except Exception as exc:
_fail_whatsapp_pairing(pairing_id, str(exc))
return
with _whatsapp_onboarding_lock:
record = _whatsapp_onboarding_sessions.get(pairing_id)
if not record or record.status in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES:
_terminate_whatsapp_pairing(proc)
try:
proc = _spawn_whatsapp_pairing_process(session_path, mode)
except Exception as exc:
_fail_whatsapp_pairing(pairing_id, str(exc))
return
record.proc = proc
record.status = "starting"
_watch_whatsapp_pairing(pairing_id, proc)
with _whatsapp_onboarding_lock:
record.starting_process = False
record.proc = proc
if record.status in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES:
_terminate_whatsapp_pairing(proc)
return
record.status = "starting"
_watch_whatsapp_pairing(pairing_id, proc)
finally:
with _whatsapp_onboarding_lock:
record.starting_process = False
_release_whatsapp_reservation(record)
def _release_whatsapp_reservation(record: _WhatsAppOnboardingSession) -> None:
"""Call under the onboarding lock. A live writer keeps its reservation."""
if not record.session_reserved or record.starting_process:
return
if record.proc is None and record.status not in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES:
return # A queued worker has not entered its startup section yet.
if record.proc is not None and record.proc.poll() is None:
return
from gateway.status_scoped_locks import release_scoped_lock
release_scoped_lock("whatsapp-session", record.session_path)
record.session_reserved = False
if record.expiry_timer is not None:
record.expiry_timer.cancel()
def _expire_whatsapp_pairing(record: _WhatsAppOnboardingSession) -> None:
with _whatsapp_onboarding_lock:
if record.status not in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES or record.session_reserved:
record.status = "expired"
record.error = "WhatsApp QR setup expired. Start a new setup."
_terminate_whatsapp_pairing(record.proc)
_release_whatsapp_reservation(record)
def _prune_whatsapp_onboarding_sessions() -> None:
now = time.time()
remove_ids: list[str] = []
for pairing_id, record in _whatsapp_onboarding_sessions.items():
_release_whatsapp_reservation(record)
live = record.status not in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES
if live and record.proc is not None and record.proc.poll() is not None:
record.status = "error"
record.error = "WhatsApp pairing process exited before pairing completed."
live = False
if live and record.expires_at_ts <= now:
_terminate_whatsapp_pairing(record.proc)
record.status = "expired"
record.error = "WhatsApp QR setup expired. Start a new setup."
if record.status in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES and record.expires_at_ts + 300 <= now:
if record.expires_at_ts <= now:
_expire_whatsapp_pairing(record)
if record.status in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES and not record.session_reserved and record.expires_at_ts + 300 <= now:
remove_ids.append(pairing_id)
for pairing_id in remove_ids:
_whatsapp_onboarding_sessions.pop(pairing_id, None)
@@ -555,28 +589,81 @@ def _register_whatsapp_session(session_path: Path, record) -> str:
return pairing_id
def _reserve_whatsapp_session(record: _WhatsAppOnboardingSession) -> None:
"""Reserve before moving keys or launching a writer, including its queued start."""
from gateway.status import acquire_scoped_lock
busy = "WhatsApp is in use. Disable WhatsApp and restart the gateway, then try re-pairing again."
for existing in _whatsapp_onboarding_sessions.values():
if existing.session_path == record.session_path and (
existing.status not in _WHATSAPP_ONBOARDING_TERMINAL_STATUSES
or existing.session_reserved or existing.starting_process
or (existing.proc is not None and existing.proc.poll() is None)
):
raise HTTPException(status_code=409, detail="Another WhatsApp QR setup is still active. Cancel it or wait for it to stop before re-pairing.")
acquired, previous = acquire_scoped_lock("whatsapp-session", record.session_path, metadata={"owner_kind": "process"})
if not acquired or previous and previous.get("pid") == os.getpid():
# Self-reacquisition is allowed for gateway reconnects. Reset must not
# borrow or release a lock already held by this process.
raise HTTPException(status_code=409, detail=busy)
record.session_reserved = True
def _launch_whatsapp_pairing(pairing_id: str, record: _WhatsAppOnboardingSession, session_path: Path) -> None:
from agent.memory_provider import spawn_context_thread
record.expiry_timer = threading.Timer(_WHATSAPP_ONBOARDING_TTL_SECONDS, _expire_whatsapp_pairing, args=(record,))
record.expiry_timer.daemon = True
launched = False
try:
record.expiry_timer.start()
spawn_context_thread(_run_whatsapp_pairing, name="whatsapp-pairing",
args=(pairing_id, session_path, record.mode)).start()
launched = True
finally:
if not launched:
record.status = "error"
_release_whatsapp_reservation(record)
@router.post("/api/messaging/whatsapp/onboarding/start")
async def start_whatsapp_onboarding(body: WhatsAppOnboardingStart):
async def start_whatsapp_onboarding(body: WhatsAppOnboardingStart, profile: Optional[str] = None):
mode = _normalize_whatsapp_onboarding_mode(body.mode)
allowed_users = _normalize_whatsapp_allowed_users(body.allowed_users)
with _config_profile_scope(body.profile):
target_profile = body.profile or profile
with _config_profile_scope(target_profile), _whatsapp_onboarding_lock:
_prune_whatsapp_onboarding_sessions()
session_path = _whatsapp_session_path()
if any(r.session_path == str(session_path) and r.session_reserved for r in _whatsapp_onboarding_sessions.values()):
raise HTTPException(status_code=409, detail="Another WhatsApp QR setup is still active. Cancel it or wait for it to stop before re-pairing.")
expires_at_ts = time.time() + _WHATSAPP_ONBOARDING_TTL_SECONDS
fields = dict(
fields: dict[str, Any] = dict(
proc=None, mode=mode, allowed_users=allowed_users, session_path=str(session_path),
expires_at=datetime.fromtimestamp(expires_at_ts, UTC).isoformat().replace("+00:00", "Z"),
expires_at_ts=expires_at_ts, profile=body.profile,
expires_at_ts=expires_at_ts, profile=target_profile,
)
already_linked = (session_path / "creds.json").exists()
already_linked = not body.reset_session and (session_path / "creds.json").exists()
if already_linked: # creds on disk: report connected without pairing
account_id, account_name, account_phone = _whatsapp_linked_account_from_session(session_path)
fields.update(status="connected", account_id=account_id, account_name=account_name, account_phone=account_phone)
record = _WhatsAppOnboardingSession(**fields)
pairing_id = _register_whatsapp_session(session_path, record)
if not already_linked:
threading.Thread(target=_run_whatsapp_pairing, args=(pairing_id, session_path, mode), daemon=True).start()
record = _WhatsAppOnboardingSession(**fields)
if already_linked:
pairing_id = _register_whatsapp_session(session_path, record)
else:
_reserve_whatsapp_session(record)
launched = False
try:
if body.reset_session:
_backup_whatsapp_session(session_path)
pairing_id = _register_whatsapp_session(session_path, record)
_launch_whatsapp_pairing(pairing_id, record, session_path)
launched = True
finally:
if not launched:
record.status = "error"
_release_whatsapp_reservation(record)
return _whatsapp_onboarding_payload(pairing_id, record)
@@ -600,6 +687,7 @@ async def get_whatsapp_onboarding_status(pairing_id: str):
@router.post("/api/messaging/whatsapp/onboarding/{pairing_id}/apply")
async def apply_whatsapp_onboarding(pairing_id: str, body: WhatsAppOnboardingApply, profile: Optional[str] = None):
await _wait_whatsapp_pairing_flush(pairing_id)
with _whatsapp_onboarding_lock:
record = _whatsapp_record_or_404(pairing_id)
if record.status != "connected":
@@ -623,19 +711,43 @@ async def apply_whatsapp_onboarding(pairing_id: str, body: WhatsAppOnboardingApp
_write_platform_enabled("whatsapp", True)
with _whatsapp_onboarding_lock:
_whatsapp_onboarding_sessions.pop(pairing_id, None)
# A connected pair-only bridge may still be flushing credentials.
# Retain it for reset's writer guard until that process has exited.
if record.proc is None or record.proc.poll() is not None:
_whatsapp_onboarding_sessions.pop(pairing_id, None)
restart_result = _restart_gateway_after_whatsapp_onboarding(effective_profile)
return {"ok": True, "platform": "whatsapp", "needs_restart": not restart_result["restart_started"], **restart_result}
async def _wait_whatsapp_pairing_flush(pairing_id: str) -> None:
with _whatsapp_onboarding_lock:
record = _whatsapp_record_or_404(pairing_id)
if record.status != "connected":
raise HTTPException(status_code=409, detail="WhatsApp setup is not connected yet.")
if record.session_reserved and record.proc is not None:
try:
await asyncio.to_thread(record.proc.wait, timeout=3)
except subprocess.TimeoutExpired as exc:
raise HTTPException(status_code=409, detail="WhatsApp is still saving its linked device. Wait a moment and try Save and restart again.") from exc
with _whatsapp_onboarding_lock:
_release_whatsapp_reservation(record)
if record.session_reserved:
raise HTTPException(status_code=409, detail="WhatsApp is still finishing pairing. Wait a moment and try Save and restart again.")
@router.delete("/api/messaging/whatsapp/onboarding/{pairing_id}")
async def cancel_whatsapp_onboarding(pairing_id: str):
with _whatsapp_onboarding_lock:
record = _whatsapp_onboarding_sessions.pop(pairing_id, None)
if record:
record.status = "cancelled"
_terminate_whatsapp_pairing(record.proc)
record = _whatsapp_onboarding_sessions.get(pairing_id)
if record:
record.status = "cancelled"
_terminate_whatsapp_pairing(record.proc)
_release_whatsapp_reservation(record)
# Keep a cancelled worker visible until its process launch has
# finished. It may still be opening the session directory.
if not record.starting_process and (record.proc is None or record.proc.poll() is not None):
_whatsapp_onboarding_sessions.pop(pairing_id, None)
return {"ok": True}
+3
View File
@@ -199,6 +199,9 @@ class _WhatsAppOnboardingSession:
account_name: str | None = None
account_phone: str | None = None
error: str | None = None
starting_process: bool = False
session_reserved: bool = False
expiry_timer: threading.Timer | None = None
_whatsapp_onboarding_sessions: dict[str, _WhatsAppOnboardingSession] = {}
+57
View File
@@ -0,0 +1,57 @@
"""Preserve WhatsApp authentication state before dashboard re-pairing."""
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from fastapi import HTTPException
def _assert_whatsapp_bridge_stopped(session_path: Path) -> None:
from gateway.status import _pid_exists, get_process_start_time
pid_file = session_path / "bridge.pid"
if not pid_file.exists():
return
unknown = "Cannot verify the WhatsApp bridge is stopped. Check its bridge.pid before re-pairing."
try:
lines = pid_file.read_text(encoding="utf-8-sig").splitlines()
pid = int(lines[0])
if pid <= 0:
raise ValueError("invalid bridge PID")
except (OSError, ValueError, IndexError) as exc:
raise HTTPException(status_code=409, detail=unknown) from exc
if not _pid_exists(pid):
return
try:
recorded_start = int(lines[1])
if recorded_start <= 0:
raise ValueError("invalid bridge start time")
except (ValueError, IndexError) as exc:
raise HTTPException(status_code=409, detail=unknown) from exc
current_start = get_process_start_time(pid)
if current_start is None:
raise HTTPException(status_code=409, detail=unknown)
if current_start == recorded_start:
raise HTTPException(status_code=409, detail="The WhatsApp bridge is still running. Stop it before re-pairing.")
# Both fingerprints are available and differ. This is a recycled PID,
# not evidence that the old bridge still owns the session.
def _backup_whatsapp_session(session_path: Path) -> None:
"""Called while onboarding owns the session reservation. Never delete keys."""
_assert_whatsapp_bridge_stopped(session_path)
if session_path.is_symlink():
raise HTTPException(status_code=409, detail="The WhatsApp session is a symbolic link. Back it up manually before re-pairing.")
backup = None
try:
if session_path.exists():
stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
backup = Path(tempfile.mkdtemp(prefix=f"session-backup-{stamp}-", dir=session_path.parent)) / "session"
session_path.rename(backup)
session_path.mkdir(mode=0o700)
except OSError as exc:
detail = f"Could not back up the WhatsApp session: {exc}. Pairing was not started."
if backup is not None and backup.exists():
detail += f" The previous session is preserved at {backup}."
raise HTTPException(status_code=500, detail=detail) from exc
@@ -0,0 +1,39 @@
"""Non-gateway setup processes can reserve an existing adapter lock."""
import os
from gateway import status
from gateway.status_scoped_locks import _scoped_lock_record_is_stale
def test_live_process_owner_does_not_need_gateway_argv():
record = status._build_pid_record()
record["metadata"] = {"owner_kind": "process"}
assert not _scoped_lock_record_is_stale(record, os.getpid())
def test_default_gateway_owner_still_requires_gateway_argv():
record = status._build_pid_record()
assert _scoped_lock_record_is_stale(record, os.getpid())
def test_recycled_process_owner_is_stale():
record = status._build_pid_record()
assert record["start_time"] is not None
record["start_time"] -= 10
record["metadata"] = {"owner_kind": "process"}
assert _scoped_lock_record_is_stale(record, os.getpid())
def test_unknown_live_process_identity_remains_reserved(monkeypatch):
record = status._build_pid_record()
record["start_time"] = None
record["metadata"] = {"owner_kind": "process"}
monkeypatch.setattr(status, "_get_process_start_time", lambda pid: None)
assert not _scoped_lock_record_is_stale(record, os.getpid())
def test_invalid_metadata_does_not_bypass_gateway_validation():
record = status._build_pid_record()
record["metadata"] = ["process"]
assert _scoped_lock_record_is_stale(record, os.getpid())
@@ -0,0 +1,443 @@
"""Dashboard recovery must bypass revoked credentials without losing them."""
import asyncio
import json
import os
import re
import subprocess
import sys
import time
from pathlib import Path
from subprocess import Popen
from types import SimpleNamespace
from typing import cast
import pytest
from fastapi import HTTPException
from hermes_cli import web_models, web_server_messaging
from hermes_cli.web_routers import messaging
@pytest.fixture
def pairing(tmp_path, monkeypatch):
from gateway import status
home = tmp_path / "home"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "locks"))
messaging._whatsapp_onboarding_sessions.clear()
session = home / "platforms" / "whatsapp" / "session"
session.mkdir(parents=True)
creds = '{"me":{"id":"15550001111:1@s.whatsapp.net"},"registered":true}'
(session / "creds.json").write_text(creds, encoding="utf-8")
(session / "app-state.json").write_text("preserve", encoding="utf-8")
starts = []
workers = []
timers = []
real_thread = messaging.threading.Thread
class Timer:
def __init__(self, interval, function, args):
self.function, self.args = function, args
self.cancelled = False
def start(self):
timers.append(self)
def cancel(self):
self.cancelled = True
def fire(self):
if not self.cancelled:
self.function(*self.args)
class Thread:
def __init__(self, *, target, args, daemon, kwargs=None, name=None):
self.args = args
self.target = target
def start(self):
starts.append(self.args)
workers.append(self)
monkeypatch.setattr(messaging.threading, "Thread", Thread)
monkeypatch.setattr(messaging.threading, "Timer", Timer)
def start(**kwargs):
return asyncio.run(messaging.start_whatsapp_onboarding(web_models.WhatsAppOnboardingStart(**kwargs)))
def gateway_can_acquire():
from hermes_cli._subprocess_compat import windows_hide_flags
output = tmp_path / "lock-probe.txt"
code = "from gateway.status import acquire_scoped_lock; import sys; print(acquire_scoped_lock('whatsapp-session', sys.argv[1])[0])"
with output.open("w", encoding="utf-8") as stream:
subprocess.run([sys.executable, "-c", code, str(session)], stdout=stream, stderr=subprocess.STDOUT,
timeout=20, check=True, creationflags=windows_hide_flags())
return output.read_text(encoding="utf-8").strip() == "True"
yield SimpleNamespace(home=home, session=session, creds=creds, starts=starts, workers=workers, timers=timers,
start=start, gateway_can_acquire=gateway_can_acquire, real_thread=real_thread)
from gateway.status_scoped_locks import release_scoped_lock
for record in messaging._whatsapp_onboarding_sessions.values():
timer = record.expiry_timer
if timer is not None:
timer.cancel()
release_scoped_lock("whatsapp-session", record.session_path)
messaging._whatsapp_onboarding_sessions.clear()
def test_repair_preserves_credentials_and_starts_fresh_pairing(pairing):
result = pairing.start(reset_session=True)
assert result["status"] == "starting"
assert result["qr_payload"] is None
assert result["account_id"] is None
assert len(pairing.starts) == 1
assert pairing.starts[0][1] == pairing.session
assert not (pairing.session / "creds.json").exists()
backups = list(pairing.session.parent.glob("session-backup-*/session"))
assert len(backups) == 1
assert (backups[0] / "creds.json").read_text(encoding="utf-8") == pairing.creds
assert (backups[0] / "app-state.json").read_text(encoding="utf-8") == "preserve"
assert re.fullmatch(r"session-backup-\d{8}T\d{6}Z-.+", backups[0].parent.name)
def test_reservation_blocks_gateway_until_cancel_before_worker_start(pairing):
result = pairing.start(reset_session=True)
assert pairing.gateway_can_acquire() is False
asyncio.run(messaging.cancel_whatsapp_onboarding(result["pairing_id"]))
assert pairing.gateway_can_acquire() is True
assert pairing.timers[0].cancelled is True
def test_unattended_setup_expires_and_releases_reservation(pairing):
result = pairing.start(reset_session=True)
assert pairing.gateway_can_acquire() is False
pairing.timers[0].fire()
assert messaging._whatsapp_onboarding_sessions[result["pairing_id"]].status == "expired"
assert pairing.gateway_can_acquire() is True
def test_failed_process_start_releases_reservation(pairing, monkeypatch):
result = pairing.start(reset_session=True)
def fail_spawn(*args):
assert pairing.gateway_can_acquire() is False
raise OSError("cannot start bridge")
monkeypatch.setattr(messaging, "_spawn_whatsapp_pairing_process", fail_spawn)
messaging._run_whatsapp_pairing(*pairing.starts[0])
assert messaging._whatsapp_onboarding_sessions[result["pairing_id"]].status == "error"
assert pairing.gateway_can_acquire() is True
def test_recycled_pid_does_not_block_repair(pairing):
from gateway.status import get_process_start_time
started = get_process_start_time(os.getpid())
assert started is not None
(pairing.session / "bridge.pid").write_text(f"{os.getpid()}\n{started - 1}\n19877", encoding="utf-8")
assert pairing.start(reset_session=True)["status"] == "starting"
def test_matching_pid_fingerprint_refuses_repair(pairing):
from gateway.status import get_process_start_time
(pairing.session / "bridge.pid").write_text(f"{os.getpid()}\n{get_process_start_time(os.getpid())}", encoding="utf-8")
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 409
assert (pairing.session / "creds.json").exists()
class _Bridge:
def __init__(self, lines=()):
self.stdout = lines
self.returncode: int | None = None
def poll(self):
return self.returncode
def wait(self, timeout=None):
self.returncode = 0 if self.returncode is None else self.returncode
return self.returncode
def terminate(self):
self.returncode = -15
def kill(self):
self.returncode = -9
def test_reservation_covers_connected_event_and_credential_flush(pairing, monkeypatch):
result = pairing.start(reset_session=True)
record = messaging._whatsapp_onboarding_sessions[result["pairing_id"]]
def events():
yield '{"event":"connected","user":{"id":"15550001111"}}'
assert record.status == "connected"
assert pairing.gateway_can_acquire() is False
# A status poll must not release the lock before the process exits.
asyncio.run(messaging.get_whatsapp_onboarding_status(result["pairing_id"]))
assert pairing.gateway_can_acquire() is False
proc = _Bridge(events())
monkeypatch.setattr(messaging, "_spawn_whatsapp_pairing_process", lambda *args: proc)
messaging._run_whatsapp_pairing(*pairing.starts[0])
assert record.status == "connected"
assert record.session_reserved is False
assert pairing.gateway_can_acquire() is True
assert pairing.timers[0].cancelled
def test_expiry_stops_active_writer_and_releases_reservation(pairing, monkeypatch):
result = pairing.start(reset_session=True)
proc = _Bridge()
def events():
yield '{"event":"qr","qr":"test"}'
assert pairing.gateway_can_acquire() is False
pairing.timers[0].fire()
assert proc.poll() is not None
yield '{"event":"connected"}' # Late events cannot undo expiry.
proc.stdout = events()
monkeypatch.setattr(messaging, "_spawn_whatsapp_pairing_process", lambda *args: proc)
messaging._run_whatsapp_pairing(*pairing.starts[0])
assert messaging._whatsapp_onboarding_sessions[result["pairing_id"]].status == "expired"
assert pairing.gateway_can_acquire() is True
def test_expiry_during_spawn_keeps_reservation_until_late_process_stops(pairing, monkeypatch):
result = pairing.start(reset_session=True)
proc = _Bridge()
def spawn(*args):
pairing.timers[0].fire()
assert pairing.gateway_can_acquire() is False
return proc
monkeypatch.setattr(messaging, "_spawn_whatsapp_pairing_process", spawn)
messaging._run_whatsapp_pairing(*pairing.starts[0])
assert proc.poll() is not None
assert messaging._whatsapp_onboarding_sessions[result["pairing_id"]].status == "expired"
assert pairing.gateway_can_acquire() is True
def test_thread_start_failure_releases_reservation(pairing, monkeypatch):
def fail_thread(*args, **kwargs):
raise RuntimeError("cannot start thread")
monkeypatch.setattr(messaging.threading, "Thread", fail_thread)
with pytest.raises(RuntimeError, match="cannot start thread"):
pairing.start(reset_session=True)
assert pairing.gateway_can_acquire() is True
assert pairing.timers[0].cancelled
def test_apply_waits_for_flush_before_restarting_gateway(pairing, monkeypatch):
from hermes_cli import config
result = pairing.start(reset_session=True)
record = messaging._whatsapp_onboarding_sessions[result["pairing_id"]]
record.status = "connected"
proc = _Bridge()
record.proc = cast(Popen, proc)
monkeypatch.setattr(messaging.threading, "Thread", pairing.real_thread)
monkeypatch.setattr(config, "save_env_value", lambda *args: None)
monkeypatch.setattr(web_server_messaging, "_write_platform_enabled", lambda *args: None)
def wait(timeout=None):
assert pairing.gateway_can_acquire() is False
proc.returncode = 0
return 0
def restart(profile=None):
assert proc.poll() == 0
assert pairing.gateway_can_acquire() is True
return {"restart_started": True}
monkeypatch.setattr(proc, "wait", wait)
monkeypatch.setattr(web_server_messaging, "_restart_gateway_after_whatsapp_onboarding", restart)
assert asyncio.run(messaging.apply_whatsapp_onboarding(result["pairing_id"], web_models.WhatsAppOnboardingApply()))["ok"]
def test_unverifiable_live_pid_stays_blocked(pairing, monkeypatch):
from gateway import status
(pairing.session / "bridge.pid").write_text(f"{os.getpid()}\n123", encoding="utf-8")
monkeypatch.setattr(status, "get_process_start_time", lambda pid: None)
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 409
assert pairing.gateway_can_acquire() is True
def test_regular_start_keeps_existing_link(pairing):
assert pairing.start()["status"] == "connected"
assert pairing.starts == []
assert (pairing.session / "creds.json").read_text(encoding="utf-8") == pairing.creds
assert list(pairing.session.parent.glob("session-backup-*")) == []
def test_repair_refuses_live_bridge_without_moving_credentials(pairing):
(pairing.session / "bridge.pid").write_text(str(os.getpid()), encoding="utf-8")
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 409
assert "bridge" in refused.value.detail.lower()
assert (pairing.session / "creds.json").read_text(encoding="utf-8") == pairing.creds
assert pairing.starts == []
def test_repair_refuses_pending_pairing(pairing):
record = web_server_messaging._WhatsAppOnboardingSession(
proc=None, mode="bot", allowed_users="", session_path=str(pairing.session),
expires_at="2099-01-01T00:00:00Z", expires_at_ts=time.time() + 600,
status="installing",
)
messaging._whatsapp_onboarding_sessions["pending"] = record
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 409
assert record.status == "installing"
assert (pairing.session / "creds.json").exists()
def test_repair_refuses_session_lock_owned_by_live_gateway(pairing):
from gateway import status
lock = status._get_scope_lock_path("whatsapp-session", str(pairing.session))
lock.parent.mkdir(parents=True)
# A second process is not needed to establish a refusal. Self-owned locks
# must also remain intact instead of being borrowed and released by reset.
record = status._build_pid_record()
record["identity_hash"] = status._scope_hash(str(pairing.session))
lock.write_text(json.dumps(record), encoding="utf-8")
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 409
assert lock.exists()
assert (pairing.session / "creds.json").exists()
def test_failed_backup_aborts_before_pairing(pairing, monkeypatch):
rename = Path.rename
def refuse(path, target):
if path == pairing.session:
raise PermissionError("session is locked")
return rename(path, target)
monkeypatch.setattr(Path, "rename", refuse)
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 500
assert (pairing.session / "creds.json").read_text(encoding="utf-8") == pairing.creds
assert pairing.starts == []
@pytest.mark.platforms("windows")
def test_open_credentials_refuse_backup_on_windows(pairing):
with (pairing.session / "creds.json").open("rb"):
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 500
assert (pairing.session / "creds.json").read_text(encoding="utf-8") == pairing.creds
assert pairing.starts == []
def test_apply_retains_flushing_bridge_for_reset_guard(pairing, monkeypatch):
from hermes_cli import config
result = pairing.start()
record = messaging._whatsapp_onboarding_sessions[result["pairing_id"]]
record.proc = cast(Popen, SimpleNamespace(poll=lambda: None))
monkeypatch.setattr(config, "save_env_value", lambda *args: None)
monkeypatch.setattr(web_server_messaging, "_write_platform_enabled", lambda *args: None)
monkeypatch.setattr(web_server_messaging, "_restart_gateway_after_whatsapp_onboarding", lambda profile=None: {"restart_started": True})
asyncio.run(messaging.apply_whatsapp_onboarding(result["pairing_id"], web_models.WhatsAppOnboardingApply()))
assert messaging._whatsapp_onboarding_sessions[result["pairing_id"]] is record
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 409
assert (pairing.session / "creds.json").read_text(encoding="utf-8") == pairing.creds
def test_cancel_during_process_start_cannot_race_session_backup(pairing, monkeypatch):
result = pairing.start(reset_session=True)
record = messaging._whatsapp_onboarding_sessions[result["pairing_id"]]
def spawn(session_path, mode):
asyncio.run(messaging.cancel_whatsapp_onboarding(result["pairing_id"]))
with pytest.raises(HTTPException) as refused:
pairing.start(reset_session=True)
assert refused.value.status_code == 409
raise OSError("pairing cancelled during process startup")
monkeypatch.setattr(messaging, "_spawn_whatsapp_pairing_process", spawn)
messaging._run_whatsapp_pairing(*pairing.starts[0])
assert record.status == "cancelled"
assert record.starting_process is False
assert pairing.start(reset_session=True)["status"] == "starting"
def test_repair_selected_profile_leaves_default_session_untouched(pairing, monkeypatch):
from hermes_cli import web_server_profiles
other = pairing.home / "profiles" / "work"
other.mkdir(parents=True)
(other / "config.yaml").write_text("model: test\n", encoding="utf-8")
session = other / "platforms" / "whatsapp" / "session"
session.mkdir(parents=True)
(session / "creds.json").write_text('{"registered":true}', encoding="utf-8")
monkeypatch.setattr(web_server_profiles, "_resolve_profile_dir", lambda profile: other)
result = pairing.start(profile="work", reset_session=True)
assert result["status"] == "starting"
assert pairing.starts[0][1] == session
assert not (session / "creds.json").exists()
assert (pairing.session / "creds.json").read_text(encoding="utf-8") == pairing.creds
observed = []
def spawn(*args):
from hermes_constants import get_hermes_home
observed.append(get_hermes_home())
raise OSError("stop after checking the worker profile")
monkeypatch.setattr(messaging, "_spawn_whatsapp_pairing_process", spawn)
pairing.workers[0].target(*pairing.starts[0])
assert observed == [other]
def test_query_profile_is_honored_for_repair(pairing, monkeypatch):
from fastapi import FastAPI
import httpx
from hermes_cli import web_server_profiles
other = pairing.home / "profiles" / "work"
other.mkdir(parents=True)
(other / "config.yaml").write_text("model: test\n", encoding="utf-8")
session = other / "platforms" / "whatsapp" / "session"
session.mkdir(parents=True)
(session / "creds.json").write_text('{"registered":true}', encoding="utf-8")
monkeypatch.setattr(web_server_profiles, "_resolve_profile_dir", lambda profile: other)
app = FastAPI()
app.include_router(messaging.router)
async def request():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
return await client.post("/api/messaging/whatsapp/onboarding/start?profile=work", json={"reset_session": True})
response = asyncio.run(request())
assert response.status_code == 200
assert response.json()["status"] == "starting"
assert pairing.starts[0][1] == session
assert (pairing.session / "creds.json").exists()
+37
View File
@@ -0,0 +1,37 @@
export interface WhatsAppOnboardingStartRequest {
mode?: "bot" | "self-chat";
allowed_users?: string;
reset_session?: boolean;
}
export interface WhatsAppOnboardingStartResponse {
pairing_id: string;
status:
| "starting"
| "installing"
| "waiting"
| "connected"
| "error"
| "expired"
| "cancelled";
qr_payload?: string | null;
expires_at: string;
mode: "bot" | "self-chat";
allowed_users: string;
account_id?: string | null;
account_name?: string | null;
account_phone?: string | null;
error?: string | null;
}
export type WhatsAppOnboardingStatusResponse = WhatsAppOnboardingStartResponse;
export interface WhatsAppOnboardingApplyResponse {
ok: boolean;
platform: "whatsapp";
needs_restart: boolean;
restart_started?: boolean;
restart_action?: string;
restart_pid?: number | null;
restart_error?: string;
}
+12
View File
@@ -114,6 +114,18 @@ describe("api.getModelOptions", () => {
});
describe("management profile scope", () => {
it("sends explicit WhatsApp re-pair consent to the selected profile", async () => {
const fetchMock = jsonFetchMock({ pairing_id: "fresh", status: "starting" });
vi.stubGlobal("fetch", fetchMock);
setManagementProfile("worker");
await api.startWhatsAppOnboarding({ mode: "bot", reset_session: true });
const [url, init] = fetchMock.mock.calls[0];
expect(url).toBe("/api/messaging/whatsapp/onboarding/start?profile=worker");
expect(JSON.parse(String(init?.body))).toEqual({ mode: "bot", reset_session: true });
});
// Every family whose routes write into a named profile's home must carry the
// scope; an unprofiled request 400s on a host that merely HAS a second profile.
it.each([
+12 -36
View File
@@ -9,6 +9,17 @@ const mutateSessionHttp = createDashboardSessionMutations(fetchJSON);
import type { AuxiliaryModelsResponse } from "./api-aux";
import type { AgentPluginUpdateResponse } from "./api-plugins";
import type {
WhatsAppOnboardingApplyResponse,
WhatsAppOnboardingStartRequest,
WhatsAppOnboardingStartResponse,
WhatsAppOnboardingStatusResponse,
} from "./api-whatsapp";
export type {
WhatsAppOnboardingApplyResponse,
WhatsAppOnboardingStartResponse,
WhatsAppOnboardingStatusResponse,
} from "./api-whatsapp";
import type {
ManagedFileReadResponse,
ManagedFilesResponse,
@@ -1004,10 +1015,7 @@ export const api = {
`/api/messaging/telegram/onboarding/${encodeURIComponent(pairingId)}`,
{ method: "DELETE" },
),
startWhatsAppOnboarding: (body: {
mode?: "bot" | "self-chat";
allowed_users?: string;
}) =>
startWhatsAppOnboarding: (body: WhatsAppOnboardingStartRequest) =>
fetchJSON<WhatsAppOnboardingStartResponse>(
"/api/messaging/whatsapp/onboarding/start",
{
@@ -2184,38 +2192,6 @@ export interface TelegramOnboardingApplyResponse {
restart_error?: string;
}
export interface WhatsAppOnboardingStartResponse {
pairing_id: string;
status:
| "starting"
| "installing"
| "waiting"
| "connected"
| "error"
| "expired"
| "cancelled";
qr_payload?: string | null;
expires_at: string;
mode: "bot" | "self-chat";
allowed_users: string;
account_id?: string | null;
account_name?: string | null;
account_phone?: string | null;
error?: string | null;
}
export type WhatsAppOnboardingStatusResponse = WhatsAppOnboardingStartResponse;
export interface WhatsAppOnboardingApplyResponse {
ok: boolean;
platform: "whatsapp";
needs_restart: boolean;
restart_started?: boolean;
restart_action?: string;
restart_pid?: number | null;
restart_error?: string;
}
export interface SessionMessage {
role: "user" | "assistant" | "system" | "tool";
content: string | null;
+27 -3
View File
@@ -35,6 +35,7 @@ import type {
} from "@/lib/api";
import { useModalBehavior } from "@/hooks/useModalBehavior";
import { AllowlistInput } from "@/components/AllowlistInput";
import { ConfirmDialog } from "@/components/ConfirmDialog";
import { usePageHeader } from "@/contexts/usePageHeader";
import { cn, themedBody } from "@/lib/utils";
import { errorMessage } from "@/lib/api-error";
@@ -684,7 +685,7 @@ export default function ChannelsPage() {
);
}
function WhatsAppOnboardingPanel({
export function WhatsAppOnboardingPanel({
onChanged,
onRestartNeeded,
platform,
@@ -712,6 +713,7 @@ function WhatsAppOnboardingPanel({
configuredMode ?? "bot",
);
const [allowedUsers, setAllowedUsers] = useState("");
const [confirmRepair, setConfirmRepair] = useState(false);
const [error, setError] = useState("");
const [tick, setTick] = useState(0);
@@ -796,7 +798,9 @@ function WhatsAppOnboardingPanel({
setError("");
};
const start = async () => {
const start = async (resetSession = false) => {
setConfirmRepair(false);
setSetup(null);
setPhase("starting");
setError("");
setQrDataUrl("");
@@ -804,6 +808,7 @@ function WhatsAppOnboardingPanel({
const res = await api.startWhatsAppOnboarding({
mode,
allowed_users: allowedUsers,
reset_session: resetSession,
});
setSetup(res);
if (res.qr_payload) {
@@ -923,6 +928,7 @@ function WhatsAppOnboardingPanel({
: !allowedUsers.trim() && hasSavedAllowedUsers
? "Hermes will keep the saved WhatsApp allowlist."
: "If no allowed numbers were entered, Hermes replies with a pairing code. Approve it from the dashboard Pairing page.";
const pairingBusy = ["starting", "waiting", "applying"].includes(phase);
return (
<div className="rounded-sm border border-border bg-background/35 p-4">
@@ -932,11 +938,19 @@ function WhatsAppOnboardingPanel({
size="sm"
className="uppercase"
onClick={() => void start()}
disabled={phase === "starting" || phase === "waiting" || phase === "applying"}
disabled={pairingBusy}
prefix={phase === "starting" ? <Spinner /> : <QrCode className="h-4 w-4" />}
>
{phase === "starting" ? "Starting…" : "Pair with QR"}
</Button>
<Button
size="sm"
outlined
onClick={() => setConfirmRepair(true)}
disabled={pairingBusy}
>
Re-pair with QR
</Button>
{platform.configured && (
<span className="text-xs text-muted-foreground">
Existing WhatsApp settings are configured.
@@ -944,6 +958,16 @@ function WhatsAppOnboardingPanel({
)}
</div>
<ConfirmDialog
open={confirmRepair}
title="Replace the saved WhatsApp link?"
description="Use this if WhatsApp logged out or removed this linked device. Disable WhatsApp and restart the gateway first. The old session will be kept in a private backup. Scan the new QR from Linked Devices on your phone, then Save and restart to enable WhatsApp again."
confirmLabel="Back up and re-pair"
destructive
onCancel={() => setConfirmRepair(false)}
onConfirm={() => void start(true)}
/>
<div className="flex flex-col gap-3 lg:flex-row lg:items-end">
<div className="grid gap-1.5">
<span className="text-xs uppercase tracking-[0.12em] text-muted-foreground">
@@ -0,0 +1,84 @@
// @vitest-environment jsdom
import { act } from "react";
import { createRoot, type Root } from "react-dom/client";
import { afterEach, beforeEach, expect, it, vi } from "vitest";
import type { MessagingPlatform } from "@/lib/api";
import { WhatsAppOnboardingPanel } from "./ChannelsPage";
const start = vi.hoisted(() => vi.fn());
vi.mock("@/lib/api", () => ({ api: { startWhatsAppOnboarding: start } }));
let root: Root;
let container: HTMLDivElement;
(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true;
const platform = {
id: "whatsapp",
configured: true,
whatsapp_setup: { mode: "bot", allowed_users_set: true },
} as MessagingPlatform;
function button(label: string) {
const found = Array.from(document.querySelectorAll("button")).find(
(element) => element.textContent?.trim() === label,
);
if (!found) throw new Error(`Missing button: ${label}`);
return found;
}
async function click(label: string) {
await act(async () => { button(label).click(); });
}
beforeEach(async () => {
start.mockReset();
start.mockResolvedValue({
pairing_id: "fresh", status: "starting", mode: "bot", allowed_users: "",
expires_at: new Date(Date.now() + 600_000).toISOString(),
});
container = document.createElement("div");
document.body.append(container);
root = createRoot(container);
await act(async () => {
root.render(
<WhatsAppOnboardingPanel
platform={platform}
onChanged={vi.fn(async () => undefined)}
onRestartNeeded={vi.fn()}
setRestartNeeded={vi.fn()}
showToast={vi.fn()}
/>,
);
});
});
afterEach(async () => {
await act(async () => { root.unmount(); });
container.remove();
});
it("requires explicit confirmation before replacing the saved link", async () => {
await click("Re-pair with QR");
expect(start).not.toHaveBeenCalled();
expect(document.querySelector('[role="dialog"]')?.textContent).toContain("private backup");
await click("Cancel");
expect(start).not.toHaveBeenCalled();
await click("Re-pair with QR");
await click("Back up and re-pair");
expect(start).toHaveBeenCalledWith({ mode: "bot", allowed_users: "", reset_session: true });
expect(button("Re-pair with QR").disabled).toBe(true);
});
it("keeps ordinary pairing non-destructive", async () => {
await click("Pair with QR");
expect(start).toHaveBeenCalledWith({ mode: "bot", allowed_users: "", reset_session: false });
});
it("shows an active-bridge refusal and allows retry", async () => {
start.mockRejectedValueOnce(new Error("The WhatsApp bridge is still running. Stop it before re-pairing."));
await click("Re-pair with QR");
await click("Back up and re-pair");
expect(container.textContent).toContain("Stop it before re-pairing");
expect(button("Re-pair with QR").disabled).toBe(false);
});