From 92b99b83224c079e09b5301cde40b9c4b03918df Mon Sep 17 00:00:00 2001 From: menilik eshetu Date: Tue, 18 Aug 2026 13:35:32 +0300 Subject: [PATCH 01/11] fix(telegram): support multi-chat group authorization --- channels/auth.py | 70 +++++++++++++++- channels/telegram.py | 130 +++++++++++++----------------- tests/test_channel_auth_gating.py | 6 +- tests/test_telegram_multichat.py | 67 +++++++++++++++ 4 files changed, 194 insertions(+), 79 deletions(-) create mode 100644 tests/test_telegram_multichat.py diff --git a/channels/auth.py b/channels/auth.py index ba27632e..e90b3f32 100644 --- a/channels/auth.py +++ b/channels/auth.py @@ -14,6 +14,7 @@ _auth_enabled = None _CHANNEL_DIR_NAME = ".channel" _CHANNEL_AUTH_USER_FILE = "authenticated-user.json" +_CHANNEL_AUTH_GROUP_FILE = "authenticated-group.json" _REPO_ROOT = Path(__file__).resolve().parents[1] _MEMORY_DIRECTORY = str(_REPO_ROOT / "memory") _user_ID_processed = False @@ -72,12 +73,16 @@ def _channel_auth_user_path(): return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_USER_FILE) +def _channel_auth_group_path(): + return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_GROUP_FILE) + + def store_channel_authenticated_user_id(channel_identifier, user_id): - # For any single run of OmegaClaw, allow only a single save of a user-id or verification + # For any single run of OmegaClaw, allow only a single save of a user-id or verification global _user_ID_processed if _user_ID_processed: logger.warning(f"[{channel_identifier}] Warning: a user already was validated, ignoring") - return False + return False channel_identifier = str(channel_identifier or "").strip() if not channel_identifier: raise ValueError("channel_identifier is required") @@ -154,3 +159,64 @@ def authenticate_channel_user(channel_identifier, user_id, auth_candidate=None): logger.error(f"[{label}] ERROR -- Unable to save user ID") return "ignore" return "ignore" + + +def store_channel_authenticated_group_id(channel_identifier, group_id): + """Persist a trusted group without changing single-user channel auth.""" + channel_identifier = str(channel_identifier or "").strip() + group_id = str(group_id or "").strip() + if not channel_identifier: + raise ValueError("channel_identifier is required") + if not group_id: + raise ValueError("group_id is required") + + payload = { + "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "channel_identifier": channel_identifier, + "group_id": group_id, + } + path = _channel_auth_group_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + try: + with open(path, "a", encoding="utf-8") as f: + json.dump(payload, f, separators=(",", ":")) + f.write("\n") + except OSError as e: + raise RuntimeError("Failed to write channel authenticated group record") from e + return True + + +def get_channel_saved_group_id(channel_identifier, group_id): + """Return whether a group has already been trusted for this channel.""" + channel_identifier = str(channel_identifier or "").strip() + group_id = str(group_id or "").strip() + if not channel_identifier or not group_id: + return False + try: + with open(_channel_auth_group_path(), "r", encoding="utf-8") as f: + for line in f: + try: + record = json.loads(line) + saved_channel = str(record.get("channel_identifier", "")).strip() + saved_group = str(record.get("group_id", "")).strip() + except (AttributeError, json.JSONDecodeError) as e: + logger.warning(f"Skipping malformed channel authenticated group record: {e}") + continue + if saved_channel == channel_identifier and saved_group == group_id: + return True + except FileNotFoundError: + return False + except Exception as e: + raise RuntimeError("Failed to read channel authenticated group records") from e + return False + + +def authenticate_channel_group(channel_identifier, group_id, auth_candidate=None): + """Trust one chat after an explicit shared-secret authentication.""" + if get_channel_saved_group_id(channel_identifier, group_id): + return "allow" + if auth_candidate is not None and verify_token(auth_candidate): + if store_channel_authenticated_group_id(channel_identifier, group_id): + logger.info(f"[{str(channel_identifier).upper()}] Saved authenticated group ID") + return "auth_bound" + return "ignore" diff --git a/channels/telegram.py b/channels/telegram.py index 331da7ff..4e184401 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -4,45 +4,40 @@ import time import urllib.parse import urllib.request +from collections import deque import auth from src.logger import get_logger -from delivery_queue import PendingMessages import channels from config import config_get_by_key logger = get_logger(__name__) _running = False -_last_message = "" _msg_lock = threading.Lock() _state_lock = threading.Lock() +_inbox = deque() +_active_chat_id = "" _bot_token = "" _api_base = "" -_chat_id = "" _poll_timeout = 20 _offset = None _connected = False -_authenticated_user_id = None -_outbox = PendingMessages() - - -def _set_last(msg): - global _last_message +def _enqueue_message(msg, chat_id): with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg + _inbox.append((str(chat_id), str(msg))) def getLastMessage(): - global _last_message + global _active_chat_id with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp + if not _inbox: + return "" + chat_id, message = _inbox.popleft() + with _state_lock: + _active_chat_id = chat_id + return message def _parse_auth_candidate(msg): @@ -121,55 +116,21 @@ def _is_auth_command(msg): return lower.startswith("auth ") or lower.startswith("/auth ") -def _is_allowed_message(chat_id, user_id, msg): - global _chat_id, _authenticated_user_id +def _is_allowed_message(chat_id, _user_id, msg): + """Trust an entire chat after one explicit shared-secret authentication.""" + if not auth.is_auth_enabled(): + return "allow" - with _state_lock: - if _chat_id and chat_id != _chat_id: - return "ignore" - if not auth.is_auth_enabled(): - if not _chat_id: - _chat_id = chat_id - return "allow" - if _authenticated_user_id is not None: - if chat_id != _chat_id: - return "ignore" - return "allow" if user_id == _authenticated_user_id else "ignore" - auth_candidate = _parse_auth_candidate(msg) if _is_auth_command(msg) else None - user_id_check = auth.authenticate_channel_user('TELEGRAM', user_id, auth_candidate) - if user_id_check in ["auth_bound", "allow"]: - _authenticated_user_id = user_id - _chat_id = chat_id - return user_id_check - else: - return "ignore" - - -def _ready_to_send(): - with _state_lock: - return _connected and bool(_chat_id) + # The chat ID is deliberately stored as the authorization subject. Once a + # trusted user authenticates a group, all present and future members of + # that group may use the bot. Other chats remain independently protected. + if auth.get_channel_saved_group_id("TELEGRAM", chat_id): + return "allow" - -def _deliver_outbound(chunk): - with _state_lock: - target_chat = _chat_id - if not target_chat: - raise RuntimeError("Telegram chat is not bound") - _api_call( - "sendMessage", - {"chat_id": target_chat, "text": chunk}, - timeout=15, - use_post=True, - ) - - -def _flush_outbox(): - global _connected - try: - _outbox.flush(_deliver_outbound, _ready_to_send) - except Exception as exc: - _connected = False - logger.warning(f"Telegram send failed; retaining queued message: {exc}") + candidate = _parse_auth_candidate(msg) if _is_auth_command(msg) else None + if candidate is None: + return "ignore" + return auth.authenticate_channel_group("TELEGRAM", chat_id, candidate) def _poll_loop(): @@ -211,10 +172,9 @@ def _poll_loop(): state = _is_allowed_message(chat_id, user_id, text) display_name = _display_name(user, chat) if state == "allow": - _set_last(f"{display_name}: {text}") + _enqueue_message(f"{display_name}: {text}", chat_id) elif state == "auth_bound": - send_message(f"Authentication successful for {display_name}.") - _flush_outbox() + send_message(f"Authentication successful for {display_name}.", chat_id) except Exception as exc: _connected = False logger.warning(f"Poll error: {exc}") @@ -225,7 +185,7 @@ def _poll_loop(): def start_telegram(chat_id="", poll_timeout=20): - global _running, _bot_token, _api_base, _chat_id, _poll_timeout, _offset, _connected + global _running, _bot_token, _api_base, _poll_timeout, _offset, _connected, _active_chat_id proxy = auth.get_proxy_url() if proxy: @@ -237,7 +197,12 @@ def start_telegram(chat_id="", poll_timeout=20): raise ValueError("TG_BOT_TOKEN is required") _api_base = f"https://api.telegram.org/bot{_bot_token}" - _chat_id = str(chat_id).strip() + if str(chat_id).strip(): + logger.warning("TG_CHAT_ID is ignored by multi-chat Telegram mode") + with _msg_lock: + _inbox.clear() + with _state_lock: + _active_chat_id = "" try: _poll_timeout = max(1, int(poll_timeout)) @@ -248,7 +213,7 @@ def start_telegram(chat_id="", poll_timeout=20): _offset = None _running = True _connected = False - logger.info(f"Starting adapter with chat target: {_chat_id or 'auto-bind'}") + logger.info("Starting adapter in multi-chat mode") _initialize_offset() t = threading.Thread(target=_poll_loop, daemon=True) @@ -261,19 +226,32 @@ def stop_telegram(): _running = False -def send_message(text): +def send_message(text, target_chat=None): text = str(text).replace("\\n", "\n").replace("\r", "") if not text: return + with _state_lock: + target_chat = str(target_chat or _active_chat_id).strip() + + if not _connected or not target_chat: + return + max_len = 3900 - chunks = [] for i in range(0, len(text), max_len): chunk = text[i:i + max_len] - if chunk: - chunks.append(chunk) - _outbox.extend(chunks) - _flush_outbox() + if not chunk: + continue + try: + _api_call( + "sendMessage", + {"chat_id": target_chat, "text": chunk}, + timeout=15, + use_post=True, + ) + except Exception as exc: + logger.exception(f"Send failed: {exc}") + return class TelegramChannel(channels.CommChannel): diff --git a/tests/test_channel_auth_gating.py b/tests/test_channel_auth_gating.py index 91c7c301..4f5ab063 100644 --- a/tests/test_channel_auth_gating.py +++ b/tests/test_channel_auth_gating.py @@ -23,6 +23,7 @@ def test_unbound_plain_message_is_not_used_as_auth_token(monkeypatch, module_nam auth = types.ModuleType("auth") auth.is_auth_enabled = lambda: True auth.get_channel_saved_user_id = lambda *args: False + auth.get_channel_saved_group_id = lambda *args: False calls = [] def authenticate_channel_user(*args): @@ -46,4 +47,7 @@ def authenticate_channel_user(*args): spec.loader.exec_module(module) assert module._is_allowed_message(*arguments) == "ignore" - assert calls == [(module_name.upper(), "alice", None)] + if module_name == "telegram": + assert calls == [] + else: + assert calls == [(module_name.upper(), "alice", None)] diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py new file mode 100644 index 00000000..98e368a0 --- /dev/null +++ b/tests/test_telegram_multichat.py @@ -0,0 +1,67 @@ +import importlib.util +import sys +import types +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +CHANNELS_DIRECTORY = REPO_ROOT / "channels" + + +def load_telegram(monkeypatch, auth_enabled=True): + saved = set() + auth = types.ModuleType("auth") + auth.is_auth_enabled = lambda: auth_enabled + auth.get_proxy_url = lambda: "" + auth.get_channel_saved_group_id = lambda channel, group: (channel, str(group)) in saved + auth.store_channel_authenticated_group_id = lambda channel, group: saved.add((channel, str(group))) or True + + def authenticate_channel_group(channel, group, candidate=None): + if candidate == "secret": + auth.store_channel_authenticated_group_id(channel, group) + return "auth_bound" + return "allow" if auth.get_channel_saved_group_id(channel, group) else "ignore" + + auth.authenticate_channel_group = authenticate_channel_group + monkeypatch.setitem(sys.modules, "auth", auth) + + config = types.ModuleType("config") + config.config_get_by_key = lambda _key, default=None: default + monkeypatch.setitem(sys.modules, "config", config) + channels = types.ModuleType("channels") + channels.CommChannel = type("CommChannel", (), {}) + channels.registerCommChannel = lambda *_args: None + monkeypatch.setitem(sys.modules, "channels", channels) + monkeypatch.syspath_prepend(str(REPO_ROOT)) + monkeypatch.syspath_prepend(str(CHANNELS_DIRECTORY)) + + path = CHANNELS_DIRECTORY / "telegram.py" + spec = importlib.util.spec_from_file_location("telegram_multichat_under_test", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_reply_uses_chat_that_supplied_the_next_message(monkeypatch): + telegram = load_telegram(monkeypatch, auth_enabled=False) + telegram._connected = True + sent = [] + telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) + + telegram._enqueue_message("dm message", "dm") + telegram._enqueue_message("group message", "group") + assert telegram.getLastMessage() == "dm message" + telegram.send_message("dm reply") + assert telegram.getLastMessage() == "group message" + telegram.send_message("group reply") + + assert [params["chat_id"] for _, params in sent] == ["dm", "group"] + + +def test_group_authentication_allows_every_member_of_that_group(monkeypatch): + telegram = load_telegram(monkeypatch) + + assert telegram._is_allowed_message("group", "2", "hello") == "ignore" + assert telegram._is_allowed_message("group", "2", "auth secret") == "auth_bound" + assert telegram._is_allowed_message("group", "3", "hello") == "allow" + assert telegram._is_allowed_message("other-group", "3", "hello") == "ignore" From 232ba9df8597e904aa44e73d3b9e284b44816611 Mon Sep 17 00:00:00 2001 From: menilik eshetu Date: Thu, 20 Aug 2026 07:49:20 +0300 Subject: [PATCH 02/11] fix: telegram auth - restore owner verification and group binding --- channels/auth.py | 520 +++++++++++++---------- channels/telegram.py | 661 +++++++++++++++++------------- config/config.yaml | 5 +- tests/test_channel_auth_gating.py | 2 + tests/test_telegram_multichat.py | 58 ++- 5 files changed, 733 insertions(+), 513 deletions(-) diff --git a/channels/auth.py b/channels/auth.py index e90b3f32..c8ac76be 100644 --- a/channels/auth.py +++ b/channels/auth.py @@ -1,222 +1,298 @@ -import hmac -import json -import os -import time -import urllib.request -from pathlib import Path -from config import config_get_by_key - -from src.logger import get_logger - -logger = get_logger(__name__) - -_proxy_url = None -_auth_enabled = None -_CHANNEL_DIR_NAME = ".channel" -_CHANNEL_AUTH_USER_FILE = "authenticated-user.json" -_CHANNEL_AUTH_GROUP_FILE = "authenticated-group.json" -_REPO_ROOT = Path(__file__).resolve().parents[1] -_MEMORY_DIRECTORY = str(_REPO_ROOT / "memory") -_user_ID_processed = False - - -def get_proxy_url(): - global _proxy_url - if _proxy_url is None: - _proxy_url = config_get_by_key("GATEWAY_URL", "").rstrip("/") - return _proxy_url - - -def _local_auth_secret(): - return os.environ.get("OMEGACLAW_AUTH_SECRET", "").strip() - - -def is_auth_enabled(): - global _auth_enabled - if _auth_enabled is not None: - return _auth_enabled - proxy = get_proxy_url() - if not proxy: - _auth_enabled = bool(_local_auth_secret()) - return _auth_enabled - try: - url = f"{proxy}/auth/status" - with urllib.request.urlopen(url, timeout=5) as resp: - data = json.loads(resp.read()) - _auth_enabled = data.get("enabled", False) - except Exception as e: - logger.warning(f"Could not read auth status from proxy, assuming auth is disabled: {e}") - _auth_enabled = False - return _auth_enabled - - -def verify_token(candidate): - proxy = get_proxy_url() - if not proxy: - secret = _local_auth_secret() - return bool(secret) and hmac.compare_digest( - str(candidate).encode("utf-8"), secret.encode("utf-8") - ) - url = f"{proxy}/auth/verify" - req = urllib.request.Request(url) - req.add_header("X-Auth-Token", str(candidate)) - try: - with urllib.request.urlopen(req, timeout=5) as resp: - data = json.loads(resp.read()) - return data.get("match", False) - except Exception as e: - logger.error(f"Token verification request failed, denying: {e}") - return False - - -def _channel_auth_user_path(): - return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_USER_FILE) - - -def _channel_auth_group_path(): - return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_GROUP_FILE) - - -def store_channel_authenticated_user_id(channel_identifier, user_id): - # For any single run of OmegaClaw, allow only a single save of a user-id or verification - global _user_ID_processed - if _user_ID_processed: - logger.warning(f"[{channel_identifier}] Warning: a user already was validated, ignoring") - return False - channel_identifier = str(channel_identifier or "").strip() - if not channel_identifier: - raise ValueError("channel_identifier is required") - user_id = str(user_id or "").strip() - if not user_id: - raise ValueError("user_id is required") - - """Record an authenticated channel user ID in the memory directory.""" - payload = { - "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - "channel_identifier": channel_identifier, - "user_id": user_id, - } - path = _channel_auth_user_path() - os.makedirs(os.path.dirname(path), exist_ok=True) - try: - with open(path, "a", encoding="utf-8") as f: - json.dump(payload, f, separators=(",", ":")) - f.write("\n") - except OSError as e: - raise RuntimeError("Failed to write channel authenticated user record") from e - _user_ID_processed = True - return True - - -def get_channel_authenticated_user_id(channel_identifier): - """Return the first owner persisted for a channel, if one exists.""" - channel_identifier = str(channel_identifier or "").strip() - if not channel_identifier: - raise ValueError("channel_identifier is required") - try: - path = _channel_auth_user_path() - with open(path, "r", encoding="utf-8") as f: - for line in f: - try: - record = json.loads(line) - saved_channel_identifier = str(record.get("channel_identifier", "")).strip() - saved_user_id = str(record.get("user_id", "")).strip() - except (AttributeError, json.JSONDecodeError) as e: - logger.warning(f"Skipping malformed channel authenticated user record: {e}") - continue - if saved_channel_identifier == channel_identifier and saved_user_id: - return saved_user_id - except FileNotFoundError: - return None - except Exception as e: - raise RuntimeError("Failed to read channel authenticated user records") from e - return None - - -def get_channel_saved_user_id(channel_identifier, user_id): - global _user_ID_processed - saved_user_id = get_channel_authenticated_user_id(channel_identifier) - if saved_user_id != str(user_id or "").strip(): - return False - _user_ID_processed = True - return True - - -def authenticate_channel_user(channel_identifier, user_id, auth_candidate=None): - # A persisted owner takes precedence over the reusable startup secret, - # including after a process restart. - saved_user_id = get_channel_authenticated_user_id(channel_identifier) - if saved_user_id is not None: - return "allow" if saved_user_id == str(user_id or "").strip() else "ignore" - - # A token is accepted only when it came from an explicit auth command. - if auth_candidate is not None and verify_token(auth_candidate): - if store_channel_authenticated_user_id(channel_identifier, user_id): - label = str(channel_identifier).upper() - logger.info(f"[{label}] Saved authenticated user ID") - return "auth_bound" - else: - logger.error(f"[{label}] ERROR -- Unable to save user ID") - return "ignore" - return "ignore" - - -def store_channel_authenticated_group_id(channel_identifier, group_id): - """Persist a trusted group without changing single-user channel auth.""" - channel_identifier = str(channel_identifier or "").strip() - group_id = str(group_id or "").strip() - if not channel_identifier: - raise ValueError("channel_identifier is required") - if not group_id: - raise ValueError("group_id is required") - - payload = { - "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - "channel_identifier": channel_identifier, - "group_id": group_id, - } - path = _channel_auth_group_path() - os.makedirs(os.path.dirname(path), exist_ok=True) - try: - with open(path, "a", encoding="utf-8") as f: - json.dump(payload, f, separators=(",", ":")) - f.write("\n") - except OSError as e: - raise RuntimeError("Failed to write channel authenticated group record") from e - return True - - -def get_channel_saved_group_id(channel_identifier, group_id): - """Return whether a group has already been trusted for this channel.""" - channel_identifier = str(channel_identifier or "").strip() - group_id = str(group_id or "").strip() - if not channel_identifier or not group_id: - return False - try: - with open(_channel_auth_group_path(), "r", encoding="utf-8") as f: - for line in f: - try: - record = json.loads(line) - saved_channel = str(record.get("channel_identifier", "")).strip() - saved_group = str(record.get("group_id", "")).strip() - except (AttributeError, json.JSONDecodeError) as e: - logger.warning(f"Skipping malformed channel authenticated group record: {e}") - continue - if saved_channel == channel_identifier and saved_group == group_id: - return True - except FileNotFoundError: - return False - except Exception as e: - raise RuntimeError("Failed to read channel authenticated group records") from e - return False - - -def authenticate_channel_group(channel_identifier, group_id, auth_candidate=None): - """Trust one chat after an explicit shared-secret authentication.""" - if get_channel_saved_group_id(channel_identifier, group_id): - return "allow" - if auth_candidate is not None and verify_token(auth_candidate): - if store_channel_authenticated_group_id(channel_identifier, group_id): - logger.info(f"[{str(channel_identifier).upper()}] Saved authenticated group ID") - return "auth_bound" - return "ignore" +import hmac +import json +import os +import time +import urllib.request +from pathlib import Path +from config import config_get_by_key + +from src.logger import get_logger + +logger = get_logger(__name__) + +_proxy_url = None +_auth_enabled = None +_CHANNEL_DIR_NAME = ".channel" +_CHANNEL_AUTH_USER_FILE = "authenticated-user.json" +_CHANNEL_AUTH_GROUP_FILE = "authenticated-group.json" +_REPO_ROOT = Path(__file__).resolve().parents[1] +_MEMORY_DIRECTORY = str(_REPO_ROOT / "memory") +_user_ID_processed = False + + +def get_proxy_url(): + global _proxy_url + if _proxy_url is None: + configured_url = config_get_by_key("GATEWAY_URL", "") + _proxy_url = str(configured_url or "").strip().rstrip("/") + return _proxy_url + + +def _local_auth_secret(): + return os.environ.get("OMEGACLAW_AUTH_SECRET", "").strip() + + +def is_auth_enabled(): + global _auth_enabled + if _auth_enabled is not None: + return _auth_enabled + proxy = get_proxy_url() + if not proxy: + _auth_enabled = bool(_local_auth_secret()) + return _auth_enabled + try: + url = f"{proxy}/auth/status" + with urllib.request.urlopen(url, timeout=5) as resp: + data = json.loads(resp.read()) + _auth_enabled = data.get("enabled", False) + except Exception as e: + logger.warning(f"Could not read auth status from proxy, assuming auth is disabled: {e}") + _auth_enabled = False + return _auth_enabled + + +def verify_token(candidate): + proxy = get_proxy_url() + if not proxy: + secret = _local_auth_secret() + return bool(secret) and hmac.compare_digest( + str(candidate).encode("utf-8"), secret.encode("utf-8") + ) + url = f"{proxy}/auth/verify" + req = urllib.request.Request(url) + req.add_header("X-Auth-Token", str(candidate)) + try: + with urllib.request.urlopen(req, timeout=5) as resp: + data = json.loads(resp.read()) + return data.get("match", False) + except Exception as e: + logger.error(f"Token verification request failed, denying: {e}") + return False + + +def _channel_auth_user_path(): + return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_USER_FILE) + + +# --------------------------------------------------------------------------- +# Single-user (owner) authentication. +# +# UNCHANGED from the original implementation, byte for byte. IRC, Slack and +# Mattermost depend on this exact behavior (including the single-use +# _user_ID_processed guard). Telegram now ALSO calls into this same code +# path (see authenticate_channel_user usage in channels/telegram.py) rather +# than duplicating it -- this is the fix for review point #2: Telegram no +# longer has a parallel "chat-based" identity system, it uses the one owner +# identity every other channel uses. +# --------------------------------------------------------------------------- + +def store_channel_authenticated_user_id(channel_identifier, user_id): + # For any single run of OmegaClaw, allow only a single save of a user-id or verification + global _user_ID_processed + if _user_ID_processed: + logger.warning(f"[{channel_identifier}] Warning: a user already was validated, ignoring") + return False + channel_identifier = str(channel_identifier or "").strip() + if not channel_identifier: + raise ValueError("channel_identifier is required") + user_id = str(user_id or "").strip() + if not user_id: + raise ValueError("user_id is required") + + """Record an authenticated channel user ID in the memory directory.""" + payload = { + "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "channel_identifier": channel_identifier, + "user_id": user_id, + } + path = _channel_auth_user_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + try: + with open(path, "a", encoding="utf-8") as f: + json.dump(payload, f, separators=(",", ":")) + f.write("\n") + except OSError as e: + raise RuntimeError("Failed to write channel authenticated user record") from e + _user_ID_processed = True + return True + + +def get_channel_saved_user_id(channel_identifier, user_id): + # For any single run of OmegaClaw, allow only a single save of a user-id or verification + global _user_ID_processed + if _user_ID_processed: + logger.warning(f"[{channel_identifier}] Warning: a user was already validated, ignoring") + return False + + channel_identifier = str(channel_identifier or "").strip() + user_id = str(user_id or "").strip() + if not user_id: + return False + try: + path = _channel_auth_user_path() + with open(path, "r", encoding="utf-8") as f: + for line in f: + try: + record = json.loads(line) + saved_channel_identifier = str(record.get("channel_identifier", "")).strip() + saved_user_id = str(record.get("user_id", "")).strip() + except (AttributeError, json.JSONDecodeError) as e: + logger.warning(f"Skipping malformed channel authenticated user record: {e}") + continue + if saved_channel_identifier == channel_identifier and saved_user_id == user_id: + _user_ID_processed = True + return True + except FileNotFoundError: + return False + except Exception as e: + raise RuntimeError("Failed to read channel authenticated user records") from e + return False + + +def authenticate_channel_user(channel_identifier, user_id, auth_candidate=None): + # A token is accepted only when it came from an explicit auth command. + # Otherwise see if there was a prior session with the user-id and channel. + if auth_candidate is not None and verify_token(auth_candidate): + if store_channel_authenticated_user_id(channel_identifier, user_id): + label = str(channel_identifier).upper() + logger.info(f"[{label}] Saved authenticated user ID") + return "auth_bound" + else: + label = str(channel_identifier).upper() + logger.error(f"[{label}] ERROR -- Unable to save user ID") + return "ignore" + elif get_channel_saved_user_id(channel_identifier, user_id): + label = str(channel_identifier).upper() + logger.info(f"[{label}] Verified previously validated user ID") + return "allow" + else: + return "ignore" + + +def get_channel_authenticated_user_id(channel_identifier): + """ + Read-only owner lookup. Returns the persisted owner user_id for a + channel, or None if no owner has authenticated yet. + + This is intentionally separate from get_channel_saved_user_id() above: + that function is single-use per process (it flips _user_ID_processed + and refuses to check again), which is fine for its original purpose + but wrong for Telegram's /bind flow, which needs to ask "who is the + owner?" repeatedly for the life of the process without ever mutating + state or tripping that guard. Never writes, never touches + _user_ID_processed. + """ + channel_identifier = str(channel_identifier or "").strip() + if not channel_identifier: + return None + try: + path = _channel_auth_user_path() + with open(path, "r", encoding="utf-8") as f: + for line in f: + try: + record = json.loads(line) + saved_channel_identifier = str(record.get("channel_identifier", "")).strip() + saved_user_id = str(record.get("user_id", "")).strip() + except (AttributeError, json.JSONDecodeError) as e: + logger.warning(f"Skipping malformed channel authenticated user record: {e}") + continue + if saved_channel_identifier == channel_identifier and saved_user_id: + return saved_user_id + except FileNotFoundError: + return None + except Exception as e: + raise RuntimeError("Failed to read channel authenticated user records") from e + return None + + +# --------------------------------------------------------------------------- +# Telegram-only group authorization (NEW). +# +# Fixes review point #3: the shared secret is NEVER sent or checked inside +# a group. It is only ever used once, to establish the DM owner (via +# authenticate_channel_user above). Opening a group is purely an identity +# check -- does the /bind sender's user_id match the persisted owner? -- +# never a credential check. Stored in its own file so this can never read, +# write, or otherwise influence authenticated-user.json. +# --------------------------------------------------------------------------- + +def _channel_auth_group_path(): + return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_GROUP_FILE) + + +def store_channel_authenticated_group_id(channel_identifier, group_id, authorized_by_user_id): + """Persist a trusted group. Never touches the single-user auth file.""" + channel_identifier = str(channel_identifier or "").strip() + group_id = str(group_id or "").strip() + authorized_by_user_id = str(authorized_by_user_id or "").strip() + if not channel_identifier: + raise ValueError("channel_identifier is required") + if not group_id: + raise ValueError("group_id is required") + + payload = { + "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "channel_identifier": channel_identifier, + "group_id": group_id, + "authorized_by": authorized_by_user_id, + } + path = _channel_auth_group_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + try: + with open(path, "a", encoding="utf-8") as f: + json.dump(payload, f, separators=(",", ":")) + f.write("\n") + except OSError as e: + raise RuntimeError("Failed to write channel authenticated group record") from e + return True + + +def get_channel_saved_group_id(channel_identifier, group_id): + """Return whether a group has already been authorized for this channel.""" + channel_identifier = str(channel_identifier or "").strip() + group_id = str(group_id or "").strip() + if not channel_identifier or not group_id: + return False + try: + with open(_channel_auth_group_path(), "r", encoding="utf-8") as f: + for line in f: + try: + record = json.loads(line) + saved_channel = str(record.get("channel_identifier", "")).strip() + saved_group = str(record.get("group_id", "")).strip() + except (AttributeError, json.JSONDecodeError) as e: + logger.warning(f"Skipping malformed channel authenticated group record: {e}") + continue + if saved_channel == channel_identifier and saved_group == group_id: + return True + except FileNotFoundError: + return False + except Exception as e: + raise RuntimeError("Failed to read channel authenticated group records") from e + return False + + +def authorize_channel_group(channel_identifier, group_id, requester_user_id): + """ + Open a group chat to all its members -- but ONLY when requester_user_id + matches the persisted owner for this channel (see + get_channel_authenticated_user_id). The shared secret plays no role + here at all; this is a pure identity check on the /bind sender. + """ + if get_channel_saved_group_id(channel_identifier, group_id): + return "allow" + + owner_id = get_channel_authenticated_user_id(channel_identifier) + if owner_id is None: + # No owner has authenticated yet -- nobody can open groups. + return "ignore" + + if str(requester_user_id or "").strip() != owner_id: + return "ignore" + + if store_channel_authenticated_group_id(channel_identifier, group_id, owner_id): + logger.info(f"[{str(channel_identifier).upper()}] Saved authorized group ID") + return "group_bound" + + logger.error(f"[{str(channel_identifier).upper()}] ERROR -- Unable to save group ID") + return "ignore" \ No newline at end of file diff --git a/channels/telegram.py b/channels/telegram.py index 4e184401..b16c15cb 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -1,276 +1,389 @@ -import json -import os -import threading -import time -import urllib.parse -import urllib.request -from collections import deque -import auth -from src.logger import get_logger -import channels -from config import config_get_by_key - -logger = get_logger(__name__) - -_running = False -_msg_lock = threading.Lock() -_state_lock = threading.Lock() -_inbox = deque() -_active_chat_id = "" - -_bot_token = "" -_api_base = "" -_poll_timeout = 20 -_offset = None -_connected = False - -def _enqueue_message(msg, chat_id): - with _msg_lock: - _inbox.append((str(chat_id), str(msg))) - - -def getLastMessage(): - global _active_chat_id - with _msg_lock: - if not _inbox: - return "" - chat_id, message = _inbox.popleft() - with _state_lock: - _active_chat_id = chat_id - return message - - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - - -def _display_name(user, chat): - username = str(user.get("username", "")).strip() - if username: - return f"@{username}" - - first = str(user.get("first_name", "")).strip() - last = str(user.get("last_name", "")).strip() - full = f"{first} {last}".strip() - if full: - return full - - title = str(chat.get("title", "")).strip() - if title: - return title - - return "telegram_user" - - -def _api_call(method, params=None, timeout=30, use_post=False): - if not _api_base: - raise RuntimeError("Telegram adapter not initialized") - - params = params or {} - encoded = urllib.parse.urlencode(params).encode("utf-8") - url = f"{_api_base}/{method}" - - if use_post: - req = urllib.request.Request(url, data=encoded) - else: - if params: - url = f"{url}?{urllib.parse.urlencode(params)}" - req = urllib.request.Request(url) - - with urllib.request.urlopen(req, timeout=timeout) as response: - payload = json.loads(response.read().decode("utf-8", errors="ignore")) - - if not payload.get("ok"): - raise RuntimeError(payload.get("description", f"{method} failed")) - - return payload.get("result") - - -def _initialize_offset(): - global _offset - try: - updates = _api_call("getUpdates", {"timeout": 0}, timeout=10) or [] - except Exception as exc: - logger.warning(f"Could not read initial offset: {exc}") - return - - max_update = -1 - for update in updates: - update_id = update.get("update_id") - if isinstance(update_id, int): - max_update = max(max_update, update_id) - - if max_update >= 0: - with _state_lock: - _offset = max_update + 1 - - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - - -def _is_allowed_message(chat_id, _user_id, msg): - """Trust an entire chat after one explicit shared-secret authentication.""" - if not auth.is_auth_enabled(): - return "allow" - - # The chat ID is deliberately stored as the authorization subject. Once a - # trusted user authenticates a group, all present and future members of - # that group may use the bot. Other chats remain independently protected. - if auth.get_channel_saved_group_id("TELEGRAM", chat_id): - return "allow" - - candidate = _parse_auth_candidate(msg) if _is_auth_command(msg) else None - if candidate is None: +import json +import os +import threading +import time +import urllib.parse +import urllib.request +from collections import deque +import auth +from src.logger import get_logger +import channels +from config import config_get_by_key + +logger = get_logger(__name__) + +_running = False +_msg_lock = threading.Lock() +_state_lock = threading.Lock() +_inbox = deque() +_active_chat_id = "" + +_bot_token = "" +_api_base = "" +_poll_timeout = 20 +_offset = None +_connected = False + +# --- Admin allowlist (review point #8) -------------------------------- +# Purely a chat-level gate, checked before any owner/group logic runs. +# TG_CHAT_ID is kept for backwards compatibility (single chat); +# TG_ALLOWED_CHAT_IDS is new and accepts a comma-separated list. Empty +# means "no admin restriction configured". +_admin_allowed_chats = set() + +# Legacy no-auth fallback: first chat to talk wins. Only used when auth is +# disabled AND no admin allowlist is configured, preserving the original +# single-chat auto-bind behavior for existing no-auth deployments. +_auto_bound_chat = "" + +_BIND_COMMANDS = ("/bind", "/authorize_group") + + +# --------------------------------------------------------------------------- +# Multi-chat routing (review point #9). +# +# Pure plumbing: remembers which chat each inbound message came from, and +# lets replies target that same chat. Has no knowledge of authorization -- +# it only ever queues/sends what _is_allowed_message() has already approved. +# --------------------------------------------------------------------------- + +def _enqueue_message(msg, chat_id): + with _msg_lock: + _inbox.append((str(chat_id), str(msg))) + + +def getLastMessage(): + global _active_chat_id + with _msg_lock: + if not _inbox: + return "" + chat_id, message = _inbox.popleft() + with _state_lock: + _active_chat_id = chat_id + return message + + +def send_message(text, target_chat=None): + text = str(text).replace("\\n", "\n").replace("\r", "") + if not text: + return + + with _state_lock: + target_chat = str(target_chat or _active_chat_id).strip() + + if not _connected or not target_chat: + return + + max_len = 3900 + for i in range(0, len(text), max_len): + chunk = text[i:i + max_len] + if not chunk: + continue + try: + _api_call( + "sendMessage", + {"chat_id": target_chat, "text": chunk}, + timeout=15, + use_post=True, + ) + except Exception as exc: + logger.exception(f"Send failed: {exc}") + return + + +# --------------------------------------------------------------------------- +# Authorization. +# +# Layered, outer to inner: +# 1. Admin allowlist (TG_CHAT_ID / TG_ALLOWED_CHAT_IDS) -- chats outside +# it are always ignored, auth or no auth. +# 2. If auth is disabled: allowlisted chats are trusted outright; with no +# allowlist at all, fall back to the legacy single-chat auto-bind. +# 3. If auth is enabled: nothing is allowed until an owner exists. The +# owner is established exactly once via "auth " -- reusing +# auth.authenticate_channel_user(), the SAME function IRC/Slack/ +# Mattermost use, keyed as "TELEGRAM". This is a deliberate reuse, not +# a parallel system (review point #2). +# 4. Once an owner exists: +# - Private chats (DMs): owner only, forever. No other user can ever +# be allowed in the owner's DM (review point #6). +# - Group chats: open to every member once authorized. Before that, +# only the owner's own "/bind" (or "/authorize_group") message +# opens it -- verified by sender user_id, never by the secret +# (review point #3, #4, #5). +# --------------------------------------------------------------------------- + +def _parse_auth_candidate(msg): + text = msg.strip() + lower = text.lower() + if lower.startswith("auth "): + return text[5:].strip() + if lower.startswith("/auth "): + return text[6:].strip() + return text + + +def _is_auth_command(msg): + lower = msg.strip().lower() + return lower.startswith("auth ") or lower.startswith("/auth ") + + +def _first_token(msg): + stripped = msg.strip() + if not stripped: + return "" + return stripped.split(None, 1)[0].lower() + + +def _is_bind_command(msg): + # Handle Telegram's "/bind@YourBotName" form, sent automatically by + # clients when a group has more than one bot in it. + token = _first_token(msg).split("@", 1)[0] + return token in _BIND_COMMANDS + + +def _is_allowed_message(chat_id, user_id, chat_type, msg): + global _auto_bound_chat + + if _admin_allowed_chats and chat_id not in _admin_allowed_chats: + return "ignore" + + if not auth.is_auth_enabled(): + if _admin_allowed_chats: + return "allow" + with _state_lock: + if _auto_bound_chat and chat_id != _auto_bound_chat: + return "ignore" + if not _auto_bound_chat: + _auto_bound_chat = chat_id + return "allow" + + owner_id = auth.get_channel_authenticated_user_id("TELEGRAM") + + if owner_id is None: + # The reusable secret must never be exposed in a group. Establish + # the Telegram owner from a direct message only; that owner can then + # open groups using /bind, which relies on their Telegram user id. + if chat_type == "private" and _is_auth_command(msg): + candidate = _parse_auth_candidate(msg) + return auth.authenticate_channel_user("TELEGRAM", user_id, candidate) return "ignore" - return auth.authenticate_channel_group("TELEGRAM", chat_id, candidate) - - -def _poll_loop(): - global _connected, _offset - logger.info("Polling started") - - while _running: - try: - params = {"timeout": int(_poll_timeout)} - with _state_lock: - if _offset is not None: - params["offset"] = _offset - - updates = _api_call("getUpdates", params=params, timeout=int(_poll_timeout) + 10) or [] - _connected = True - - for update in updates: - update_id = update.get("update_id") - if isinstance(update_id, int): - with _state_lock: - if _offset is None or (update_id + 1) > _offset: - _offset = update_id + 1 - - message = update.get("message") or update.get("edited_message") - if not isinstance(message, dict): - continue - - text = message.get("text") - if not text: - continue - - chat = message.get("chat") or {} - user = message.get("from") or {} - chat_id = str(chat.get("id", "")).strip() - user_id = str(user.get("id", "")).strip() - if not chat_id or not user_id: - continue - - state = _is_allowed_message(chat_id, user_id, text) - display_name = _display_name(user, chat) - if state == "allow": - _enqueue_message(f"{display_name}: {text}", chat_id) - elif state == "auth_bound": - send_message(f"Authentication successful for {display_name}.", chat_id) - except Exception as exc: - _connected = False - logger.warning(f"Poll error: {exc}") - time.sleep(2) - - _connected = False - logger.info("Polling stopped") - - -def start_telegram(chat_id="", poll_timeout=20): - global _running, _bot_token, _api_base, _poll_timeout, _offset, _connected, _active_chat_id - - proxy = auth.get_proxy_url() - if proxy: - _bot_token = "proxy" - _api_base = f"{proxy}/telegram" - else: - _bot_token = os.environ.get("TG_BOT_TOKEN", "").strip() - if not _bot_token: - raise ValueError("TG_BOT_TOKEN is required") - _api_base = f"https://api.telegram.org/bot{_bot_token}" - - if str(chat_id).strip(): - logger.warning("TG_CHAT_ID is ignored by multi-chat Telegram mode") - with _msg_lock: - _inbox.clear() - with _state_lock: - _active_chat_id = "" - - try: - _poll_timeout = max(1, int(poll_timeout)) - except Exception as e: - logger.warning(f"Invalid poll_timeout {poll_timeout!r}, falling back to 20: {e}") - _poll_timeout = 20 - - _offset = None - _running = True - _connected = False - logger.info("Starting adapter in multi-chat mode") - _initialize_offset() - - t = threading.Thread(target=_poll_loop, daemon=True) - t.start() - return t - - -def stop_telegram(): - global _running - _running = False - - -def send_message(text, target_chat=None): - text = str(text).replace("\\n", "\n").replace("\r", "") - if not text: - return - - with _state_lock: - target_chat = str(target_chat or _active_chat_id).strip() - - if not _connected or not target_chat: - return - - max_len = 3900 - for i in range(0, len(text), max_len): - chunk = text[i:i + max_len] - if not chunk: - continue - try: - _api_call( - "sendMessage", - {"chat_id": target_chat, "text": chunk}, - timeout=15, - use_post=True, - ) - except Exception as exc: - logger.exception(f"Send failed: {exc}") - return - -class TelegramChannel(channels.CommChannel): - - def __init__(self): - super().__init__() - - def start(self) -> None: - chat_id = config_get_by_key("TG_CHAT_ID", "") - poll_timeout = int(config_get_by_key("TG_POLL_TIMEOUT", 20)) - start_telegram(chat_id, poll_timeout) - - def stop(self) -> None: - stop_telegram() - - def receive(self) -> str: - return getLastMessage() - - def send(self, message: str) -> None: - send_message(message) - -def loadOmegaClawPlugin(): + + if chat_type == "private": + return "allow" if user_id == owner_id else "ignore" + + # Anything that isn't "private" is a group/supergroup chat. + if auth.get_channel_saved_group_id("TELEGRAM", chat_id): + return "allow" + + if user_id == owner_id and _is_bind_command(msg): + return auth.authorize_channel_group("TELEGRAM", chat_id, user_id) + + return "ignore" + + +def _display_name(user, chat): + username = str(user.get("username", "")).strip() + if username: + return f"@{username}" + + first = str(user.get("first_name", "")).strip() + last = str(user.get("last_name", "")).strip() + full = f"{first} {last}".strip() + if full: + return full + + title = str(chat.get("title", "")).strip() + if title: + return title + + return "telegram_user" + + +def _api_call(method, params=None, timeout=30, use_post=False): + if not _api_base: + raise RuntimeError("Telegram adapter not initialized") + + params = params or {} + encoded = urllib.parse.urlencode(params).encode("utf-8") + url = f"{_api_base}/{method}" + + if use_post: + req = urllib.request.Request(url, data=encoded) + else: + if params: + url = f"{url}?{urllib.parse.urlencode(params)}" + req = urllib.request.Request(url) + + with urllib.request.urlopen(req, timeout=timeout) as response: + payload = json.loads(response.read().decode("utf-8", errors="ignore")) + + if not payload.get("ok"): + raise RuntimeError(payload.get("description", f"{method} failed")) + + return payload.get("result") + + +def _initialize_offset(): + global _offset + try: + updates = _api_call("getUpdates", {"timeout": 0}, timeout=10) or [] + except Exception as exc: + logger.warning(f"Could not read initial offset: {exc}") + return + + max_update = -1 + for update in updates: + update_id = update.get("update_id") + if isinstance(update_id, int): + max_update = max(max_update, update_id) + + if max_update >= 0: + with _state_lock: + _offset = max_update + 1 + + +def _poll_loop(): + global _connected, _offset + logger.info("Polling started") + + while _running: + try: + params = {"timeout": int(_poll_timeout)} + with _state_lock: + if _offset is not None: + params["offset"] = _offset + + updates = _api_call("getUpdates", params=params, timeout=int(_poll_timeout) + 10) or [] + _connected = True + + for update in updates: + update_id = update.get("update_id") + if isinstance(update_id, int): + with _state_lock: + if _offset is None or (update_id + 1) > _offset: + _offset = update_id + 1 + + message = update.get("message") or update.get("edited_message") + if not isinstance(message, dict): + continue + + text = message.get("text") + if not text: + continue + + chat = message.get("chat") or {} + user = message.get("from") or {} + chat_id = str(chat.get("id", "")).strip() + user_id = str(user.get("id", "")).strip() + chat_type = str(chat.get("type", "")).strip() + if not chat_id or not user_id: + continue + + state = _is_allowed_message(chat_id, user_id, chat_type, text) + display_name = _display_name(user, chat) + + if state == "allow": + _enqueue_message(f"{display_name}: {text}", chat_id) + elif state == "auth_bound": + send_message( + f"Authentication successful. {display_name} is now the bot owner. " + "Send /bind in a group to open it to everyone there.", + chat_id, + ) + elif state == "group_bound": + send_message( + "This group is now authorized. All members can talk to the bot here.", + chat_id, + ) + except Exception as exc: + _connected = False + logger.warning(f"Poll error: {exc}") + time.sleep(2) + + _connected = False + logger.info("Polling stopped") + + +def _parse_admin_allowed_chats(chat_id_config, allowed_config): + ids = set() + single = str(chat_id_config or "").strip() + if single: + ids.add(single) + for part in str(allowed_config or "").split(","): + part = part.strip() + if part: + ids.add(part) + return ids + + +def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): + global _running, _bot_token, _api_base, _poll_timeout, _offset, _connected + global _active_chat_id, _admin_allowed_chats, _auto_bound_chat + + proxy = auth.get_proxy_url() + if proxy: + _bot_token = "proxy" + _api_base = f"{proxy}/telegram" + else: + _bot_token = os.environ.get("TG_BOT_TOKEN", "").strip() + if not _bot_token: + raise ValueError("TG_BOT_TOKEN is required") + _api_base = f"https://api.telegram.org/bot{_bot_token}" + + _admin_allowed_chats = _parse_admin_allowed_chats(chat_id, allowed_chat_ids) + _auto_bound_chat = "" + + with _msg_lock: + _inbox.clear() + with _state_lock: + _active_chat_id = "" + + try: + _poll_timeout = max(1, int(poll_timeout)) + except Exception as e: + logger.warning(f"Invalid poll_timeout {poll_timeout!r}, falling back to 20: {e}") + _poll_timeout = 20 + + _offset = None + _running = True + _connected = False + if _admin_allowed_chats: + logger.info(f"Starting adapter, admin-restricted to chats: {sorted(_admin_allowed_chats)}") + else: + logger.info("Starting adapter with no admin chat restriction") + _initialize_offset() + + t = threading.Thread(target=_poll_loop, daemon=True) + t.start() + return t + + +def stop_telegram(): + global _running + _running = False + + +class TelegramChannel(channels.CommChannel): + + def __init__(self): + super().__init__() + + def start(self) -> None: + chat_id = config_get_by_key("TG_CHAT_ID", "") + allowed_chat_ids = config_get_by_key("TG_ALLOWED_CHAT_IDS", "") + poll_timeout = int(config_get_by_key("TG_POLL_TIMEOUT", 20)) + start_telegram(chat_id, allowed_chat_ids, poll_timeout) + + def stop(self) -> None: + stop_telegram() + + def receive(self) -> str: + return getLastMessage() + + def send(self, message: str) -> None: + send_message(message) + + +def loadOmegaClawPlugin(): channels.registerCommChannel("telegram", TelegramChannel()) diff --git a/config/config.yaml b/config/config.yaml index 1034f7a4..091f568f 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -73,8 +73,11 @@ SL_CHANNEL_ID: "" SL_POLL_INTERVAL: 60 # Telegram -# Telegram chat ID. If empty, OmegaClaw auto-binds after first valid inbound auth/message. +# Legacy single-chat allowlist. When set, only this chat is accepted. TG_CHAT_ID: "" +# Optional additional comma-separated chat IDs. These are a hard +# administrator boundary, including when user authentication is disabled. +TG_ALLOWED_CHAT_IDS: "" # Telegram polling timeout in seconds. TG_POLL_TIMEOUT: 20 diff --git a/tests/test_channel_auth_gating.py b/tests/test_channel_auth_gating.py index 4f5ab063..c6ab9521 100644 --- a/tests/test_channel_auth_gating.py +++ b/tests/test_channel_auth_gating.py @@ -24,6 +24,8 @@ def test_unbound_plain_message_is_not_used_as_auth_token(monkeypatch, module_nam auth.is_auth_enabled = lambda: True auth.get_channel_saved_user_id = lambda *args: False auth.get_channel_saved_group_id = lambda *args: False + auth.get_channel_authenticated_user_id = lambda *args: None + auth.authorize_channel_group = lambda *args: "ignore" calls = [] def authenticate_channel_user(*args): diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py index 98e368a0..896943b2 100644 --- a/tests/test_telegram_multichat.py +++ b/tests/test_telegram_multichat.py @@ -9,27 +9,36 @@ def load_telegram(monkeypatch, auth_enabled=True): - saved = set() + state = {"owner": None, "groups": set()} auth = types.ModuleType("auth") auth.is_auth_enabled = lambda: auth_enabled auth.get_proxy_url = lambda: "" - auth.get_channel_saved_group_id = lambda channel, group: (channel, str(group)) in saved - auth.store_channel_authenticated_group_id = lambda channel, group: saved.add((channel, str(group))) or True + auth.get_channel_authenticated_user_id = lambda channel: state["owner"] + auth.get_channel_saved_group_id = lambda channel, group: ( + channel, str(group) + ) in state["groups"] - def authenticate_channel_group(channel, group, candidate=None): - if candidate == "secret": - auth.store_channel_authenticated_group_id(channel, group) - return "auth_bound" - return "allow" if auth.get_channel_saved_group_id(channel, group) else "ignore" + def authenticate_channel_user(channel, user, candidate=None): + if candidate != "secret" or state["owner"] is not None: + return "ignore" + state["owner"] = str(user) + return "auth_bound" - auth.authenticate_channel_group = authenticate_channel_group + def authorize_channel_group(channel, group, requester): + if str(requester) != state["owner"]: + return "ignore" + state["groups"].add((channel, str(group))) + return "group_bound" + + auth.authenticate_channel_user = authenticate_channel_user + auth.authorize_channel_group = authorize_channel_group monkeypatch.setitem(sys.modules, "auth", auth) config = types.ModuleType("config") config.config_get_by_key = lambda _key, default=None: default monkeypatch.setitem(sys.modules, "config", config) channels = types.ModuleType("channels") - channels.CommChannel = type("CommChannel", (), {}) + channels.CommChannel = type("CommChannel", (), {"__init__": lambda self: None}) channels.registerCommChannel = lambda *_args: None monkeypatch.setitem(sys.modules, "channels", channels) monkeypatch.syspath_prepend(str(REPO_ROOT)) @@ -42,7 +51,7 @@ def authenticate_channel_group(channel, group, candidate=None): return module -def test_reply_uses_chat_that_supplied_the_next_message(monkeypatch): +def test_reply_uses_the_chat_that_supplied_the_message(monkeypatch): telegram = load_telegram(monkeypatch, auth_enabled=False) telegram._connected = True sent = [] @@ -58,10 +67,27 @@ def test_reply_uses_chat_that_supplied_the_next_message(monkeypatch): assert [params["chat_id"] for _, params in sent] == ["dm", "group"] -def test_group_authentication_allows_every_member_of_that_group(monkeypatch): +def test_owner_binds_group_without_exposing_secret(monkeypatch): telegram = load_telegram(monkeypatch) - assert telegram._is_allowed_message("group", "2", "hello") == "ignore" - assert telegram._is_allowed_message("group", "2", "auth secret") == "auth_bound" - assert telegram._is_allowed_message("group", "3", "hello") == "allow" - assert telegram._is_allowed_message("other-group", "3", "hello") == "ignore" + # A secret sent in a group cannot establish an owner. + assert telegram._is_allowed_message("group", "1", "group", "auth secret") == "ignore" + # The owner authenticates once in a private DM. + assert telegram._is_allowed_message("dm", "1", "private", "auth secret") == "auth_bound" + assert telegram._is_allowed_message("dm", "2", "private", "hello") == "ignore" + # Only that owner can open the group; then every group member is allowed. + assert telegram._is_allowed_message("group", "2", "group", "/bind") == "ignore" + assert telegram._is_allowed_message("group", "1", "group", "/bind@ExampleBot") == "group_bound" + assert telegram._is_allowed_message("group", "2", "group", "hello") == "allow" + assert telegram._is_allowed_message("other-group", "2", "group", "hello") == "ignore" + + +def test_configured_chats_are_a_hard_boundary_when_auth_is_disabled(monkeypatch): + telegram = load_telegram(monkeypatch, auth_enabled=False) + telegram._admin_allowed_chats = telegram._parse_admin_allowed_chats( + "legacy-chat", "group-a, group-b" + ) + + assert telegram._is_allowed_message("legacy-chat", "1", "private", "hello") == "allow" + assert telegram._is_allowed_message("group-a", "2", "group", "hello") == "allow" + assert telegram._is_allowed_message("unlisted", "3", "group", "hello") == "ignore" From a1f1cd2eea1cbe0b69abe0e823631d103f73a211 Mon Sep 17 00:00:00 2001 From: menilik eshetu Date: Thu, 20 Aug 2026 13:44:36 +0300 Subject: [PATCH 03/11] fix(auth): preserve persisted channel owner --- channels/auth.py | 60 +++++++++++++------------------- channels/telegram.py | 59 ++++++++++++++++++++----------- config/config.yaml | 3 +- tests/test_auth_standalone.py | 17 +++++++++ tests/test_telegram_multichat.py | 9 +++++ 5 files changed, 91 insertions(+), 57 deletions(-) diff --git a/channels/auth.py b/channels/auth.py index c8ac76be..d8c4a03e 100644 --- a/channels/auth.py +++ b/channels/auth.py @@ -124,49 +124,37 @@ def get_channel_saved_user_id(channel_identifier, user_id): logger.warning(f"[{channel_identifier}] Warning: a user was already validated, ignoring") return False - channel_identifier = str(channel_identifier or "").strip() - user_id = str(user_id or "").strip() - if not user_id: - return False - try: - path = _channel_auth_user_path() - with open(path, "r", encoding="utf-8") as f: - for line in f: - try: - record = json.loads(line) - saved_channel_identifier = str(record.get("channel_identifier", "")).strip() - saved_user_id = str(record.get("user_id", "")).strip() - except (AttributeError, json.JSONDecodeError) as e: - logger.warning(f"Skipping malformed channel authenticated user record: {e}") - continue - if saved_channel_identifier == channel_identifier and saved_user_id == user_id: - _user_ID_processed = True - return True - except FileNotFoundError: + # The first persisted record is the owner. Do not scan later records + # looking for another matching user: older installations may contain + # accidental duplicate records, but they must not create extra owners. + saved_user_id = get_channel_authenticated_user_id(channel_identifier) + if saved_user_id != str(user_id or "").strip(): return False - except Exception as e: - raise RuntimeError("Failed to read channel authenticated user records") from e - return False + + _user_ID_processed = True + return True def authenticate_channel_user(channel_identifier, user_id, auth_candidate=None): - # A token is accepted only when it came from an explicit auth command. - # Otherwise see if there was a prior session with the user-id and channel. + channel_identifier = str(channel_identifier or "").strip() + user_id = str(user_id or "").strip() + + # A persisted owner always wins over a reusable secret. This prevents a + # restart from allowing someone else who knows the secret to replace the + # original owner. + saved_user_id = get_channel_authenticated_user_id(channel_identifier) + if saved_user_id is not None: + return "allow" if saved_user_id == user_id else "ignore" + + # The secret can establish an owner only before an owner has been saved. if auth_candidate is not None and verify_token(auth_candidate): + label = channel_identifier.upper() if store_channel_authenticated_user_id(channel_identifier, user_id): - label = str(channel_identifier).upper() logger.info(f"[{label}] Saved authenticated user ID") return "auth_bound" - else: - label = str(channel_identifier).upper() - logger.error(f"[{label}] ERROR -- Unable to save user ID") - return "ignore" - elif get_channel_saved_user_id(channel_identifier, user_id): - label = str(channel_identifier).upper() - logger.info(f"[{label}] Verified previously validated user ID") - return "allow" - else: - return "ignore" + logger.error(f"[{label}] ERROR -- Unable to save user ID") + + return "ignore" def get_channel_authenticated_user_id(channel_identifier): @@ -295,4 +283,4 @@ def authorize_channel_group(channel_identifier, group_id, requester_user_id): return "group_bound" logger.error(f"[{str(channel_identifier).upper()}] ERROR -- Unable to save group ID") - return "ignore" \ No newline at end of file + return "ignore" diff --git a/channels/telegram.py b/channels/telegram.py index b16c15cb..31f85470 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -25,7 +25,8 @@ _connected = False # --- Admin allowlist (review point #8) -------------------------------- -# Purely a chat-level gate, checked before any owner/group logic runs. +# A chat-level administrator boundary. The initial owner-authentication DM +# and the persisted owner's DM are intentionally exempt when auth is enabled. # TG_CHAT_ID is kept for backwards compatibility (single chat); # TG_ALLOWED_CHAT_IDS is new and accepts a comma-separated list. Empty # means "no admin restriction configured". @@ -95,8 +96,9 @@ def send_message(text, target_chat=None): # Authorization. # # Layered, outer to inner: -# 1. Admin allowlist (TG_CHAT_ID / TG_ALLOWED_CHAT_IDS) -- chats outside -# it are always ignored, auth or no auth. +# 1. Admin allowlist (TG_CHAT_ID / TG_ALLOWED_CHAT_IDS) -- chats outside +# it are ignored, except for private owner bootstrap/owner DMs when +# authentication is enabled. # 2. If auth is disabled: allowlisted chats are trusted outright; with no # allowlist at all, fall back to the legacy single-chat auto-bind. # 3. If auth is enabled: nothing is allowed until an owner exists. The @@ -142,29 +144,46 @@ def _is_bind_command(msg): return token in _BIND_COMMANDS -def _is_allowed_message(chat_id, user_id, chat_type, msg): - global _auto_bound_chat - - if _admin_allowed_chats and chat_id not in _admin_allowed_chats: - return "ignore" - - if not auth.is_auth_enabled(): - if _admin_allowed_chats: - return "allow" - with _state_lock: - if _auto_bound_chat and chat_id != _auto_bound_chat: - return "ignore" +def _is_allowed_message(chat_id, user_id, chat_type, msg): + global _auto_bound_chat + + auth_enabled = auth.is_auth_enabled() + + if not auth_enabled: + if _admin_allowed_chats: + return "allow" if chat_id in _admin_allowed_chats else "ignore" + with _state_lock: + if _auto_bound_chat and chat_id != _auto_bound_chat: + return "ignore" if not _auto_bound_chat: _auto_bound_chat = chat_id return "allow" - + owner_id = auth.get_channel_authenticated_user_id("TELEGRAM") + is_owner_bootstrap = ( + owner_id is None + and chat_type == "private" + and _is_auth_command(msg) + ) + is_owner_private_chat = ( + owner_id is not None + and chat_type == "private" + and user_id == owner_id + ) + + if ( + _admin_allowed_chats + and chat_id not in _admin_allowed_chats + and not is_owner_bootstrap + and not is_owner_private_chat + ): + return "ignore" + if owner_id is None: - # The reusable secret must never be exposed in a group. Establish - # the Telegram owner from a direct message only; that owner can then - # open groups using /bind, which relies on their Telegram user id. - if chat_type == "private" and _is_auth_command(msg): + # The reusable secret is accepted only in a private DM. The owner can + # then open groups using /bind, which relies on Telegram user id. + if is_owner_bootstrap: candidate = _parse_auth_candidate(msg) return auth.authenticate_channel_user("TELEGRAM", user_id, candidate) return "ignore" diff --git a/config/config.yaml b/config/config.yaml index 091f568f..ebda71f7 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -76,7 +76,8 @@ SL_POLL_INTERVAL: 60 # Legacy single-chat allowlist. When set, only this chat is accepted. TG_CHAT_ID: "" # Optional additional comma-separated chat IDs. These are a hard -# administrator boundary, including when user authentication is disabled. +# administrator boundary. With authentication enabled, the owner can still +# bootstrap and use their private DM. TG_ALLOWED_CHAT_IDS: "" # Telegram polling timeout in seconds. TG_POLL_TIMEOUT: 20 diff --git a/tests/test_auth_standalone.py b/tests/test_auth_standalone.py index 1cc00ffa..6a90461a 100644 --- a/tests/test_auth_standalone.py +++ b/tests/test_auth_standalone.py @@ -87,6 +87,23 @@ def test_saved_owner_blocks_auth_secret_reuse_after_restart(monkeypatch, tmp_pat assert owner_process.authenticate_channel_user("IRC", "owner") == "allow" +def test_saved_owner_cannot_be_replaced_by_a_reused_secret(monkeypatch, tmp_path): + monkeypatch.setenv("OMEGACLAW_AUTH_SECRET", "one-time-secret") + + first_process = load_auth_module(monkeypatch) + monkeypatch.setattr(first_process, "_MEMORY_DIRECTORY", str(tmp_path)) + assert first_process.authenticate_channel_user( + "TELEGRAM", "owner", "one-time-secret" + ) == "auth_bound" + + restarted_process = load_auth_module(monkeypatch) + monkeypatch.setattr(restarted_process, "_MEMORY_DIRECTORY", str(tmp_path)) + assert restarted_process.authenticate_channel_user( + "TELEGRAM", "attacker", "one-time-secret" + ) == "ignore" + assert restarted_process.get_channel_authenticated_user_id("TELEGRAM") == "owner" + + def test_plain_message_does_not_verify_a_token(monkeypatch): auth = load_auth_module(monkeypatch) monkeypatch.setattr( diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py index 896943b2..11b1371f 100644 --- a/tests/test_telegram_multichat.py +++ b/tests/test_telegram_multichat.py @@ -91,3 +91,12 @@ def test_configured_chats_are_a_hard_boundary_when_auth_is_disabled(monkeypatch) assert telegram._is_allowed_message("legacy-chat", "1", "private", "hello") == "allow" assert telegram._is_allowed_message("group-a", "2", "group", "hello") == "allow" assert telegram._is_allowed_message("unlisted", "3", "group", "hello") == "ignore" + + +def test_allowlist_still_permits_owner_dm_bootstrap(monkeypatch): + telegram = load_telegram(monkeypatch) + telegram._admin_allowed_chats = {"approved-group"} + + assert telegram._is_allowed_message("owner-dm", "1", "private", "auth secret") == "auth_bound" + assert telegram._is_allowed_message("owner-dm", "1", "private", "hello") == "allow" + assert telegram._is_allowed_message("unlisted-group", "2", "group", "hello") == "ignore" From dee5ef9ea3345b9d67155b700c4f5d6137b031a3 Mon Sep 17 00:00:00 2001 From: menilik eshetu Date: Thu, 20 Aug 2026 15:22:20 +0300 Subject: [PATCH 04/11] chore: simplify Telegram auth comments --- channels/auth.py | 37 ++----------- channels/telegram.py | 122 +++++++++++++++---------------------------- 2 files changed, 44 insertions(+), 115 deletions(-) diff --git a/channels/auth.py b/channels/auth.py index d8c4a03e..debb93b3 100644 --- a/channels/auth.py +++ b/channels/auth.py @@ -76,15 +76,6 @@ def _channel_auth_user_path(): # --------------------------------------------------------------------------- # Single-user (owner) authentication. -# -# UNCHANGED from the original implementation, byte for byte. IRC, Slack and -# Mattermost depend on this exact behavior (including the single-use -# _user_ID_processed guard). Telegram now ALSO calls into this same code -# path (see authenticate_channel_user usage in channels/telegram.py) rather -# than duplicating it -- this is the fix for review point #2: Telegram no -# longer has a parallel "chat-based" identity system, it uses the one owner -# identity every other channel uses. -# --------------------------------------------------------------------------- def store_channel_authenticated_user_id(channel_identifier, user_id): # For any single run of OmegaClaw, allow only a single save of a user-id or verification @@ -125,8 +116,6 @@ def get_channel_saved_user_id(channel_identifier, user_id): return False # The first persisted record is the owner. Do not scan later records - # looking for another matching user: older installations may contain - # accidental duplicate records, but they must not create extra owners. saved_user_id = get_channel_authenticated_user_id(channel_identifier) if saved_user_id != str(user_id or "").strip(): return False @@ -139,9 +128,7 @@ def authenticate_channel_user(channel_identifier, user_id, auth_candidate=None): channel_identifier = str(channel_identifier or "").strip() user_id = str(user_id or "").strip() - # A persisted owner always wins over a reusable secret. This prevents a - # restart from allowing someone else who knows the secret to replace the - # original owner. + # A persisted owner always wins over a reusable secret. saved_user_id = get_channel_authenticated_user_id(channel_identifier) if saved_user_id is not None: return "allow" if saved_user_id == user_id else "ignore" @@ -161,14 +148,6 @@ def get_channel_authenticated_user_id(channel_identifier): """ Read-only owner lookup. Returns the persisted owner user_id for a channel, or None if no owner has authenticated yet. - - This is intentionally separate from get_channel_saved_user_id() above: - that function is single-use per process (it flips _user_ID_processed - and refuses to check again), which is fine for its original purpose - but wrong for Telegram's /bind flow, which needs to ask "who is the - owner?" repeatedly for the life of the process without ever mutating - state or tripping that guard. Never writes, never touches - _user_ID_processed. """ channel_identifier = str(channel_identifier or "").strip() if not channel_identifier: @@ -194,15 +173,7 @@ def get_channel_authenticated_user_id(channel_identifier): # --------------------------------------------------------------------------- -# Telegram-only group authorization (NEW). -# -# Fixes review point #3: the shared secret is NEVER sent or checked inside -# a group. It is only ever used once, to establish the DM owner (via -# authenticate_channel_user above). Opening a group is purely an identity -# check -- does the /bind sender's user_id match the persisted owner? -- -# never a credential check. Stored in its own file so this can never read, -# write, or otherwise influence authenticated-user.json. -# --------------------------------------------------------------------------- +# Telegram-only group authorization. def _channel_auth_group_path(): return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_GROUP_FILE) @@ -263,9 +234,7 @@ def get_channel_saved_group_id(channel_identifier, group_id): def authorize_channel_group(channel_identifier, group_id, requester_user_id): """ Open a group chat to all its members -- but ONLY when requester_user_id - matches the persisted owner for this channel (see - get_channel_authenticated_user_id). The shared secret plays no role - here at all; this is a pure identity check on the /bind sender. + matches the persisted owner for this channel """ if get_channel_saved_group_id(channel_identifier, group_id): return "allow" diff --git a/channels/telegram.py b/channels/telegram.py index 31f85470..6c4a774b 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -23,31 +23,13 @@ _poll_timeout = 20 _offset = None _connected = False - -# --- Admin allowlist (review point #8) -------------------------------- -# A chat-level administrator boundary. The initial owner-authentication DM -# and the persisted owner's DM are intentionally exempt when auth is enabled. -# TG_CHAT_ID is kept for backwards compatibility (single chat); -# TG_ALLOWED_CHAT_IDS is new and accepts a comma-separated list. Empty -# means "no admin restriction configured". _admin_allowed_chats = set() -# Legacy no-auth fallback: first chat to talk wins. Only used when auth is -# disabled AND no admin allowlist is configured, preserving the original -# single-chat auto-bind behavior for existing no-auth deployments. _auto_bound_chat = "" _BIND_COMMANDS = ("/bind", "/authorize_group") -# --------------------------------------------------------------------------- -# Multi-chat routing (review point #9). -# -# Pure plumbing: remembers which chat each inbound message came from, and -# lets replies target that same chat. Has no knowledge of authorization -- -# it only ever queues/sends what _is_allowed_message() has already approved. -# --------------------------------------------------------------------------- - def _enqueue_message(msg, chat_id): with _msg_lock: _inbox.append((str(chat_id), str(msg))) @@ -94,26 +76,6 @@ def send_message(text, target_chat=None): # --------------------------------------------------------------------------- # Authorization. -# -# Layered, outer to inner: -# 1. Admin allowlist (TG_CHAT_ID / TG_ALLOWED_CHAT_IDS) -- chats outside -# it are ignored, except for private owner bootstrap/owner DMs when -# authentication is enabled. -# 2. If auth is disabled: allowlisted chats are trusted outright; with no -# allowlist at all, fall back to the legacy single-chat auto-bind. -# 3. If auth is enabled: nothing is allowed until an owner exists. The -# owner is established exactly once via "auth " -- reusing -# auth.authenticate_channel_user(), the SAME function IRC/Slack/ -# Mattermost use, keyed as "TELEGRAM". This is a deliberate reuse, not -# a parallel system (review point #2). -# 4. Once an owner exists: -# - Private chats (DMs): owner only, forever. No other user can ever -# be allowed in the owner's DM (review point #6). -# - Group chats: open to every member once authorized. Before that, -# only the owner's own "/bind" (or "/authorize_group") message -# opens it -- verified by sender user_id, never by the secret -# (review point #3, #4, #5). -# --------------------------------------------------------------------------- def _parse_auth_candidate(msg): text = msg.strip() @@ -138,55 +100,53 @@ def _first_token(msg): def _is_bind_command(msg): - # Handle Telegram's "/bind@YourBotName" form, sent automatically by - # clients when a group has more than one bot in it. + # Handle Telegram's "/bind@YourBotName" form token = _first_token(msg).split("@", 1)[0] return token in _BIND_COMMANDS -def _is_allowed_message(chat_id, user_id, chat_type, msg): - global _auto_bound_chat - - auth_enabled = auth.is_auth_enabled() - - if not auth_enabled: - if _admin_allowed_chats: - return "allow" if chat_id in _admin_allowed_chats else "ignore" - with _state_lock: - if _auto_bound_chat and chat_id != _auto_bound_chat: - return "ignore" +def _is_allowed_message(chat_id, user_id, chat_type, msg): + global _auto_bound_chat + + auth_enabled = auth.is_auth_enabled() + + if not auth_enabled: + if _admin_allowed_chats: + return "allow" if chat_id in _admin_allowed_chats else "ignore" + with _state_lock: + if _auto_bound_chat and chat_id != _auto_bound_chat: + return "ignore" if not _auto_bound_chat: _auto_bound_chat = chat_id return "allow" - - owner_id = auth.get_channel_authenticated_user_id("TELEGRAM") - - is_owner_bootstrap = ( - owner_id is None - and chat_type == "private" - and _is_auth_command(msg) - ) - is_owner_private_chat = ( - owner_id is not None - and chat_type == "private" - and user_id == owner_id - ) - - if ( - _admin_allowed_chats - and chat_id not in _admin_allowed_chats - and not is_owner_bootstrap - and not is_owner_private_chat - ): - return "ignore" - - if owner_id is None: - # The reusable secret is accepted only in a private DM. The owner can - # then open groups using /bind, which relies on Telegram user id. - if is_owner_bootstrap: - candidate = _parse_auth_candidate(msg) - return auth.authenticate_channel_user("TELEGRAM", user_id, candidate) - return "ignore" + + owner_id = auth.get_channel_authenticated_user_id("TELEGRAM") + + is_owner_bootstrap = ( + owner_id is None + and chat_type == "private" + and _is_auth_command(msg) + ) + is_owner_private_chat = ( + owner_id is not None + and chat_type == "private" + and user_id == owner_id + ) + + if ( + _admin_allowed_chats + and chat_id not in _admin_allowed_chats + and not is_owner_bootstrap + and not is_owner_private_chat + ): + return "ignore" + + if owner_id is None: + # The reusable secret is accepted only in a private DM. The owner can + if is_owner_bootstrap: + candidate = _parse_auth_candidate(msg) + return auth.authenticate_channel_user("TELEGRAM", user_id, candidate) + return "ignore" if chat_type == "private": return "allow" if user_id == owner_id else "ignore" @@ -405,4 +365,4 @@ def send(self, message: str) -> None: def loadOmegaClawPlugin(): - channels.registerCommChannel("telegram", TelegramChannel()) + channels.registerCommChannel("telegram", TelegramChannel()) From 8865b8bdc8f1678c416568aec8a058eac9b17449 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Thu, 20 Aug 2026 18:56:39 +0300 Subject: [PATCH 05/11] Fix: enabled strict bot username check for /bind command and added unit tests --- channels/telegram.py | 27 +++++++++++++++++++++++++-- tests/test_channel_auth_gating.py | 2 +- tests/test_telegram_multichat.py | 17 +++++++++++++++++ 3 files changed, 43 insertions(+), 3 deletions(-) diff --git a/channels/telegram.py b/channels/telegram.py index 6c4a774b..33eefaf8 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -20,6 +20,7 @@ _bot_token = "" _api_base = "" +_bot_username= "" _poll_timeout = 20 _offset = None _connected = False @@ -101,8 +102,16 @@ def _first_token(msg): def _is_bind_command(msg): # Handle Telegram's "/bind@YourBotName" form - token = _first_token(msg).split("@", 1)[0] - return token in _BIND_COMMANDS + token = _first_token(msg) + command, separator, target_username = token.partition("@") + + if command not in _BIND_COMMANDS: + return False + + if not separator: + return True + + return bool(_bot_username) and target_username == _bot_username def _is_allowed_message(chat_id, user_id, chat_type, msg): @@ -202,6 +211,18 @@ def _api_call(method, params=None, timeout=30, use_post=False): return payload.get("result") +def _initialize_bot_identity(): + global _bot_username + + result = _api_call("getMe", timeout=10) + if not isinstance(result, dict): + raise RuntimeError("Telegram getMe returned an invalid response") + + username = str(result.get("username", "")).strip().lstrip("@").lower() + if not username: + raise RuntimeError("Telegram getMe did not return the bot username") + + _bot_username = username def _initialize_offset(): global _offset @@ -331,6 +352,8 @@ def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): logger.info(f"Starting adapter, admin-restricted to chats: {sorted(_admin_allowed_chats)}") else: logger.info("Starting adapter with no admin chat restriction") + + _initialize_bot_identity() _initialize_offset() t = threading.Thread(target=_poll_loop, daemon=True) diff --git a/tests/test_channel_auth_gating.py b/tests/test_channel_auth_gating.py index c6ab9521..1e1788af 100644 --- a/tests/test_channel_auth_gating.py +++ b/tests/test_channel_auth_gating.py @@ -14,7 +14,7 @@ ("module_name", "arguments"), [ ("irc", ("alice", "hello 🌍")), - ("telegram", ("chat", "alice", "hello 🌍")), + ("telegram", ("chat", "alice", "private", "hello 🌍")), ("slack", ("channel", "alice", "hello 🌍")), ("mattermost", ("alice", "hello 🌍")), ], diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py index 11b1371f..fdc40f6f 100644 --- a/tests/test_telegram_multichat.py +++ b/tests/test_telegram_multichat.py @@ -69,6 +69,7 @@ def test_reply_uses_the_chat_that_supplied_the_message(monkeypatch): def test_owner_binds_group_without_exposing_secret(monkeypatch): telegram = load_telegram(monkeypatch) + telegram._bot_username = "examplebot" # A secret sent in a group cannot establish an owner. assert telegram._is_allowed_message("group", "1", "group", "auth secret") == "ignore" @@ -82,6 +83,22 @@ def test_owner_binds_group_without_exposing_secret(monkeypatch): assert telegram._is_allowed_message("other-group", "2", "group", "hello") == "ignore" +def test_bind_command_targets_this_bot_only(monkeypatch): + telegram = load_telegram(monkeypatch) + telegram._bot_username = "examplebot" + + assert telegram._is_bind_command("/bind") + assert telegram._is_bind_command("/authorize_group") + + assert telegram._is_bind_command("/bind@ExampleBot") + assert telegram._is_bind_command("/BIND@examplebot") + assert telegram._is_bind_command("/authorize_group@ExampleBot") + + assert not telegram._is_bind_command("/bind@AnotherBot") + assert not telegram._is_bind_command("/authorize_group@AnotherBot") + assert not telegram._is_bind_command("/binder@ExampleBot") + + def test_configured_chats_are_a_hard_boundary_when_auth_is_disabled(monkeypatch): telegram = load_telegram(monkeypatch, auth_enabled=False) telegram._admin_allowed_chats = telegram._parse_admin_allowed_chats( From 936ff5dec22e8f95851adc5fae4c7d8b6c915709 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Tue, 25 Aug 2026 08:41:17 +0300 Subject: [PATCH 06/11] Fix: enabled permission revoke, and message lock to prevent cross-chat bleeding effect. --- channels/auth.py | 40 ++++++++- channels/telegram.py | 136 ++++++++++++++++++++++++------- tests/test_auth_standalone.py | 18 ++++ tests/test_telegram_multichat.py | 86 +++++++++++++++++++ 4 files changed, 249 insertions(+), 31 deletions(-) diff --git a/channels/auth.py b/channels/auth.py index debb93b3..2eb5bc75 100644 --- a/channels/auth.py +++ b/channels/auth.py @@ -212,6 +212,7 @@ def get_channel_saved_group_id(channel_identifier, group_id): group_id = str(group_id or "").strip() if not channel_identifier or not group_id: return False + authorized = False try: with open(_channel_auth_group_path(), "r", encoding="utf-8") as f: for line in f: @@ -223,12 +224,12 @@ def get_channel_saved_group_id(channel_identifier, group_id): logger.warning(f"Skipping malformed channel authenticated group record: {e}") continue if saved_channel == channel_identifier and saved_group == group_id: - return True + authorized = not bool(record.get("revoked", False)) except FileNotFoundError: return False except Exception as e: raise RuntimeError("Failed to read channel authenticated group records") from e - return False + return authorized def authorize_channel_group(channel_identifier, group_id, requester_user_id): @@ -253,3 +254,38 @@ def authorize_channel_group(channel_identifier, group_id, requester_user_id): logger.error(f"[{str(channel_identifier).upper()}] ERROR -- Unable to save group ID") return "ignore" + + +def revoke_channel_group(channel_identifier, group_id, requester_user_id): + """Remove a trusted group when requested by the persisted channel owner.""" + channel_identifier = str(channel_identifier or "").strip() + group_id = str(group_id or "").strip() + requester_user_id = str(requester_user_id or "").strip() + if not channel_identifier or not group_id: + return "ignore" + + owner_id = get_channel_authenticated_user_id(channel_identifier) + if owner_id is None or requester_user_id != owner_id: + return "ignore" + + if not get_channel_saved_group_id(channel_identifier, group_id): + return "ignore" + + payload = { + "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "channel_identifier": channel_identifier, + "group_id": group_id, + "authorized_by": owner_id, + "revoked": True, + } + path = _channel_auth_group_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + try: + with open(path, "a", encoding="utf-8") as f: + json.dump(payload, f, separators=(",", ":")) + f.write("\n") + except OSError as e: + raise RuntimeError("Failed to write channel group revocation record") from e + + logger.info(f"[{channel_identifier.upper()}] Removed authorized group ID {group_id}") + return "group_unbound" diff --git a/channels/telegram.py b/channels/telegram.py index 33eefaf8..3d2be696 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -7,6 +7,7 @@ from collections import deque import auth from src.logger import get_logger +from delivery_queue import PendingMessages import channels from config import config_get_by_key @@ -17,6 +18,11 @@ _state_lock = threading.Lock() _inbox = deque() _active_chat_id = "" +_active_message_token = None +_active_replied = False +_next_message_token = 0 +_default_chat_id = "" +_outbox = PendingMessages() _bot_token = "" _api_base = "" @@ -29,50 +35,88 @@ _auto_bound_chat = "" _BIND_COMMANDS = ("/bind", "/authorize_group") +_UNBIND_COMMANDS = ("/unbind", "/unauthorize_group") def _enqueue_message(msg, chat_id): + global _next_message_token with _msg_lock: - _inbox.append((str(chat_id), str(msg))) + _next_message_token += 1 + _inbox.append((_next_message_token, str(chat_id), str(msg))) def getLastMessage(): - global _active_chat_id + global _active_chat_id, _active_message_token, _active_replied with _msg_lock: + if _active_chat_id and not _active_replied: + return "" + if _active_replied: + _active_chat_id = "" + _active_message_token = None + _active_replied = False if not _inbox: return "" - chat_id, message = _inbox.popleft() - with _state_lock: + message_token, chat_id, message = _inbox.popleft() _active_chat_id = chat_id + _active_message_token = message_token return message +def _ready_to_send(): + return _connected + + +def _deliver_outbound(item): + global _active_replied + target_chat, chunk, completed_message_token = item + _api_call( + "sendMessage", + {"chat_id": target_chat, "text": chunk}, + timeout=15, + use_post=True, + ) + if completed_message_token is not None: + with _msg_lock: + if completed_message_token == _active_message_token: + _active_replied = True + + +def _flush_outbox(): + try: + _outbox.flush(_deliver_outbound, _ready_to_send) + except Exception as exc: + logger.warning(f"Telegram send failed; retaining queued message: {exc}") + + def send_message(text, target_chat=None): text = str(text).replace("\\n", "\n").replace("\r", "") if not text: - return - - with _state_lock: - target_chat = str(target_chat or _active_chat_id).strip() + return False - if not _connected or not target_chat: - return + explicit_target = str(target_chat or "").strip() + with _msg_lock: + active_chat = _active_chat_id + active_message_token = _active_message_token + target_chat = explicit_target or active_chat or _default_chat_id + completed_message_token = ( + active_message_token + if not explicit_target and active_chat and target_chat == active_chat + else None + ) + + if not target_chat: + logger.warning("Telegram send skipped: no active or default chat is available") + return False max_len = 3900 - for i in range(0, len(text), max_len): - chunk = text[i:i + max_len] - if not chunk: - continue - try: - _api_call( - "sendMessage", - {"chat_id": target_chat, "text": chunk}, - timeout=15, - use_post=True, - ) - except Exception as exc: - logger.exception(f"Send failed: {exc}") - return + chunks = [text[i:i + max_len] for i in range(0, len(text), max_len)] + outbound = [] + for index, chunk in enumerate(chunks): + completes_message = completed_message_token if index == len(chunks) - 1 else None + outbound.append((target_chat, chunk, completes_message)) + _outbox.extend(outbound) + _flush_outbox() + return True # --------------------------------------------------------------------------- @@ -100,12 +144,11 @@ def _first_token(msg): return stripped.split(None, 1)[0].lower() -def _is_bind_command(msg): - # Handle Telegram's "/bind@YourBotName" form +def _is_targeted_command(msg, commands): token = _first_token(msg) command, separator, target_username = token.partition("@") - if command not in _BIND_COMMANDS: + if command not in commands: return False if not separator: @@ -114,6 +157,19 @@ def _is_bind_command(msg): return bool(_bot_username) and target_username == _bot_username +def _is_bind_command(msg): + return _is_targeted_command(msg, _BIND_COMMANDS) + + +def _is_unbind_command(msg): + return _is_targeted_command(msg, _UNBIND_COMMANDS) + + +def _command_argument(msg): + parts = str(msg or "").strip().split(None, 1) + return parts[1].strip() if len(parts) == 2 else "" + + def _is_allowed_message(chat_id, user_id, chat_type, msg): global _auto_bound_chat @@ -158,9 +214,19 @@ def _is_allowed_message(chat_id, user_id, chat_type, msg): return "ignore" if chat_type == "private": + if user_id == owner_id and _is_unbind_command(msg): + group_id = _command_argument(msg) + if group_id: + return auth.revoke_channel_group("TELEGRAM", group_id, user_id) + return "ignore" return "allow" if user_id == owner_id else "ignore" # Anything that isn't "private" is a group/supergroup chat. + if _is_unbind_command(msg): + if user_id == owner_id: + return auth.revoke_channel_group("TELEGRAM", chat_id, user_id) + return "ignore" + if auth.get_channel_saved_group_id("TELEGRAM", chat_id): return "allow" @@ -256,6 +322,7 @@ def _poll_loop(): updates = _api_call("getUpdates", params=params, timeout=int(_poll_timeout) + 10) or [] _connected = True + _flush_outbox() for update in updates: update_id = update.get("update_id") @@ -296,6 +363,11 @@ def _poll_loop(): "This group is now authorized. All members can talk to the bot here.", chat_id, ) + elif state == "group_unbound": + send_message( + "This group is no longer authorized.", + chat_id, + ) except Exception as exc: _connected = False logger.warning(f"Poll error: {exc}") @@ -319,7 +391,9 @@ def _parse_admin_allowed_chats(chat_id_config, allowed_config): def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): global _running, _bot_token, _api_base, _poll_timeout, _offset, _connected - global _active_chat_id, _admin_allowed_chats, _auto_bound_chat + global _active_chat_id, _active_message_token, _active_replied + global _next_message_token, _default_chat_id + global _admin_allowed_chats, _auto_bound_chat proxy = auth.get_proxy_url() if proxy: @@ -332,12 +406,16 @@ def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): _api_base = f"https://api.telegram.org/bot{_bot_token}" _admin_allowed_chats = _parse_admin_allowed_chats(chat_id, allowed_chat_ids) + _default_chat_id = str(chat_id or "").strip() _auto_bound_chat = "" with _msg_lock: _inbox.clear() - with _state_lock: _active_chat_id = "" + _active_message_token = None + _active_replied = False + _next_message_token = 0 + _outbox.clear() try: _poll_timeout = max(1, int(poll_timeout)) diff --git a/tests/test_auth_standalone.py b/tests/test_auth_standalone.py index 6a90461a..5e0050e5 100644 --- a/tests/test_auth_standalone.py +++ b/tests/test_auth_standalone.py @@ -112,3 +112,21 @@ def test_plain_message_does_not_verify_a_token(monkeypatch): monkeypatch.setattr(auth, "get_channel_authenticated_user_id", lambda *args: None) assert auth.authenticate_channel_user("IRC", "alice") == "ignore" + + +def test_owner_can_revoke_an_authorized_group(monkeypatch, tmp_path): + auth = load_auth_module(monkeypatch) + monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path)) + + monkeypatch.setattr(auth, "get_channel_authenticated_user_id", lambda _channel: "owner") + assert auth.store_channel_authenticated_group_id("TELEGRAM", "group", "owner") is True + assert auth.get_channel_saved_group_id("TELEGRAM", "group") is True + + assert auth.revoke_channel_group("TELEGRAM", "group", "attacker") == "ignore" + assert auth.get_channel_saved_group_id("TELEGRAM", "group") is True + + assert auth.revoke_channel_group("TELEGRAM", "group", "owner") == "group_unbound" + assert auth.get_channel_saved_group_id("TELEGRAM", "group") is False + + assert auth.store_channel_authenticated_group_id("TELEGRAM", "group", "owner") is True + assert auth.get_channel_saved_group_id("TELEGRAM", "group") is True diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py index fdc40f6f..02265949 100644 --- a/tests/test_telegram_multichat.py +++ b/tests/test_telegram_multichat.py @@ -30,8 +30,16 @@ def authorize_channel_group(channel, group, requester): state["groups"].add((channel, str(group))) return "group_bound" + def revoke_channel_group(channel, group, requester): + key = (channel, str(group)) + if str(requester) != state["owner"] or key not in state["groups"]: + return "ignore" + state["groups"].remove(key) + return "group_unbound" + auth.authenticate_channel_user = authenticate_channel_user auth.authorize_channel_group = authorize_channel_group + auth.revoke_channel_group = revoke_channel_group monkeypatch.setitem(sys.modules, "auth", auth) config = types.ModuleType("config") @@ -60,6 +68,7 @@ def test_reply_uses_the_chat_that_supplied_the_message(monkeypatch): telegram._enqueue_message("dm message", "dm") telegram._enqueue_message("group message", "group") assert telegram.getLastMessage() == "dm message" + assert telegram.getLastMessage() == "" telegram.send_message("dm reply") assert telegram.getLastMessage() == "group message" telegram.send_message("group reply") @@ -67,6 +76,58 @@ def test_reply_uses_the_chat_that_supplied_the_message(monkeypatch): assert [params["chat_id"] for _, params in sent] == ["dm", "group"] +def test_identical_messages_from_different_chats_are_processed(monkeypatch): + telegram = load_telegram(monkeypatch, auth_enabled=False) + telegram._connected = True + sent = [] + telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) + + telegram._enqueue_message("same message", "dm") + telegram._enqueue_message("same message", "group") + + assert telegram.getLastMessage() == "same message" + telegram.send_message("dm reply") + assert telegram.getLastMessage() == "same message" + telegram.send_message("group reply") + + assert [params["chat_id"] for _, params in sent] == ["dm", "group"] + + +def test_failed_delivery_retains_the_original_chat(monkeypatch): + telegram = load_telegram(monkeypatch, auth_enabled=False) + telegram._connected = True + attempts = [] + + def flaky_api(method, params, **_kwargs): + attempts.append((method, params.copy())) + if len(attempts) == 1: + raise RuntimeError("temporary failure") + + telegram._api_call = flaky_api + telegram._enqueue_message("dm message", "dm") + assert telegram.getLastMessage() == "dm message" + + telegram.send_message("dm reply") + telegram._enqueue_message("group message", "group") + assert telegram.getLastMessage() == "" + + telegram._flush_outbox() + assert telegram.getLastMessage() == "group message" + assert [params["chat_id"] for _, params in attempts] == ["dm", "dm"] + + +def test_proactive_message_uses_only_the_configured_default_chat(monkeypatch): + telegram = load_telegram(monkeypatch, auth_enabled=False) + telegram._connected = True + telegram._default_chat_id = "configured-default" + sent = [] + telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) + + telegram.send_message("startup message") + + assert sent[0][1]["chat_id"] == "configured-default" + + def test_owner_binds_group_without_exposing_secret(monkeypatch): telegram = load_telegram(monkeypatch) telegram._bot_username = "examplebot" @@ -98,6 +159,31 @@ def test_bind_command_targets_this_bot_only(monkeypatch): assert not telegram._is_bind_command("/authorize_group@AnotherBot") assert not telegram._is_bind_command("/binder@ExampleBot") + assert telegram._is_unbind_command("/unbind") + assert telegram._is_unbind_command("/UNBIND@examplebot") + assert not telegram._is_unbind_command("/unbind@AnotherBot") + + +def test_only_owner_can_unbind_group(monkeypatch): + telegram = load_telegram(monkeypatch) + telegram._bot_username = "examplebot" + + assert telegram._is_allowed_message("dm", "1", "private", "auth secret") == "auth_bound" + assert telegram._is_allowed_message("group", "1", "group", "/bind") == "group_bound" + assert telegram._is_allowed_message("group", "2", "group", "/unbind") == "ignore" + assert telegram._is_allowed_message("group", "2", "group", "hello") == "allow" + assert telegram._is_allowed_message("group", "1", "group", "/unbind@ExampleBot") == "group_unbound" + assert telegram._is_allowed_message("group", "2", "group", "hello") == "ignore" + + +def test_owner_can_unbind_group_from_dm(monkeypatch): + telegram = load_telegram(monkeypatch) + + assert telegram._is_allowed_message("dm", "1", "private", "auth secret") == "auth_bound" + assert telegram._is_allowed_message("group", "1", "group", "/bind") == "group_bound" + assert telegram._is_allowed_message("dm", "1", "private", "/unbind group") == "group_unbound" + assert telegram._is_allowed_message("group", "2", "group", "hello") == "ignore" + def test_configured_chats_are_a_hard_boundary_when_auth_is_disabled(monkeypatch): telegram = load_telegram(monkeypatch, auth_enabled=False) From e4111c9262aee3aa87293393ed9c76f59867ef80 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Tue, 25 Aug 2026 09:25:03 +0300 Subject: [PATCH 07/11] Fix: harden multi-chat routing and authorization - serialize inbound Telegram chats to prevent reply misrouting - retain destination-aware outbound messages for retry - use TG_CHAT_ID for proactive messages - add owner-only /unbind support - validate and cache persisted authorization before polling - advance update offsets only after successful processing - tolerate Telegram getMe failures - document Telegram binding, routing, and retry behavior --- channels/auth.py | 50 +++++++++++ channels/telegram.py | 141 ++++++++++++++++--------------- docs/reference-channels.md | 10 ++- docs/reference-configuration.md | 3 +- tests/test_auth_standalone.py | 26 ++++++ tests/test_telegram_multichat.py | 91 ++++++++++++++++++++ 6 files changed, 248 insertions(+), 73 deletions(-) diff --git a/channels/auth.py b/channels/auth.py index 2eb5bc75..bdf538c7 100644 --- a/channels/auth.py +++ b/channels/auth.py @@ -179,6 +179,56 @@ def _channel_auth_group_path(): return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_GROUP_FILE) +def load_channel_auth_state(channel_identifier): + """Validate and load one channel's persisted owner and active groups.""" + channel_identifier = str(channel_identifier or "").strip() + if not channel_identifier: + raise ValueError("channel_identifier is required") + + def read_records(path, label): + try: + with open(path, "r", encoding="utf-8") as source: + records = [] + for line_number, line in enumerate(source, 1): + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"Malformed {label} record at line {line_number}" + ) from exc + if not isinstance(record, dict): + raise RuntimeError( + f"Malformed {label} record at line {line_number}" + ) + records.append(record) + return records + except FileNotFoundError: + return [] + except (OSError, UnicodeError) as exc: + raise RuntimeError(f"Failed to read {label} records") from exc + + owner_id = None + for record in read_records(_channel_auth_user_path(), "channel authenticated user"): + saved_channel = str(record.get("channel_identifier", "")).strip() + saved_user = str(record.get("user_id", "")).strip() + if saved_channel == channel_identifier and saved_user and owner_id is None: + owner_id = saved_user + + group_states = {} + for record in read_records(_channel_auth_group_path(), "channel authenticated group"): + saved_channel = str(record.get("channel_identifier", "")).strip() + saved_group = str(record.get("group_id", "")).strip() + if saved_channel == channel_identifier and saved_group: + group_states[saved_group] = not bool(record.get("revoked", False)) + + authorized_groups = { + group_id for group_id, authorized in group_states.items() if authorized + } + return owner_id, authorized_groups + + def store_channel_authenticated_group_id(channel_identifier, group_id, authorized_by_user_id): """Persist a trusted group. Never touches the single-user auth file.""" channel_identifier = str(channel_identifier or "").strip() diff --git a/channels/telegram.py b/channels/telegram.py index 3d2be696..b8ace781 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -31,6 +31,8 @@ _offset = None _connected = False _admin_allowed_chats = set() +_owner_id = None +_authorized_groups = set() _auto_bound_chat = "" @@ -171,7 +173,7 @@ def _command_argument(msg): def _is_allowed_message(chat_id, user_id, chat_type, msg): - global _auto_bound_chat + global _auto_bound_chat, _owner_id auth_enabled = auth.is_auth_enabled() @@ -185,7 +187,7 @@ def _is_allowed_message(chat_id, user_id, chat_type, msg): _auto_bound_chat = chat_id return "allow" - owner_id = auth.get_channel_authenticated_user_id("TELEGRAM") + owner_id = _owner_id is_owner_bootstrap = ( owner_id is None @@ -210,28 +212,40 @@ def _is_allowed_message(chat_id, user_id, chat_type, msg): # The reusable secret is accepted only in a private DM. The owner can if is_owner_bootstrap: candidate = _parse_auth_candidate(msg) - return auth.authenticate_channel_user("TELEGRAM", user_id, candidate) + state = auth.authenticate_channel_user("TELEGRAM", user_id, candidate) + if state == "auth_bound": + _owner_id = user_id + return state return "ignore" if chat_type == "private": if user_id == owner_id and _is_unbind_command(msg): group_id = _command_argument(msg) if group_id: - return auth.revoke_channel_group("TELEGRAM", group_id, user_id) + state = auth.revoke_channel_group("TELEGRAM", group_id, user_id) + if state == "group_unbound": + _authorized_groups.discard(group_id) + return state return "ignore" return "allow" if user_id == owner_id else "ignore" # Anything that isn't "private" is a group/supergroup chat. if _is_unbind_command(msg): if user_id == owner_id: - return auth.revoke_channel_group("TELEGRAM", chat_id, user_id) + state = auth.revoke_channel_group("TELEGRAM", chat_id, user_id) + if state == "group_unbound": + _authorized_groups.discard(chat_id) + return state return "ignore" - if auth.get_channel_saved_group_id("TELEGRAM", chat_id): + if chat_id in _authorized_groups: return "allow" if user_id == owner_id and _is_bind_command(msg): - return auth.authorize_channel_group("TELEGRAM", chat_id, user_id) + state = auth.authorize_channel_group("TELEGRAM", chat_id, user_id) + if state in {"allow", "group_bound"}: + _authorized_groups.add(chat_id) + return state return "ignore" @@ -280,33 +294,58 @@ def _api_call(method, params=None, timeout=30, use_post=False): def _initialize_bot_identity(): global _bot_username - result = _api_call("getMe", timeout=10) - if not isinstance(result, dict): - raise RuntimeError("Telegram getMe returned an invalid response") - - username = str(result.get("username", "")).strip().lstrip("@").lower() - if not username: - raise RuntimeError("Telegram getMe did not return the bot username") + try: + result = _api_call("getMe", timeout=10) + if not isinstance(result, dict): + raise RuntimeError("Telegram getMe returned an invalid response") + username = str(result.get("username", "")).strip().lstrip("@").lower() + if not username: + raise RuntimeError("Telegram getMe did not return the bot username") + except Exception as exc: + _bot_username = "" + logger.warning(f"Could not read Telegram bot identity: {exc}") + return False _bot_username = username + return True -def _initialize_offset(): - global _offset - try: - updates = _api_call("getUpdates", {"timeout": 0}, timeout=10) or [] - except Exception as exc: - logger.warning(f"Could not read initial offset: {exc}") + +def _process_update(update): + message = update.get("message") or update.get("edited_message") + if not isinstance(message, dict): return - max_update = -1 - for update in updates: - update_id = update.get("update_id") - if isinstance(update_id, int): - max_update = max(max_update, update_id) + text = message.get("text") + if not text: + return + + chat = message.get("chat") or {} + user = message.get("from") or {} + chat_id = str(chat.get("id", "")).strip() + user_id = str(user.get("id", "")).strip() + chat_type = str(chat.get("type", "")).strip() + if not chat_id or not user_id: + return + + state = _is_allowed_message(chat_id, user_id, chat_type, text) + display_name = _display_name(user, chat) + + if state == "allow": + _enqueue_message(f"{display_name}: {text}", chat_id) + elif state == "auth_bound": + send_message( + f"Authentication successful. {display_name} is now the bot owner. " + "Send /bind in a group to open it to everyone there.", + chat_id, + ) + elif state == "group_bound": + send_message( + "This group is now authorized. All members can talk to the bot here.", + chat_id, + ) + elif state == "group_unbound": + send_message("This group is no longer authorized.", chat_id) - if max_update >= 0: - with _state_lock: - _offset = max_update + 1 def _poll_loop(): @@ -326,48 +365,11 @@ def _poll_loop(): for update in updates: update_id = update.get("update_id") + _process_update(update) if isinstance(update_id, int): with _state_lock: if _offset is None or (update_id + 1) > _offset: _offset = update_id + 1 - - message = update.get("message") or update.get("edited_message") - if not isinstance(message, dict): - continue - - text = message.get("text") - if not text: - continue - - chat = message.get("chat") or {} - user = message.get("from") or {} - chat_id = str(chat.get("id", "")).strip() - user_id = str(user.get("id", "")).strip() - chat_type = str(chat.get("type", "")).strip() - if not chat_id or not user_id: - continue - - state = _is_allowed_message(chat_id, user_id, chat_type, text) - display_name = _display_name(user, chat) - - if state == "allow": - _enqueue_message(f"{display_name}: {text}", chat_id) - elif state == "auth_bound": - send_message( - f"Authentication successful. {display_name} is now the bot owner. " - "Send /bind in a group to open it to everyone there.", - chat_id, - ) - elif state == "group_bound": - send_message( - "This group is now authorized. All members can talk to the bot here.", - chat_id, - ) - elif state == "group_unbound": - send_message( - "This group is no longer authorized.", - chat_id, - ) except Exception as exc: _connected = False logger.warning(f"Poll error: {exc}") @@ -393,7 +395,7 @@ def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): global _running, _bot_token, _api_base, _poll_timeout, _offset, _connected global _active_chat_id, _active_message_token, _active_replied global _next_message_token, _default_chat_id - global _admin_allowed_chats, _auto_bound_chat + global _admin_allowed_chats, _auto_bound_chat, _owner_id, _authorized_groups proxy = auth.get_proxy_url() if proxy: @@ -408,6 +410,10 @@ def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): _admin_allowed_chats = _parse_admin_allowed_chats(chat_id, allowed_chat_ids) _default_chat_id = str(chat_id or "").strip() _auto_bound_chat = "" + if auth.is_auth_enabled(): + _owner_id, _authorized_groups = auth.load_channel_auth_state("TELEGRAM") + else: + _owner_id, _authorized_groups = None, set() with _msg_lock: _inbox.clear() @@ -432,7 +438,6 @@ def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): logger.info("Starting adapter with no admin chat restriction") _initialize_bot_identity() - _initialize_offset() t = threading.Thread(target=_poll_loop, daemon=True) t.start() diff --git a/docs/reference-channels.md b/docs/reference-channels.md index 7d8b69d8..70216104 100644 --- a/docs/reference-channels.md +++ b/docs/reference-channels.md @@ -50,10 +50,12 @@ Mattermost adapter using a bot token. Telegram adapter using Bot API long polling. -- `start_telegram(chat_id, poll_timeout)` — starts a poll loop. -- `TG_CHAT_ID` is optional; if empty, the adapter can auto-bind to the first valid inbound chat. -- Outbound messages are chunked to Telegram-safe lengths. -- Uses the same one-time `auth ` ownership gate as the other adapters. +- `start_telegram(chat_id, allowed_chat_ids, poll_timeout)` — validates saved authorization state and starts the poll loop. +- `TG_CHAT_ID` is the default destination for startup, heartbeat, and other proactive messages. If it is empty, proactive messages are not sent until a chat is active. +- With authentication enabled, the owner authenticates with `auth ` in a private DM, then uses `/bind` in a group to authorize all members of that group. +- The owner can revoke the current group with `/unbind`, or revoke a group from DM with `/unbind `. `/bind@BotName` and `/unbind@BotName` are also supported. +- Authorized chats are processed serially: a later chat remains queued until the active chat receives its response. +- Outbound messages are split into Telegram-safe chunks and retained with their destination for retry after transient delivery failures. ## `channels/slack.py` diff --git a/docs/reference-configuration.md b/docs/reference-configuration.md index ee4ffe9c..add91740 100644 --- a/docs/reference-configuration.md +++ b/docs/reference-configuration.md @@ -42,7 +42,8 @@ This reads a command-line override via `argk` (`name=value` on the MeTTa command | `IRC_server` | `irc.quakenet.org` | IRC server hostname. | | `IRC_port` | 6667 | IRC port. | | `IRC_user` | `omegaclaw` | IRC nickname. | -| `TG_CHAT_ID` | *(empty — auto-bind supported)* | Optional fixed Telegram chat ID. Leave empty to auto-bind on first valid inbound auth/message. | +| `TG_CHAT_ID` | *(empty — auto-bind supported)* | Optional Telegram chat ID used as the default destination for startup, heartbeat, and other proactive messages. Leave empty to auto-bind inbound traffic without a proactive destination. | +| `TG_ALLOWED_CHAT_IDS` | *(empty)* | Optional comma-separated boundary for Telegram chats. Owner authentication in private DM remains available when authentication is enabled. | | `TG_POLL_TIMEOUT` | 20 | Telegram long-poll timeout in seconds. | | `SL_CHANNEL_ID` | *(empty — auto-bind supported)* | Optional Slack channel ID where OmegaClaw reads/writes messages. Leave empty to auto-bind on first valid inbound auth/message. | | `SL_POLL_INTERVAL` | 60 | Slack poll interval in seconds (minimum effective value is 60). | diff --git a/tests/test_auth_standalone.py b/tests/test_auth_standalone.py index 5e0050e5..0a964d6e 100644 --- a/tests/test_auth_standalone.py +++ b/tests/test_auth_standalone.py @@ -130,3 +130,29 @@ def test_owner_can_revoke_an_authorized_group(monkeypatch, tmp_path): assert auth.store_channel_authenticated_group_id("TELEGRAM", "group", "owner") is True assert auth.get_channel_saved_group_id("TELEGRAM", "group") is True + + +def test_load_channel_auth_state_validates_and_loads_records(monkeypatch, tmp_path): + auth = load_auth_module(monkeypatch) + monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path)) + + assert auth.store_channel_authenticated_user_id("TELEGRAM", "owner") is True + assert auth.store_channel_authenticated_group_id("TELEGRAM", "active", "owner") is True + assert auth.store_channel_authenticated_group_id("TELEGRAM", "revoked", "owner") is True + assert auth.revoke_channel_group("TELEGRAM", "revoked", "owner") == "group_unbound" + + owner, groups = auth.load_channel_auth_state("TELEGRAM") + + assert owner == "owner" + assert groups == {"active"} + + +def test_load_channel_auth_state_rejects_malformed_records(monkeypatch, tmp_path): + auth = load_auth_module(monkeypatch) + monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path)) + path = tmp_path / ".channel" / "authenticated-group.json" + path.parent.mkdir() + path.write_text("not-json\n", encoding="utf-8") + + with pytest.raises(RuntimeError, match="Malformed channel authenticated group record"): + auth.load_channel_auth_state("TELEGRAM") diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py index 02265949..febd1a97 100644 --- a/tests/test_telegram_multichat.py +++ b/tests/test_telegram_multichat.py @@ -3,6 +3,8 @@ import types from pathlib import Path +import pytest + REPO_ROOT = Path(__file__).resolve().parents[1] CHANNELS_DIRECTORY = REPO_ROOT / "channels" @@ -13,6 +15,10 @@ def load_telegram(monkeypatch, auth_enabled=True): auth = types.ModuleType("auth") auth.is_auth_enabled = lambda: auth_enabled auth.get_proxy_url = lambda: "" + auth.load_channel_auth_state = lambda _channel: ( + state["owner"], + {group for channel, group in state["groups"] if channel == "TELEGRAM"}, + ) auth.get_channel_authenticated_user_id = lambda channel: state["owner"] auth.get_channel_saved_group_id = lambda channel, group: ( channel, str(group) @@ -128,6 +134,91 @@ def test_proactive_message_uses_only_the_configured_default_chat(monkeypatch): assert sent[0][1]["chat_id"] == "configured-default" +def test_cached_authorization_avoids_disk_reads_per_message(monkeypatch): + telegram = load_telegram(monkeypatch) + telegram._owner_id = "owner" + telegram._authorized_groups = {"group"} + telegram.auth.get_channel_authenticated_user_id = lambda *_args: (_ for _ in ()).throw( + AssertionError("unexpected owner file read") + ) + telegram.auth.get_channel_saved_group_id = lambda *_args: (_ for _ in ()).throw( + AssertionError("unexpected group file read") + ) + + assert telegram._is_allowed_message("owner-dm", "owner", "private", "hello") == "allow" + assert telegram._is_allowed_message("group", "member", "group", "hello") == "allow" + + +def test_get_me_failure_does_not_abort_initialization(monkeypatch): + telegram = load_telegram(monkeypatch, auth_enabled=False) + telegram._bot_username = "stale-name" + telegram._api_call = lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("temporary Telegram failure") + ) + + assert telegram._initialize_bot_identity() is False + assert telegram._bot_username == "" + + +def test_invalid_auth_state_stops_startup_before_telegram_is_polled(monkeypatch): + telegram = load_telegram(monkeypatch) + monkeypatch.setenv("TG_BOT_TOKEN", "token") + telegram.auth.load_channel_auth_state = lambda _channel: (_ for _ in ()).throw( + RuntimeError("malformed authorization state") + ) + telegram._api_call = lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("Telegram API called before authorization validation") + ) + + with pytest.raises(RuntimeError, match="malformed authorization state"): + telegram.start_telegram() + + +def test_offset_advances_only_after_update_processing(monkeypatch): + telegram = load_telegram(monkeypatch, auth_enabled=False) + update = { + "update_id": 7, + "message": { + "text": "hello", + "chat": {"id": "dm", "type": "private"}, + "from": {"id": "user", "username": "alice"}, + }, + } + telegram._running = True + telegram._offset = None + telegram._flush_outbox = lambda: None + + def get_updates(*_args, **_kwargs): + telegram._running = False + return [update] + + telegram._api_call = get_updates + telegram._poll_loop() + + assert telegram._offset == 8 + assert telegram.getLastMessage() == "@alice: hello" + + +def test_failed_update_processing_retains_offset(monkeypatch): + telegram = load_telegram(monkeypatch, auth_enabled=False) + telegram._running = True + telegram._offset = None + telegram._flush_outbox = lambda: None + telegram.time.sleep = lambda _seconds: None + + def get_updates(*_args, **_kwargs): + telegram._running = False + return [{"update_id": 7, "message": {"text": "hello"}}] + + telegram._api_call = get_updates + telegram._process_update = lambda _update: (_ for _ in ()).throw( + RuntimeError("authorization state failure") + ) + telegram._poll_loop() + + assert telegram._offset is None + + def test_owner_binds_group_without_exposing_secret(monkeypatch): telegram = load_telegram(monkeypatch) telegram._bot_username = "examplebot" From 1169b644ab0f6acd5026bd7cd21939778b1aa65e Mon Sep 17 00:00:00 2001 From: CodersKin Date: Fri, 28 Aug 2026 11:55:20 +0300 Subject: [PATCH 08/11] Fix: Changed message routing scheme and added prompt extension for routing descriptions --- channels/auth.py | 30 ++++- channels/telegram.py | 191 ++++++++++++++++++++++--------- docs/reference-channels.md | 13 ++- docs/reference-configuration.md | 17 ++- memory/tg_prompt.txt | 19 +++ src/loop.metta | 18 ++- tests/src_skills.metta | 16 +++ tests/test_auth_standalone.py | 21 ++++ tests/test_telegram_multichat.py | 163 ++++++++++++++++++++------ 9 files changed, 385 insertions(+), 103 deletions(-) create mode 100644 memory/tg_prompt.txt diff --git a/channels/auth.py b/channels/auth.py index bdf538c7..7283c148 100644 --- a/channels/auth.py +++ b/channels/auth.py @@ -128,7 +128,7 @@ def authenticate_channel_user(channel_identifier, user_id, auth_candidate=None): channel_identifier = str(channel_identifier or "").strip() user_id = str(user_id or "").strip() - # A persisted owner always wins over a reusable secret. + # A persisted owner always wins over a reusable secret. saved_user_id = get_channel_authenticated_user_id(channel_identifier) if saved_user_id is not None: return "allow" if saved_user_id == user_id else "ignore" @@ -173,7 +173,7 @@ def get_channel_authenticated_user_id(channel_identifier): # --------------------------------------------------------------------------- -# Telegram-only group authorization. +# Owner-managed group authorization. def _channel_auth_group_path(): return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_GROUP_FILE) @@ -220,8 +220,19 @@ def read_records(path, label): for record in read_records(_channel_auth_group_path(), "channel authenticated group"): saved_channel = str(record.get("channel_identifier", "")).strip() saved_group = str(record.get("group_id", "")).strip() - if saved_channel == channel_identifier and saved_group: + authorized_by = str(record.get("authorized_by", "")).strip() + if ( + saved_channel == channel_identifier + and saved_group + and owner_id is not None + and authorized_by == owner_id + ): group_states[saved_group] = not bool(record.get("revoked", False)) + elif saved_channel == channel_identifier and saved_group: + logger.warning( + f"[{channel_identifier.upper()}] Skipping group authorization " + f"for {saved_group}: record is not owned by the authenticated user" + ) authorized_groups = { group_id for group_id, authorized in group_states.items() if authorized @@ -238,6 +249,8 @@ def store_channel_authenticated_group_id(channel_identifier, group_id, authorize raise ValueError("channel_identifier is required") if not group_id: raise ValueError("group_id is required") + if not authorized_by_user_id: + raise ValueError("authorized_by_user_id is required") payload = { "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), @@ -262,6 +275,10 @@ def get_channel_saved_group_id(channel_identifier, group_id): group_id = str(group_id or "").strip() if not channel_identifier or not group_id: return False + owner_id = get_channel_authenticated_user_id(channel_identifier) + if owner_id is None: + return False + authorized = False try: with open(_channel_auth_group_path(), "r", encoding="utf-8") as f: @@ -270,10 +287,15 @@ def get_channel_saved_group_id(channel_identifier, group_id): record = json.loads(line) saved_channel = str(record.get("channel_identifier", "")).strip() saved_group = str(record.get("group_id", "")).strip() + authorized_by = str(record.get("authorized_by", "")).strip() except (AttributeError, json.JSONDecodeError) as e: logger.warning(f"Skipping malformed channel authenticated group record: {e}") continue - if saved_channel == channel_identifier and saved_group == group_id: + if ( + saved_channel == channel_identifier + and saved_group == group_id + and authorized_by == owner_id + ): authorized = not bool(record.get("revoked", False)) except FileNotFoundError: return False diff --git a/channels/telegram.py b/channels/telegram.py index b8ace781..6a50951c 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -1,5 +1,6 @@ import json import os +import re import threading import time import urllib.parse @@ -17,12 +18,9 @@ _msg_lock = threading.Lock() _state_lock = threading.Lock() _inbox = deque() -_active_chat_id = "" -_active_message_token = None -_active_replied = False -_next_message_token = 0 _default_chat_id = "" _outbox = PendingMessages() +_deferred_default_outbox = PendingMessages() _bot_token = "" _api_base = "" @@ -38,30 +36,30 @@ _BIND_COMMANDS = ("/bind", "/authorize_group") _UNBIND_COMMANDS = ("/unbind", "/unauthorize_group") +_ROUTED_MESSAGE_RE = re.compile( + r"^\s*\[(-?\d*)\]\s*\[(\d*)\]\s*(.*)$", + re.DOTALL, +) +_TARGET_ONLY_MESSAGE_RE = re.compile(r"^\s*\[(-?\d+)\]\s*(.*)$", re.DOTALL) -def _enqueue_message(msg, chat_id): - global _next_message_token +def _enqueue_message(msg, chat_id, reply_to_id=None): with _msg_lock: - _next_message_token += 1 - _inbox.append((_next_message_token, str(chat_id), str(msg))) + _inbox.append( + ( + str(chat_id), + str(reply_to_id) if reply_to_id is not None else "", + str(msg), + ) + ) def getLastMessage(): - global _active_chat_id, _active_message_token, _active_replied with _msg_lock: - if _active_chat_id and not _active_replied: - return "" - if _active_replied: - _active_chat_id = "" - _active_message_token = None - _active_replied = False if not _inbox: return "" - message_token, chat_id, message = _inbox.popleft() - _active_chat_id = chat_id - _active_message_token = message_token - return message + chat_id, reply_to_id, message = _inbox.popleft() + return f"[{chat_id}] [{reply_to_id}] {message}" def _ready_to_send(): @@ -69,18 +67,22 @@ def _ready_to_send(): def _deliver_outbound(item): - global _active_replied - target_chat, chunk, completed_message_token = item + target_chat, reply_to_id, chunk = item + params = {"chat_id": target_chat, "text": chunk} + if reply_to_id: + params["reply_parameters"] = json.dumps( + { + "message_id": int(reply_to_id), + "allow_sending_without_reply": True, + }, + separators=(",", ":"), + ) _api_call( "sendMessage", - {"chat_id": target_chat, "text": chunk}, + params, timeout=15, use_post=True, ) - if completed_message_token is not None: - with _msg_lock: - if completed_message_token == _active_message_token: - _active_replied = True def _flush_outbox(): @@ -90,32 +92,92 @@ def _flush_outbox(): logger.warning(f"Telegram send failed; retaining queued message: {exc}") -def send_message(text, target_chat=None): +def _ready_to_route_deferred_default(): + with _state_lock: + return bool(_default_chat_id) + + +def _route_deferred_default(item): + reply_to_id, chunk = item + with _state_lock: + target_chat = str(_default_chat_id or "").strip() + if not target_chat: + raise RuntimeError("Telegram owner DM is not available") + _outbox.put((target_chat, reply_to_id, chunk)) + + +def _flush_deferred_default_outbox(): + try: + _deferred_default_outbox.flush( + _route_deferred_default, + _ready_to_route_deferred_default, + ) + except Exception as exc: + logger.warning( + f"Telegram deferred send failed; retaining queued message: {exc}" + ) + return + _flush_outbox() + + +def _parse_outbound_message(text): + """Return target, reply message ID, body, and whether routing was explicit.""" + match = _ROUTED_MESSAGE_RE.match(text) + if match: + return match.group(1), match.group(2), match.group(3), True + + # Backward compatibility with the upstream Telegram form: [chat_id] body. + match = _TARGET_ONLY_MESSAGE_RE.match(text) + if match: + return match.group(1), "", match.group(2), True + + return "", "", text, False + + +def _is_allowed_outbound_target(chat_id): + chat_id = str(chat_id or "").strip() + if not chat_id: + return False + + if auth.is_auth_enabled(): + return chat_id == str(_owner_id or "") or chat_id in _authorized_groups + + if _admin_allowed_chats: + return chat_id in _admin_allowed_chats + return chat_id in {str(_default_chat_id or ""), str(_auto_bound_chat or "")} + + +def send_message(text, target_chat=None, reply_to_id=None): text = str(text).replace("\\n", "\n").replace("\r", "") if not text: return False - explicit_target = str(target_chat or "").strip() - with _msg_lock: - active_chat = _active_chat_id - active_message_token = _active_message_token - target_chat = explicit_target or active_chat or _default_chat_id - completed_message_token = ( - active_message_token - if not explicit_target and active_chat and target_chat == active_chat - else None - ) + trusted_target = str(target_chat or "").strip() + if trusted_target: + target_chat = trusted_target + reply_to_id = str(reply_to_id or "").strip() + else: + parsed_target, parsed_reply, text, routed = _parse_outbound_message(text) + target_chat = parsed_target or str(_default_chat_id or "").strip() + reply_to_id = parsed_reply + if routed and parsed_target and not _is_allowed_outbound_target(parsed_target): + logger.warning( + f"Telegram send rejected: target chat {parsed_target} is not authorized" + ) + return False - if not target_chat: - logger.warning("Telegram send skipped: no active or default chat is available") + if not text: + logger.warning("Telegram send skipped: routed message body is empty") return False max_len = 3900 chunks = [text[i:i + max_len] for i in range(0, len(text), max_len)] - outbound = [] - for index, chunk in enumerate(chunks): - completes_message = completed_message_token if index == len(chunks) - 1 else None - outbound.append((target_chat, chunk, completes_message)) + if not target_chat: + _deferred_default_outbox.extend((reply_to_id, chunk) for chunk in chunks) + logger.info("Telegram send deferred until an owner DM is available") + return True + + outbound = [(target_chat, reply_to_id, chunk) for chunk in chunks] _outbox.extend(outbound) _flush_outbox() return True @@ -173,7 +235,7 @@ def _command_argument(msg): def _is_allowed_message(chat_id, user_id, chat_type, msg): - global _auto_bound_chat, _owner_id + global _auto_bound_chat, _owner_id, _default_chat_id auth_enabled = auth.is_auth_enabled() @@ -185,6 +247,7 @@ def _is_allowed_message(chat_id, user_id, chat_type, msg): return "ignore" if not _auto_bound_chat: _auto_bound_chat = chat_id + _default_chat_id = chat_id return "allow" owner_id = _owner_id @@ -199,22 +262,30 @@ def _is_allowed_message(chat_id, user_id, chat_type, msg): and chat_type == "private" and user_id == owner_id ) + is_owner_group_bind = ( + owner_id is not None + and chat_type != "private" + and user_id == owner_id + and _is_bind_command(msg) + ) if ( _admin_allowed_chats and chat_id not in _admin_allowed_chats and not is_owner_bootstrap and not is_owner_private_chat + and not is_owner_group_bind ): return "ignore" if owner_id is None: - # The reusable secret is accepted only in a private DM. The owner can if is_owner_bootstrap: candidate = _parse_auth_candidate(msg) state = auth.authenticate_channel_user("TELEGRAM", user_id, candidate) if state == "auth_bound": - _owner_id = user_id + with _state_lock: + _owner_id = user_id + _default_chat_id = chat_id return state return "ignore" @@ -225,6 +296,7 @@ def _is_allowed_message(chat_id, user_id, chat_type, msg): state = auth.revoke_channel_group("TELEGRAM", group_id, user_id) if state == "group_unbound": _authorized_groups.discard(group_id) + _admin_allowed_chats.discard(group_id) return state return "ignore" return "allow" if user_id == owner_id else "ignore" @@ -235,6 +307,7 @@ def _is_allowed_message(chat_id, user_id, chat_type, msg): state = auth.revoke_channel_group("TELEGRAM", chat_id, user_id) if state == "group_unbound": _authorized_groups.discard(chat_id) + _admin_allowed_chats.discard(chat_id) return state return "ignore" @@ -245,6 +318,7 @@ def _is_allowed_message(chat_id, user_id, chat_type, msg): state = auth.authorize_channel_group("TELEGRAM", chat_id, user_id) if state in {"allow", "group_bound"}: _authorized_groups.add(chat_id) + _admin_allowed_chats.add(chat_id) return state return "ignore" @@ -324,6 +398,7 @@ def _process_update(update): chat_id = str(chat.get("id", "")).strip() user_id = str(user.get("id", "")).strip() chat_type = str(chat.get("type", "")).strip() + message_id = message.get("message_id") if not chat_id or not user_id: return @@ -331,20 +406,24 @@ def _process_update(update): display_name = _display_name(user, chat) if state == "allow": - _enqueue_message(f"{display_name}: {text}", chat_id) + _flush_deferred_default_outbox() + _enqueue_message(f"{display_name}: {text}", chat_id, message_id) elif state == "auth_bound": send_message( f"Authentication successful. {display_name} is now the bot owner. " "Send /bind in a group to open it to everyone there.", chat_id, + message_id ) + _flush_deferred_default_outbox() elif state == "group_bound": send_message( "This group is now authorized. All members can talk to the bot here.", chat_id, + message_id ) elif state == "group_unbound": - send_message("This group is no longer authorized.", chat_id) + send_message("This group is no longer authorized.", chat_id, message_id) @@ -393,8 +472,7 @@ def _parse_admin_allowed_chats(chat_id_config, allowed_config): def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): global _running, _bot_token, _api_base, _poll_timeout, _offset, _connected - global _active_chat_id, _active_message_token, _active_replied - global _next_message_token, _default_chat_id + global _default_chat_id global _admin_allowed_chats, _auto_bound_chat, _owner_id, _authorized_groups proxy = auth.get_proxy_url() @@ -408,20 +486,21 @@ def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): _api_base = f"https://api.telegram.org/bot{_bot_token}" _admin_allowed_chats = _parse_admin_allowed_chats(chat_id, allowed_chat_ids) - _default_chat_id = str(chat_id or "").strip() _auto_bound_chat = "" if auth.is_auth_enabled(): _owner_id, _authorized_groups = auth.load_channel_auth_state("TELEGRAM") + _admin_allowed_chats.update(_authorized_groups) + # Telegram private chat IDs are the owner's user ID, so persisted + # ownership gives proactive messages a safe destination after restart. + _default_chat_id = str(_owner_id or chat_id or "").strip() else: _owner_id, _authorized_groups = None, set() + _default_chat_id = str(chat_id or "").strip() with _msg_lock: _inbox.clear() - _active_chat_id = "" - _active_message_token = None - _active_replied = False - _next_message_token = 0 _outbox.clear() + _deferred_default_outbox.clear() try: _poll_timeout = max(1, int(poll_timeout)) @@ -436,7 +515,7 @@ def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20): logger.info(f"Starting adapter, admin-restricted to chats: {sorted(_admin_allowed_chats)}") else: logger.info("Starting adapter with no admin chat restriction") - + _initialize_bot_identity() t = threading.Thread(target=_poll_loop, daemon=True) diff --git a/docs/reference-channels.md b/docs/reference-channels.md index 70216104..4c16303a 100644 --- a/docs/reference-channels.md +++ b/docs/reference-channels.md @@ -51,11 +51,14 @@ Mattermost adapter using a bot token. Telegram adapter using Bot API long polling. - `start_telegram(chat_id, allowed_chat_ids, poll_timeout)` — validates saved authorization state and starts the poll loop. -- `TG_CHAT_ID` is the default destination for startup, heartbeat, and other proactive messages. If it is empty, proactive messages are not sent until a chat is active. -- With authentication enabled, the owner authenticates with `auth ` in a private DM, then uses `/bind` in a group to authorize all members of that group. -- The owner can revoke the current group with `/unbind`, or revoke a group from DM with `/unbind `. `/bind@BotName` and `/unbind@BotName` are also supported. -- Authorized chats are processed serially: a later chat remains queued until the active chat receives its response. -- Outbound messages are split into Telegram-safe chunks and retained with their destination for retry after transient delivery failures. +- With authentication enabled, the owner authenticates with `auth ` in a private DM. That DM becomes the default destination for startup, heartbeat, and other proactive messages and is restored from persisted owner state after restart. +- The owner uses `/bind` in a group to add its chat ID to the runtime and persisted allowed-group sets. `/unbind` removes the current group; `/unbind ` performs the same operation from the owner's DM. Targeted forms such as `/bind@BotName` and `/unbind@BotName` are supported. +- `TG_ALLOWED_CHAT_IDS` supplies initial operator-configured chat IDs. Runtime `/bind` additions and `/unbind` removals are persisted in `memory/.channel/authenticated-group.json`; the YAML file itself is never rewritten. +- Each inbound message is delivered to the agent as `[chat_id] [message_id] message`. Dequeueing does not depend on the model producing or successfully delivering a reply, so a no-response turn cannot freeze later inbound messages. +- Outbound replies use the same `[chat_id] [message_id] message` envelope. An empty target falls back to the owner DM, and an empty message ID sends without Telegram reply metadata. Legacy plain outbound text also uses the owner-DM fallback. +- Explicit LLM-generated targets are accepted only for the owner DM or a currently authorized group. Group chat IDs may be negative. +- Outbound messages are split into Telegram-safe chunks and retained with their destination and reply ID for retry after transient delivery failures. +- When `commchannel=telegram`, startup registers the routing instructions from `memory/tg_prompt.txt` through `add-prompt-extension`. Other channels do not receive this prompt section. ## `channels/slack.py` diff --git a/docs/reference-configuration.md b/docs/reference-configuration.md index add91740..12eb765c 100644 --- a/docs/reference-configuration.md +++ b/docs/reference-configuration.md @@ -42,8 +42,8 @@ This reads a command-line override via `argk` (`name=value` on the MeTTa command | `IRC_server` | `irc.quakenet.org` | IRC server hostname. | | `IRC_port` | 6667 | IRC port. | | `IRC_user` | `omegaclaw` | IRC nickname. | -| `TG_CHAT_ID` | *(empty — auto-bind supported)* | Optional Telegram chat ID used as the default destination for startup, heartbeat, and other proactive messages. Leave empty to auto-bind inbound traffic without a proactive destination. | -| `TG_ALLOWED_CHAT_IDS` | *(empty)* | Optional comma-separated boundary for Telegram chats. Owner authentication in private DM remains available when authentication is enabled. | +| `TG_CHAT_ID` | *(empty)* | Legacy single-chat bootstrap/fallback, primarily for authentication-disabled deployments. With authentication enabled, the authenticated owner's DM becomes the proactive-message default. | +| `TG_ALLOWED_CHAT_IDS` | *(empty)* | Optional comma-separated initial allowed-chat set. The authenticated owner can add groups with `/bind` and remove them with `/unbind`; those changes are persisted in `memory/.channel/authenticated-group.json`, not written back to YAML. | | `TG_POLL_TIMEOUT` | 20 | Telegram long-poll timeout in seconds. | | `SL_CHANNEL_ID` | *(empty — auto-bind supported)* | Optional Slack channel ID where OmegaClaw reads/writes messages. Leave empty to auto-bind on first valid inbound auth/message. | | `SL_POLL_INTERVAL` | 60 | Slack poll interval in seconds (minimum effective value is 60). | @@ -55,6 +55,7 @@ This reads a command-line override via `argk` (`name=value` on the MeTTa command | Environment variable | Meaning | |---|---| | `TG_BOT_TOKEN` | Telegram bot token (from BotFather). | +| `OMEGACLAW_AUTH_SECRET` | Enables the one-time owner-authentication handshake when non-empty. The owner sends `auth ` in a private Telegram DM. | | `MM_BOT_TOKEN` | Bot auth token. | | `SL_BOT_TOKEN` | Slack bot token (`xoxb-...`). | @@ -66,6 +67,18 @@ Any `configure`d parameter can be overridden at startup: metta run.metta provider=Anthropic LLM=claude-opus-4-6 commchannel=mattermost ``` +Configuration values are resolved in this order: command-line `key=value`, +`OMEGACLAW_` environment variable, `config/config.yaml`, then the caller's +default. `TG_BOT_TOKEN` and `OMEGACLAW_AUTH_SECRET` are read directly from the +environment and must be placed before the `metta`/`petta` command. + +Telegram example: + +```bash +TG_BOT_TOKEN=... OMEGACLAW_AUTH_SECRET=... \ + metta run.metta commchannel=telegram +``` + Slack example: ```bash diff --git a/memory/tg_prompt.txt b/memory/tg_prompt.txt new file mode 100644 index 00000000..9336bc65 --- /dev/null +++ b/memory/tg_prompt.txt @@ -0,0 +1,19 @@ +TELEGRAM ROUTING: + +Incoming Telegram messages use this structure: +[chat-id] [reply-id] message + +The first field is the destination Telegram chat ID. The second field is the +Telegram message ID to reply to. Group chat IDs may be negative. + +When responding to an incoming Telegram message, every send command must begin +with the exact chat ID and reply ID received with that message: +send [chat-id] [reply-id] response +Example: +send [-3341114] [223] Hey! + +Copy both IDs exactly. Never invent, alter, exchange, or omit them when replying. +Multiple send commands responding to the same message must all use the same IDs. + +An empty target routes to the authenticated owner's direct message. An empty +reply ID sends a normal message without attaching it as a reply. diff --git a/src/loop.metta b/src/loop.metta index 2b7b7e95..a3abfc78 100644 --- a/src/loop.metta +++ b/src/loop.metta @@ -6,12 +6,12 @@ (= (reasoningMode) (empty)) (= (wakeupInterval) (empty)) (= (memoryDirectory) (empty)) -(= (spamShield) (empty)) ; TODO: this parameter is considered deprecated +; (= (spamShield) (empty)) ; TODO: this parameter is considered deprecated (= (initLoop) (progn (configure maxNewInputLoops 50) ;20 (configure maxWakeLoops 1) - (configure spamShield False) + ; (configure spamShield False) (configure sleepInterval 1) ;10 (configure provider Anthropic) (configure maxOutputToken 6000) @@ -44,6 +44,15 @@ " HISTORY: " (getHistory) " TIME: " (get_time_as_string))))) +(= (addTelegramPromptExtension) + (if (== (commchannel) telegram) + (let $path (joinPath ((memoryDirectory) "tg_prompt.txt")) + (if (exists-file $path) + (add-prompt-extension tg-prompt (read-file $path)) + (progn (log WARN "loop" (TELEGRAM_PROMPT_NOT_FOUND: $path)) + ()))) + ())) + (= (getPromptExtensions) (join (newline) (collapse (prompt-extension $_)))) @@ -61,11 +70,12 @@ (progn (if (== $k 1) (progn (initConfig) (initLoop) (initLogger) - (applySecurityPolicy) + ; (applySecurityPolicy) (initMemory) - (initKnowledge) + ; (initKnowledge) (initPlugins) (initChannels) + (addTelegramPromptExtension) (commChannelSend (version)) (llmProviderStart (provider))) (change-state! &loops (- (get-state &loops) 1))) diff --git a/tests/src_skills.metta b/tests/src_skills.metta index 24ad96ac..18d764df 100644 --- a/tests/src_skills.metta +++ b/tests/src_skills.metta @@ -2,6 +2,12 @@ !(import! &self ../src/helper.py) !(import! &self ./src/utils) !(import! &self ./src/skills) + +; Define the selected channel and memory path used by the Telegram prompt +; registration test. +(= (commchannel) telegram) +(= (memoryDirectory) (joinPath ((projectRootDirectory) "memory"))) + !(import! &self ./src/loop) !(test (progn @@ -24,6 +30,16 @@ (contains-text (getPromptExtensions) "TEST PROMPT EXTENSION")) False) +!(test (progn + (addTelegramPromptExtension) + (contains-text (getPromptExtensions) "TELEGRAM ROUTING:")) + True) + +!(test (progn + (remove-prompt-extension tg-prompt) + (contains-text (getPromptExtensions) "TELEGRAM ROUTING:")) + False) + (= (on-heartbeat $iter) (let $count (get-state &test_heartbeat_counter) (change-state! &test_heartbeat_counter (+ $count $iter)))) diff --git a/tests/test_auth_standalone.py b/tests/test_auth_standalone.py index 0a964d6e..7852d89a 100644 --- a/tests/test_auth_standalone.py +++ b/tests/test_auth_standalone.py @@ -132,6 +132,27 @@ def test_owner_can_revoke_an_authorized_group(monkeypatch, tmp_path): assert auth.get_channel_saved_group_id("TELEGRAM", "group") is True +def test_group_record_requires_authorizing_owner(monkeypatch, tmp_path): + auth = load_auth_module(monkeypatch) + monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path)) + + with pytest.raises(ValueError, match="authorized_by_user_id is required"): + auth.store_channel_authenticated_group_id("TELEGRAM", "group", "") + + +def test_group_records_from_non_owner_are_not_loaded(monkeypatch, tmp_path): + auth = load_auth_module(monkeypatch) + monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path)) + + assert auth.store_channel_authenticated_user_id("TELEGRAM", "owner") is True + assert auth.store_channel_authenticated_group_id( + "TELEGRAM", "forged-group", "attacker" + ) is True + + assert auth.get_channel_saved_group_id("TELEGRAM", "forged-group") is False + assert auth.load_channel_auth_state("TELEGRAM") == ("owner", set()) + + def test_load_channel_auth_state_validates_and_loads_records(monkeypatch, tmp_path): auth = load_auth_module(monkeypatch) monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path)) diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py index febd1a97..1865f888 100644 --- a/tests/test_telegram_multichat.py +++ b/tests/test_telegram_multichat.py @@ -1,4 +1,5 @@ import importlib.util +import json import sys import types from pathlib import Path @@ -68,40 +69,39 @@ def revoke_channel_group(channel, group, requester): def test_reply_uses_the_chat_that_supplied_the_message(monkeypatch): telegram = load_telegram(monkeypatch, auth_enabled=False) telegram._connected = True + telegram._admin_allowed_chats = {"101", "-202"} sent = [] telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) - telegram._enqueue_message("dm message", "dm") - telegram._enqueue_message("group message", "group") - assert telegram.getLastMessage() == "dm message" - assert telegram.getLastMessage() == "" - telegram.send_message("dm reply") - assert telegram.getLastMessage() == "group message" - telegram.send_message("group reply") + telegram._enqueue_message("dm message", "101", 11) + telegram._enqueue_message("group message", "-202", 12) + assert telegram.getLastMessage() == "[101] [11] dm message" + telegram.send_message("[101] [11] dm reply") + assert telegram.getLastMessage() == "[-202] [12] group message" + telegram.send_message("[-202] [12] group reply") - assert [params["chat_id"] for _, params in sent] == ["dm", "group"] + assert [params["chat_id"] for _, params in sent] == ["101", "-202"] + assert [ + json.loads(params["reply_parameters"])["message_id"] + for _, params in sent + ] == [11, 12] def test_identical_messages_from_different_chats_are_processed(monkeypatch): telegram = load_telegram(monkeypatch, auth_enabled=False) telegram._connected = True - sent = [] - telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) + telegram._enqueue_message("same message", "101", 21) + telegram._enqueue_message("same message", "-202", 22) - telegram._enqueue_message("same message", "dm") - telegram._enqueue_message("same message", "group") - - assert telegram.getLastMessage() == "same message" - telegram.send_message("dm reply") - assert telegram.getLastMessage() == "same message" - telegram.send_message("group reply") - - assert [params["chat_id"] for _, params in sent] == ["dm", "group"] + assert telegram.getLastMessage() == "[101] [21] same message" + assert telegram.getLastMessage() == "[-202] [22] same message" + assert telegram.getLastMessage() == "" def test_failed_delivery_retains_the_original_chat(monkeypatch): telegram = load_telegram(monkeypatch, auth_enabled=False) telegram._connected = True + telegram._admin_allowed_chats = {"101", "-202"} attempts = [] def flaky_api(method, params, **_kwargs): @@ -110,28 +110,90 @@ def flaky_api(method, params, **_kwargs): raise RuntimeError("temporary failure") telegram._api_call = flaky_api - telegram._enqueue_message("dm message", "dm") - assert telegram.getLastMessage() == "dm message" + telegram._enqueue_message("dm message", "101", 31) + assert telegram.getLastMessage() == "[101] [31] dm message" - telegram.send_message("dm reply") - telegram._enqueue_message("group message", "group") - assert telegram.getLastMessage() == "" + telegram.send_message("[101] [31] dm reply") + telegram._enqueue_message("group message", "-202", 32) + # A failed outbound delivery does not block the inbound queue. + assert telegram.getLastMessage() == "[-202] [32] group message" telegram._flush_outbox() - assert telegram.getLastMessage() == "group message" - assert [params["chat_id"] for _, params in attempts] == ["dm", "dm"] + assert [params["chat_id"] for _, params in attempts] == ["101", "101"] -def test_proactive_message_uses_only_the_configured_default_chat(monkeypatch): - telegram = load_telegram(monkeypatch, auth_enabled=False) +def test_proactive_message_uses_authenticated_owner_dm(monkeypatch): + telegram = load_telegram(monkeypatch) telegram._connected = True - telegram._default_chat_id = "configured-default" + telegram._owner_id = "101" + telegram._default_chat_id = "101" sent = [] telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) telegram.send_message("startup message") - assert sent[0][1]["chat_id"] == "configured-default" + assert sent[0][1]["chat_id"] == "101" + assert "reply_parameters" not in sent[0][1] + + +def test_proactive_message_before_auth_is_sent_after_owner_binding(monkeypatch): + telegram = load_telegram(monkeypatch) + telegram._connected = True + sent = [] + telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) + + assert telegram.send_message("OmegaClaw version=test") is True + assert sent == [] + + telegram._process_update( + { + "message": { + "message_id": 42, + "text": "auth secret", + "chat": {"id": 101, "type": "private"}, + "from": {"id": 101, "username": "owner"}, + } + } + ) + + assert [params["text"] for _, params in sent] == [ + "Authentication successful. @owner is now the bot owner. " + "Send /bind in a group to open it to everyone there.", + "OmegaClaw version=test", + ] + assert all(params["chat_id"] == "101" for _, params in sent) + assert json.loads(sent[0][1]["reply_parameters"]) == { + "message_id": 42, + "allow_sending_without_reply": True, + } + assert "reply_parameters" not in sent[1][1] + +def test_missing_route_fields_fall_back_to_owner_without_reply(monkeypatch): + telegram = load_telegram(monkeypatch) + telegram._connected = True + telegram._owner_id = "101" + telegram._default_chat_id = "101" + sent = [] + telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) + + assert telegram.send_message("[] [] proactive message") is True + + assert sent[0][1] == {"chat_id": "101", "text": "proactive message"} + + +def test_generated_target_must_be_owner_or_authorized_group(monkeypatch): + telegram = load_telegram(monkeypatch) + telegram._connected = True + telegram._owner_id = "101" + telegram._default_chat_id = "101" + telegram._authorized_groups = {"-202"} + sent = [] + telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params)) + + assert telegram.send_message("[-202] [] allowed") is True + assert telegram.send_message("[-999] [] rejected") is False + + assert [params["chat_id"] for _, params in sent] == ["-202"] def test_cached_authorization_avoids_disk_reads_per_message(monkeypatch): @@ -174,13 +236,33 @@ def test_invalid_auth_state_stops_startup_before_telegram_is_polled(monkeypatch) telegram.start_telegram() +def test_startup_restores_owner_default_and_bound_groups(monkeypatch): + telegram = load_telegram(monkeypatch) + monkeypatch.setenv("TG_BOT_TOKEN", "token") + telegram.auth.load_channel_auth_state = lambda _channel: ("101", {"-202"}) + telegram._initialize_bot_identity = lambda: True + + class FakeThread: + def start(self): + return None + + telegram.threading.Thread = lambda **_kwargs: FakeThread() + + telegram.start_telegram(allowed_chat_ids="-303") + + assert telegram._default_chat_id == "101" + assert telegram._authorized_groups == {"-202"} + assert telegram._admin_allowed_chats == {"-202", "-303"} + + def test_offset_advances_only_after_update_processing(monkeypatch): telegram = load_telegram(monkeypatch, auth_enabled=False) update = { "update_id": 7, "message": { + "message_id": 41, "text": "hello", - "chat": {"id": "dm", "type": "private"}, + "chat": {"id": 101, "type": "private"}, "from": {"id": "user", "username": "alice"}, }, } @@ -196,7 +278,7 @@ def get_updates(*_args, **_kwargs): telegram._poll_loop() assert telegram._offset == 8 - assert telegram.getLastMessage() == "@alice: hello" + assert telegram.getLastMessage() == "[101] [41] @alice: hello" def test_failed_update_processing_retains_offset(monkeypatch): @@ -227,10 +309,13 @@ def test_owner_binds_group_without_exposing_secret(monkeypatch): assert telegram._is_allowed_message("group", "1", "group", "auth secret") == "ignore" # The owner authenticates once in a private DM. assert telegram._is_allowed_message("dm", "1", "private", "auth secret") == "auth_bound" + assert telegram._default_chat_id == "dm" assert telegram._is_allowed_message("dm", "2", "private", "hello") == "ignore" # Only that owner can open the group; then every group member is allowed. assert telegram._is_allowed_message("group", "2", "group", "/bind") == "ignore" assert telegram._is_allowed_message("group", "1", "group", "/bind@ExampleBot") == "group_bound" + assert "group" in telegram._authorized_groups + assert "group" in telegram._admin_allowed_chats assert telegram._is_allowed_message("group", "2", "group", "hello") == "allow" assert telegram._is_allowed_message("other-group", "2", "group", "hello") == "ignore" @@ -264,6 +349,8 @@ def test_only_owner_can_unbind_group(monkeypatch): assert telegram._is_allowed_message("group", "2", "group", "/unbind") == "ignore" assert telegram._is_allowed_message("group", "2", "group", "hello") == "allow" assert telegram._is_allowed_message("group", "1", "group", "/unbind@ExampleBot") == "group_unbound" + assert "group" not in telegram._authorized_groups + assert "group" not in telegram._admin_allowed_chats assert telegram._is_allowed_message("group", "2", "group", "hello") == "ignore" @@ -273,6 +360,8 @@ def test_owner_can_unbind_group_from_dm(monkeypatch): assert telegram._is_allowed_message("dm", "1", "private", "auth secret") == "auth_bound" assert telegram._is_allowed_message("group", "1", "group", "/bind") == "group_bound" assert telegram._is_allowed_message("dm", "1", "private", "/unbind group") == "group_unbound" + assert "group" not in telegram._authorized_groups + assert "group" not in telegram._admin_allowed_chats assert telegram._is_allowed_message("group", "2", "group", "hello") == "ignore" @@ -294,3 +383,13 @@ def test_allowlist_still_permits_owner_dm_bootstrap(monkeypatch): assert telegram._is_allowed_message("owner-dm", "1", "private", "auth secret") == "auth_bound" assert telegram._is_allowed_message("owner-dm", "1", "private", "hello") == "allow" assert telegram._is_allowed_message("unlisted-group", "2", "group", "hello") == "ignore" + + +def test_owner_bind_can_expand_configured_runtime_allowlist(monkeypatch): + telegram = load_telegram(monkeypatch) + telegram._admin_allowed_chats = {"configured-group"} + + assert telegram._is_allowed_message("owner-dm", "1", "private", "auth secret") == "auth_bound" + assert telegram._is_allowed_message("new-group", "1", "group", "/bind") == "group_bound" + assert "new-group" in telegram._admin_allowed_chats + assert "new-group" in telegram._authorized_groups From 8c9c82c874e9aac86bbba3bd4ab79e64f89a864f Mon Sep 17 00:00:00 2001 From: CodersKin Date: Fri, 28 Aug 2026 12:19:20 +0300 Subject: [PATCH 09/11] Fix: uncommented a few lines in loop --- src/loop.metta | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/loop.metta b/src/loop.metta index a3abfc78..64e4d60d 100644 --- a/src/loop.metta +++ b/src/loop.metta @@ -6,12 +6,12 @@ (= (reasoningMode) (empty)) (= (wakeupInterval) (empty)) (= (memoryDirectory) (empty)) -; (= (spamShield) (empty)) ; TODO: this parameter is considered deprecated +(= (spamShield) (empty)) ; TODO: this parameter is considered deprecated (= (initLoop) (progn (configure maxNewInputLoops 50) ;20 (configure maxWakeLoops 1) - ; (configure spamShield False) + (configure spamShield False) (configure sleepInterval 1) ;10 (configure provider Anthropic) (configure maxOutputToken 6000) @@ -70,9 +70,9 @@ (progn (if (== $k 1) (progn (initConfig) (initLoop) (initLogger) - ; (applySecurityPolicy) + (applySecurityPolicy) (initMemory) - ; (initKnowledge) + (initKnowledge) (initPlugins) (initChannels) (addTelegramPromptExtension) From f2a26badcc7847fe8dd091d9ac24985b1a32a261 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Tue, 1 Sep 2026 18:50:36 +0300 Subject: [PATCH 10/11] Fix: Added safe handling of damaged json entry and hardend the json append to reduce write failures. --- channels/auth.py | 20 +++++++++++--------- tests/test_auth_standalone.py | 23 ++++++++++++++++++----- 2 files changed, 29 insertions(+), 14 deletions(-) diff --git a/channels/auth.py b/channels/auth.py index 7283c148..b76bc455 100644 --- a/channels/auth.py +++ b/channels/auth.py @@ -195,13 +195,17 @@ def read_records(path, label): try: record = json.loads(line) except json.JSONDecodeError as exc: - raise RuntimeError( - f"Malformed {label} record at line {line_number}" - ) from exc + logger.warning( + f"Skipping malformed {label} record at line " + f"{line_number}: {exc}" + ) + continue if not isinstance(record, dict): - raise RuntimeError( - f"Malformed {label} record at line {line_number}" + logger.warning( + f"Skipping malformed {label} record at line " + f"{line_number}: expected a JSON object" ) + continue records.append(record) return records except FileNotFoundError: @@ -262,8 +266,7 @@ def store_channel_authenticated_group_id(channel_identifier, group_id, authorize os.makedirs(os.path.dirname(path), exist_ok=True) try: with open(path, "a", encoding="utf-8") as f: - json.dump(payload, f, separators=(",", ":")) - f.write("\n") + f.write(json.dumps(payload, separators=(",", ":")) + "\n") except OSError as e: raise RuntimeError("Failed to write channel authenticated group record") from e return True @@ -354,8 +357,7 @@ def revoke_channel_group(channel_identifier, group_id, requester_user_id): os.makedirs(os.path.dirname(path), exist_ok=True) try: with open(path, "a", encoding="utf-8") as f: - json.dump(payload, f, separators=(",", ":")) - f.write("\n") + f.write(json.dumps(payload, separators=(",", ":")) + "\n") except OSError as e: raise RuntimeError("Failed to write channel group revocation record") from e diff --git a/tests/test_auth_standalone.py b/tests/test_auth_standalone.py index 7852d89a..7f59b350 100644 --- a/tests/test_auth_standalone.py +++ b/tests/test_auth_standalone.py @@ -168,12 +168,25 @@ def test_load_channel_auth_state_validates_and_loads_records(monkeypatch, tmp_pa assert groups == {"active"} -def test_load_channel_auth_state_rejects_malformed_records(monkeypatch, tmp_path): +@pytest.mark.parametrize("damaged_record", ["not-json\n", '{"time":']) +def test_load_channel_auth_state_skips_malformed_records( + monkeypatch, tmp_path, damaged_record +): auth = load_auth_module(monkeypatch) monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path)) + warnings = [] + monkeypatch.setattr(auth.logger, "warning", warnings.append) + + assert auth.store_channel_authenticated_user_id("TELEGRAM", "owner") is True + assert auth.store_channel_authenticated_group_id( + "TELEGRAM", "active", "owner" + ) is True path = tmp_path / ".channel" / "authenticated-group.json" - path.parent.mkdir() - path.write_text("not-json\n", encoding="utf-8") + with path.open("a", encoding="utf-8") as target: + target.write(damaged_record) - with pytest.raises(RuntimeError, match="Malformed channel authenticated group record"): - auth.load_channel_auth_state("TELEGRAM") + assert auth.load_channel_auth_state("TELEGRAM") == ("owner", {"active"}) + assert any( + "Skipping malformed channel authenticated group record at line 2" in warning + for warning in warnings + ) From 166c6a8c1b448f451ef902161508541b14ee8172 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Wed, 2 Sep 2026 13:39:07 +0300 Subject: [PATCH 11/11] Fix: fixed name mismatch --- channels/telegram.py | 4 ++-- docs/reference-configuration.md | 6 +++--- tests/test_auth_standalone.py | 2 +- tests/test_telegram_multichat.py | 6 ++++++ 4 files changed, 12 insertions(+), 6 deletions(-) diff --git a/channels/telegram.py b/channels/telegram.py index 6a50951c..73876681 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -549,5 +549,5 @@ def send(self, message: str) -> None: send_message(message) -def loadOmegaClawPlugin(): - channels.registerCommChannel("telegram", TelegramChannel()) +def loadOmegaPlugin(): + channels.registerCommChannel("telegram", TelegramChannel()) diff --git a/docs/reference-configuration.md b/docs/reference-configuration.md index 7dfb1b97..2cc28f2c 100644 --- a/docs/reference-configuration.md +++ b/docs/reference-configuration.md @@ -55,7 +55,7 @@ This reads a command-line override via `argk` (`name=value` on the MeTTa command | Environment variable | Meaning | |---|---| | `TG_BOT_TOKEN` | Telegram bot token (from BotFather). | -| `OMEGACLAW_AUTH_SECRET` | Enables the one-time owner-authentication handshake when non-empty. The owner sends `auth ` in a private Telegram DM. | +| `OMEGA_AUTH_SECRET` | Enables the one-time owner-authentication handshake when non-empty. The owner sends `auth ` in a private Telegram DM. | | `MM_BOT_TOKEN` | Bot auth token. | | `SL_BOT_TOKEN` | Slack bot token (`xoxb-...`). | @@ -69,13 +69,13 @@ metta run.metta provider=Anthropic LLM=claude-opus-4-6 commchannel=mattermost Configuration values are resolved in this order: command-line `key=value`, `OMEGACLAW_` environment variable, `config/config.yaml`, then the caller's -default. `TG_BOT_TOKEN` and `OMEGACLAW_AUTH_SECRET` are read directly from the +default. `TG_BOT_TOKEN` and `OMEGA_AUTH_SECRET` are read directly from the environment and must be placed before the `metta`/`petta` command. Telegram example: ```bash -TG_BOT_TOKEN=... OMEGACLAW_AUTH_SECRET=... \ +TG_BOT_TOKEN=... OMEGA_AUTH_SECRET=... \ metta run.metta commchannel=telegram ``` diff --git a/tests/test_auth_standalone.py b/tests/test_auth_standalone.py index 7a5adbe2..011687bf 100644 --- a/tests/test_auth_standalone.py +++ b/tests/test_auth_standalone.py @@ -88,7 +88,7 @@ def test_saved_owner_blocks_auth_secret_reuse_after_restart(monkeypatch, tmp_pat def test_saved_owner_cannot_be_replaced_by_a_reused_secret(monkeypatch, tmp_path): - monkeypatch.setenv("OMEGACLAW_AUTH_SECRET", "one-time-secret") + monkeypatch.setenv("OMEGA_AUTH_SECRET", "one-time-secret") first_process = load_auth_module(monkeypatch) monkeypatch.setattr(first_process, "_MEMORY_DIRECTORY", str(tmp_path)) diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py index 1865f888..37ca3376 100644 --- a/tests/test_telegram_multichat.py +++ b/tests/test_telegram_multichat.py @@ -66,6 +66,12 @@ def revoke_channel_group(channel, group, requester): return module +def test_exposes_omega_plugin_entrypoint(monkeypatch): + telegram = load_telegram(monkeypatch) + + assert callable(telegram.loadOmegaPlugin) + + def test_reply_uses_the_chat_that_supplied_the_message(monkeypatch): telegram = load_telegram(monkeypatch, auth_enabled=False) telegram._connected = True