mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-21 10:01:37 -06:00
Ayaan Zaidi
b7b7eec5ee
Move Mantis Telegram Desktop proof from the remote AWS/Crabbox lane to a recorder-driven local Docker desktop. Keep proof scenarios agent-authored, cache trusted build outputs, and publish exact visible Telegram evidence without writing the QA bot token to artifacts. Co-authored-by: Ayaan Zaidi <hi@obviy.us>
1444 lines
55 KiB
Python
Executable File
1444 lines
55 KiB
Python
Executable File
#!/usr/bin/env -S uv run --script
|
|
|
|
import argparse
|
|
import atexit
|
|
import base64
|
|
import ctypes
|
|
import json
|
|
import os
|
|
import secrets
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
SKILL_DIR = Path(__file__).resolve().parents[1]
|
|
STATE_DIR = Path(os.environ.get("TELEGRAM_USER_DRIVER_STATE_DIR") or (SKILL_DIR / "user-driver")).expanduser()
|
|
CONFIG_PATH = STATE_DIR / "config.local.json"
|
|
BOT_CREDENTIALS_PATH = SKILL_DIR / "credentials.local.json"
|
|
|
|
|
|
class DriverError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def read_json(path):
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except FileNotFoundError:
|
|
return {}
|
|
|
|
|
|
def write_json_private(path, data):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.parent.chmod(stat.S_IRWXU)
|
|
path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n")
|
|
path.chmod(stat.S_IRUSR | stat.S_IWUSR)
|
|
|
|
|
|
def env_or_config(env_name, config, key, default=""):
|
|
value = os.environ.get(env_name)
|
|
if value:
|
|
return value
|
|
value = config.get(key)
|
|
if value is None:
|
|
return default
|
|
return str(value)
|
|
|
|
|
|
def load_config():
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
STATE_DIR.chmod(stat.S_IRWXU)
|
|
config = read_json(CONFIG_PATH)
|
|
bot_config = read_json(BOT_CREDENTIALS_PATH)
|
|
if not valid_base64(config.get("databaseEncryptionKey", "")):
|
|
config["databaseEncryptionKey"] = base64.b64encode(secrets.token_bytes(32)).decode()
|
|
write_json_private(CONFIG_PATH, config)
|
|
return config, bot_config
|
|
|
|
|
|
def valid_base64(value):
|
|
if not isinstance(value, str) or not value:
|
|
return False
|
|
try:
|
|
base64.b64decode(value.encode(), validate=True)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def find_tdjson(config):
|
|
explicit = env_or_config("TELEGRAM_USER_DRIVER_TDLIB_PATH", config, "tdlibPath")
|
|
candidates = []
|
|
if explicit:
|
|
candidates.append(Path(explicit).expanduser())
|
|
candidates.extend(
|
|
Path(path)
|
|
for path in [
|
|
"/opt/homebrew/lib/libtdjson.dylib",
|
|
"/opt/homebrew/opt/tdlib/lib/libtdjson.dylib",
|
|
"/usr/local/lib/libtdjson.dylib",
|
|
"/usr/local/opt/tdlib/lib/libtdjson.dylib",
|
|
"/usr/lib/libtdjson.so",
|
|
"/usr/local/lib/libtdjson.so",
|
|
]
|
|
)
|
|
for path in candidates:
|
|
if path.exists():
|
|
return path
|
|
found = shutil.which("tdjson")
|
|
if found:
|
|
return Path(found)
|
|
return None
|
|
|
|
|
|
def telegram_bot(token, method, payload=None):
|
|
data = json.dumps(payload or {}).encode()
|
|
request = urllib.request.Request(
|
|
f"https://api.telegram.org/bot{token}/{method}",
|
|
data=data,
|
|
headers={"content-type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=15) as response:
|
|
body = json.loads(response.read().decode())
|
|
if not body.get("ok"):
|
|
raise DriverError(body.get("description") or f"{method} failed")
|
|
return body["result"]
|
|
|
|
|
|
def resolve_sut(config, bot_config):
|
|
username = env_or_config("TELEGRAM_USER_DRIVER_SUT_USERNAME", config, "sutUsername")
|
|
user_id = env_or_config("TELEGRAM_USER_DRIVER_SUT_ID", config, "sutId")
|
|
if username and user_id:
|
|
return {"username": username.lstrip("@"), "id": int(user_id)}
|
|
token = (
|
|
os.environ.get("TELEGRAM_E2E_SUT_BOT_TOKEN")
|
|
or os.environ.get("OPENCLAW_QA_TELEGRAM_SUT_BOT_TOKEN")
|
|
or bot_config.get("sutBotToken")
|
|
or bot_config.get("botAToken")
|
|
or bot_config.get("BOTA")
|
|
)
|
|
if token:
|
|
me = telegram_bot(token, "getMe")
|
|
return {"username": me.get("username", "").lstrip("@"), "id": int(me["id"])}
|
|
if username:
|
|
return {"username": username.lstrip("@"), "id": None}
|
|
return {"username": "", "id": None}
|
|
|
|
|
|
def default_chat(config, bot_config):
|
|
return (
|
|
os.environ.get("TELEGRAM_USER_DRIVER_CHAT_ID")
|
|
or os.environ.get("TELEGRAM_E2E_GROUP_ID")
|
|
or os.environ.get("OPENCLAW_QA_TELEGRAM_GROUP_ID")
|
|
or str(config.get("defaultChatId") or "")
|
|
or str(bot_config.get("groupId") or "")
|
|
)
|
|
|
|
|
|
class TdClient:
|
|
def __init__(self, config):
|
|
lib_path = find_tdjson(config)
|
|
if not lib_path:
|
|
raise DriverError("Missing libtdjson. Install with: brew install tdlib")
|
|
self.lib_path = lib_path
|
|
self.lib = ctypes.CDLL(str(lib_path))
|
|
self.lib.td_execute.argtypes = [ctypes.c_char_p]
|
|
self.lib.td_execute.restype = ctypes.c_char_p
|
|
self.lib.td_json_client_create.restype = ctypes.c_void_p
|
|
self.lib.td_json_client_send.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
|
|
self.lib.td_json_client_receive.argtypes = [ctypes.c_void_p, ctypes.c_double]
|
|
self.lib.td_json_client_receive.restype = ctypes.c_char_p
|
|
self.lib.td_json_client_execute.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
|
|
self.lib.td_json_client_execute.restype = ctypes.c_char_p
|
|
self.lib.td_json_client_destroy.argtypes = [ctypes.c_void_p]
|
|
self.lib.td_execute(json.dumps({"@type": "setLogStream", "log_stream": {"@type": "logStreamEmpty"}}).encode())
|
|
self.lib.td_execute(json.dumps({"@type": "setLogVerbosityLevel", "new_verbosity_level": 0}).encode())
|
|
self.client = self.lib.td_json_client_create()
|
|
self.extra = 0
|
|
self.pending = {}
|
|
self.users = {}
|
|
self.updates = []
|
|
atexit.register(self.destroy)
|
|
|
|
def execute(self, payload):
|
|
raw = self.lib.td_json_client_execute(self.client, json.dumps(payload).encode())
|
|
return json.loads(raw.decode()) if raw else None
|
|
|
|
def send(self, payload):
|
|
self.extra += 1
|
|
extra = str(self.extra)
|
|
payload = dict(payload)
|
|
payload["@extra"] = extra
|
|
self.pending[extra] = payload["@type"]
|
|
self.lib.td_json_client_send(self.client, json.dumps(payload).encode())
|
|
return extra
|
|
|
|
def receive(self, timeout=1.0):
|
|
raw = self.lib.td_json_client_receive(self.client, ctypes.c_double(timeout))
|
|
if not raw:
|
|
return None
|
|
item = json.loads(raw.decode())
|
|
if item.get("@type") == "updateUser":
|
|
user = item.get("user") or {}
|
|
self.users[int(user["id"])] = user
|
|
return item
|
|
|
|
def destroy(self):
|
|
if not self.client:
|
|
return
|
|
client = self.client
|
|
self.client = None
|
|
self.lib.td_json_client_destroy(client)
|
|
|
|
def request(self, payload, timeout=20):
|
|
extra = self.send(payload)
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
item = self.receive(1)
|
|
if not item:
|
|
continue
|
|
if item.get("@extra") == extra:
|
|
if item.get("@type") == "error":
|
|
raise DriverError(f"{payload['@type']} failed: {item.get('message')}")
|
|
return item
|
|
self.handle_update(item)
|
|
raise DriverError(f"Timed out waiting for {payload['@type']}")
|
|
|
|
def handle_update(self, item):
|
|
self.updates.append(item)
|
|
return item
|
|
|
|
def next_update(self, timeout=1.0):
|
|
if self.updates:
|
|
return self.updates.pop(0)
|
|
return self.receive(timeout)
|
|
|
|
|
|
class UserDriver:
|
|
def __init__(self, config, bot_config):
|
|
self.config = config
|
|
self.bot_config = bot_config
|
|
self.client = TdClient(config)
|
|
self.auth_state = None
|
|
self.printed_qr_link = ""
|
|
|
|
def td_params(self):
|
|
api_id = env_or_config("TELEGRAM_USER_DRIVER_API_ID", self.config, "apiId")
|
|
api_hash = env_or_config("TELEGRAM_USER_DRIVER_API_HASH", self.config, "apiHash")
|
|
if not api_id or not api_hash:
|
|
raise DriverError("Missing Telegram API app credentials. Run configure or set TELEGRAM_USER_DRIVER_API_ID/API_HASH.")
|
|
params = {
|
|
"use_test_dc": False,
|
|
"database_directory": str(STATE_DIR / "db"),
|
|
"files_directory": str(STATE_DIR / "files"),
|
|
"use_file_database": True,
|
|
"use_chat_info_database": True,
|
|
"use_message_database": True,
|
|
"use_secret_chats": False,
|
|
"api_id": int(api_id),
|
|
"api_hash": api_hash,
|
|
"system_language_code": "en",
|
|
"device_model": "OpenClaw Telegram User Driver",
|
|
"system_version": sys.platform,
|
|
"application_version": "1",
|
|
"enable_storage_optimizer": True,
|
|
"ignore_file_names": False,
|
|
}
|
|
return {
|
|
"@type": "setTdlibParameters",
|
|
"parameters": {"@type": "tdlibParameters", **params},
|
|
}
|
|
|
|
def td_params_current(self):
|
|
payload = self.td_params()
|
|
params = dict(payload["parameters"])
|
|
params.pop("@type", None)
|
|
return {"@type": "setTdlibParameters", **params}
|
|
|
|
def encryption_key(self):
|
|
return env_or_config(
|
|
"TELEGRAM_USER_DRIVER_DB_ENCRYPTION_KEY",
|
|
self.config,
|
|
"databaseEncryptionKey",
|
|
)
|
|
|
|
def encryption_key_for_current_tdlib(self):
|
|
return base64.b64encode(self.encryption_key().encode()).decode()
|
|
|
|
def authorize(self, args=None, need_ready=True):
|
|
args = args or argparse.Namespace()
|
|
self.client.execute({"@type": "setLogStream", "log_stream": {"@type": "logStreamEmpty"}})
|
|
self.client.execute({"@type": "setLogVerbosityLevel", "new_verbosity_level": 1})
|
|
self.client.send({"@type": "getOption", "name": "version"})
|
|
retried_current_params = False
|
|
retried_current_encryption_key = False
|
|
deadline = time.time() + getattr(args, "timeout_ms", 120000) / 1000
|
|
while time.time() < deadline:
|
|
item = self.client.receive(1)
|
|
if not item:
|
|
continue
|
|
if item.get("@type") == "updateAuthorizationState":
|
|
self.auth_state = item["authorization_state"]
|
|
state = self.auth_state["@type"]
|
|
if state == "authorizationStateWaitTdlibParameters":
|
|
self.client.send(self.td_params())
|
|
elif state == "authorizationStateWaitEncryptionKey":
|
|
self.client.send(
|
|
{
|
|
"@type": "checkDatabaseEncryptionKey",
|
|
"encryption_key": self.encryption_key(),
|
|
}
|
|
)
|
|
elif state == "authorizationStateWaitPhoneNumber":
|
|
if getattr(args, "phone", ""):
|
|
self.client.send(
|
|
{
|
|
"@type": "setAuthenticationPhoneNumber",
|
|
"phone_number": args.phone,
|
|
"settings": None,
|
|
}
|
|
)
|
|
elif getattr(args, "qr", False):
|
|
self.client.send(
|
|
{
|
|
"@type": "requestQrCodeAuthentication",
|
|
"other_user_ids": [],
|
|
}
|
|
)
|
|
elif need_ready:
|
|
raise DriverError("Not logged in. Run: user-driver.py login --qr")
|
|
else:
|
|
return False
|
|
elif state == "authorizationStateWaitOtherDeviceConfirmation":
|
|
self.show_qr_link(self.auth_state["link"])
|
|
elif state == "authorizationStateWaitCode":
|
|
code = getattr(args, "code", "") or prompt_secret("Telegram login code: ")
|
|
self.client.send({"@type": "checkAuthenticationCode", "code": code})
|
|
elif state == "authorizationStateWaitPassword":
|
|
password = getattr(args, "password", "") or prompt_secret("Telegram 2FA password: ")
|
|
self.client.send({"@type": "checkAuthenticationPassword", "password": password})
|
|
elif state == "authorizationStateReady":
|
|
return True
|
|
elif state in {
|
|
"authorizationStateClosing",
|
|
"authorizationStateClosed",
|
|
"authorizationStateLoggingOut",
|
|
}:
|
|
raise DriverError(f"TDLib auth state is {state}")
|
|
elif item.get("@type") == "error":
|
|
message = item.get("message") or "TDLib error"
|
|
if not retried_current_params and "Valid api_id must be provided" in message:
|
|
retried_current_params = True
|
|
self.client.send(
|
|
{
|
|
"@type": "setDatabaseEncryptionKey",
|
|
"new_encryption_key": self.encryption_key_for_current_tdlib(),
|
|
}
|
|
)
|
|
self.client.send(self.td_params_current())
|
|
continue
|
|
if not retried_current_encryption_key and "Wrong database encryption key" in message:
|
|
retried_current_encryption_key = True
|
|
self.client.send(
|
|
{
|
|
"@type": "setDatabaseEncryptionKey",
|
|
"new_encryption_key": self.encryption_key_for_current_tdlib(),
|
|
}
|
|
)
|
|
self.client.send(self.td_params_current())
|
|
continue
|
|
if retried_current_params and "Initialization parameters are needed" in message:
|
|
continue
|
|
raise DriverError(message)
|
|
raise DriverError("Timed out waiting for Telegram authorization")
|
|
|
|
def show_qr_link(self, link):
|
|
if link == self.printed_qr_link:
|
|
return
|
|
self.printed_qr_link = link
|
|
print("\nScan this with Telegram > Settings > Devices > Link Desktop Device\n")
|
|
qrencode = shutil.which("qrencode")
|
|
if qrencode:
|
|
subprocess.run([qrencode, "-t", "UTF8", link], check=False)
|
|
print(link)
|
|
print()
|
|
|
|
def resolve_chat(self, chat):
|
|
chat = chat or default_chat(self.config, self.bot_config)
|
|
if not chat:
|
|
raise DriverError(
|
|
"Missing chat. Pass --chat or configure defaultChatId. Run `user-driver.py chats --json` to list chats visible to the tester account."
|
|
)
|
|
if chat.startswith(("https://t.me/+", "tg://join")) or "joinchat" in chat:
|
|
return self.client.request({"@type": "joinChatByInviteLink", "invite_link": chat})["id"]
|
|
if chat.startswith("@"):
|
|
return self.client.request({"@type": "searchPublicChat", "username": chat[1:]})["id"]
|
|
if chat.startswith("https://t.me/") and "/" not in chat.removeprefix("https://t.me/"):
|
|
return self.client.request(
|
|
{
|
|
"@type": "searchPublicChat",
|
|
"username": chat.removeprefix("https://t.me/"),
|
|
}
|
|
)["id"]
|
|
try:
|
|
return self.client.request({"@type": "getChat", "chat_id": int(chat)}, timeout=10)["id"]
|
|
except DriverError as error:
|
|
raise DriverError(
|
|
f"Chat not found for tester account: {chat}. Add the QA user to the group, or configure the TDLib chat id from `user-driver.py chats --json`."
|
|
) from error
|
|
|
|
def text_content(self, text):
|
|
return {
|
|
"@type": "inputMessageText",
|
|
"text": {"@type": "formattedText", "text": text, "entities": []},
|
|
"disable_web_page_preview": False,
|
|
"clear_draft": True,
|
|
}
|
|
|
|
def document_content(self, file_path, caption):
|
|
return {
|
|
"@type": "inputMessageDocument",
|
|
"document": {"@type": "inputFileLocal", "path": file_path},
|
|
"thumbnail": None,
|
|
"disable_content_type_detection": False,
|
|
"caption": {
|
|
"@type": "formattedText",
|
|
"text": caption or "",
|
|
"entities": [],
|
|
},
|
|
}
|
|
|
|
def send_text(self, chat_id, text, reply_to=None, thread_id=0, file_path=None):
|
|
content = self.document_content(file_path, text) if file_path else self.text_content(text)
|
|
return self.settle_sent_message(
|
|
self.client.request(
|
|
{
|
|
"@type": "sendMessage",
|
|
"chat_id": chat_id,
|
|
"message_thread_id": int(thread_id or 0),
|
|
"reply_to_message_id": int(reply_to or 0),
|
|
"options": {
|
|
"@type": "messageSendOptions",
|
|
"disable_notification": True,
|
|
"from_background": False,
|
|
"scheduling_state": None,
|
|
},
|
|
"reply_markup": None,
|
|
"input_message_content": content,
|
|
},
|
|
timeout=60,
|
|
)
|
|
)
|
|
|
|
def settle_sent_message(self, message, timeout=30):
|
|
if not message.get("sending_state"):
|
|
return message
|
|
pending_id = message["id"]
|
|
deadline = time.time() + timeout
|
|
deferred = []
|
|
while time.time() < deadline:
|
|
item = self.client.next_update(1)
|
|
if not item:
|
|
continue
|
|
if item.get("@type") == "updateMessageSendSucceeded" and item.get("old_message_id") == pending_id:
|
|
self.client.updates = deferred + self.client.updates
|
|
return item["message"]
|
|
if item.get("@type") == "updateMessageSendFailed" and item.get("old_message_id") == pending_id:
|
|
self.client.updates = deferred + self.client.updates
|
|
raise DriverError(item.get("error_message") or "Telegram message send failed")
|
|
deferred.append(item)
|
|
self.client.updates = deferred + self.client.updates
|
|
raise DriverError("Timed out waiting for Telegram message send confirmation")
|
|
|
|
def wait_for_message(self, chat_id, args, after_message_id=0):
|
|
sut = resolve_sut(self.config, self.bot_config)
|
|
expect = args.expect or []
|
|
from_bot = (args.from_bot or sut["username"] or "").lstrip("@")
|
|
from_bot_id = int(from_bot) if from_bot.isdigit() else sut["id"]
|
|
from_bot_username = "" if from_bot.isdigit() else from_bot
|
|
deadline = time.time() + args.timeout_ms / 1000
|
|
observed = []
|
|
while time.time() < deadline:
|
|
item = self.client.next_update(1)
|
|
if not item or item.get("@type") != "updateNewMessage":
|
|
continue
|
|
message = item["message"]
|
|
if int(message.get("chat_id", 0)) != int(chat_id):
|
|
continue
|
|
normalized = normalize_message(message, self.client.users)
|
|
observed.append(normalized)
|
|
if int(normalized["messageId"]) <= int(after_message_id):
|
|
continue
|
|
if getattr(args, "thread_id", None) and str(normalized.get("threadId") or "") != str(args.thread_id):
|
|
continue
|
|
if from_bot_id and int(normalized.get("senderId") or 0) != int(from_bot_id):
|
|
continue
|
|
if not from_bot_id and from_bot_username and normalized["senderUsername"].lower() != from_bot_username.lower():
|
|
continue
|
|
if args.reply_to and str(normalized.get("replyToMessageId") or "") != str(args.reply_to):
|
|
continue
|
|
if any(entry not in normalized["text"] for entry in expect):
|
|
continue
|
|
return normalized, observed
|
|
return None, observed
|
|
|
|
|
|
def prompt_secret(label):
|
|
if not sys.stdin.isatty():
|
|
raise DriverError(f"{label.strip()} required; rerun with explicit flag in an interactive terminal.")
|
|
import getpass
|
|
|
|
return getpass.getpass(label)
|
|
|
|
|
|
def normalize_message(message, users=None):
|
|
users = users or {}
|
|
sender = message.get("sender_id") or {}
|
|
sender_id = sender.get("user_id") or sender.get("chat_id")
|
|
sender_user = users.get(int(sender_id or 0), {}) if sender_id else {}
|
|
content = message.get("content") or {}
|
|
text = ""
|
|
if content.get("@type") == "messageText":
|
|
text = (content.get("text") or {}).get("text", "")
|
|
elif "caption" in content:
|
|
text = (content.get("caption") or {}).get("text", "")
|
|
reply_to_message_id = message.get("reply_to_message_id") or (message.get("reply_to") or {}).get("message_id")
|
|
message_id = message.get("id")
|
|
return {
|
|
"messageId": message_id,
|
|
# TDLib reserves the low 20 bits; Telegram private-post links use the
|
|
# server/Bot API id. Keep both identities at the boundary that owns them.
|
|
"botApiMessageId": (int(message_id) >> 20) if message_id else None,
|
|
"chatId": message.get("chat_id"),
|
|
"senderId": sender_id,
|
|
"senderUsername": sender_user.get("username") or "",
|
|
"date": message.get("date"),
|
|
"replyToMessageId": reply_to_message_id,
|
|
"replyToBotApiMessageId": (int(reply_to_message_id) >> 20) if reply_to_message_id else None,
|
|
"threadId": message.get("message_thread_id"),
|
|
"text": text,
|
|
"contentType": content.get("@type"),
|
|
"raw": message,
|
|
}
|
|
|
|
|
|
def rich_text(value):
|
|
if not isinstance(value, dict):
|
|
return ""
|
|
kind = value.get("@type", "")
|
|
if kind == "richTextPlain":
|
|
return value.get("text", "")
|
|
if kind == "richTextCustomEmoji":
|
|
return value.get("alternative_text", "")
|
|
if kind == "richTextMathematicalExpression":
|
|
return value.get("expression", "")
|
|
if kind == "richTexts":
|
|
return "".join(rich_text(item) for item in value.get("texts") or [])
|
|
return rich_text(value.get("text"))
|
|
|
|
|
|
def rich_message_text(value):
|
|
if not isinstance(value, dict):
|
|
return ""
|
|
parts = []
|
|
|
|
def visit(node):
|
|
if isinstance(node, list):
|
|
for item in node:
|
|
visit(item)
|
|
elif isinstance(node, dict):
|
|
if str(node.get("@type", "")).startswith("richText"):
|
|
text = rich_text(node)
|
|
if text:
|
|
parts.append(text)
|
|
else:
|
|
for child in node.values():
|
|
visit(child)
|
|
|
|
visit(value.get("blocks") or [])
|
|
return "\n".join(parts)
|
|
|
|
|
|
def content_text(content):
|
|
for key in ("text", "caption"):
|
|
value = content.get(key)
|
|
if isinstance(value, dict) and isinstance(value.get("text"), str):
|
|
return value["text"]
|
|
if content.get("@type") == "messageRichMessage":
|
|
return rich_message_text(content.get("message") or {})
|
|
return ""
|
|
|
|
|
|
def server_message_id(tdlib_message_id):
|
|
return str(int(tdlib_message_id) >> 20) if tdlib_message_id else None
|
|
|
|
|
|
class UserObserver:
|
|
MAX_EVENTS = 500
|
|
|
|
def __init__(self, user_driver, chat_id, sut_user_id, sut_username, journal_path, media_root):
|
|
self.driver = user_driver
|
|
self.client = user_driver.client
|
|
self.chat_id = int(chat_id)
|
|
self.sut_user_id = int(sut_user_id)
|
|
self.sut = {"id": self.sut_user_id, "username": sut_username.lstrip("@")}
|
|
self.media_root = Path(media_root).resolve()
|
|
self.media_staging = Path(journal_path).parent / "media"
|
|
self.media_staging.mkdir(mode=0o700)
|
|
self.started_at = time.monotonic()
|
|
self.events = []
|
|
self.truncated = False
|
|
self.message_fields = {}
|
|
self.message_ids = {}
|
|
self.recorded_message_ids = set()
|
|
self.buttons = {}
|
|
self.sent_message_ids = set()
|
|
# One line-buffered handle keeps the live journal readable without reopening per event.
|
|
self.journal = Path(journal_path).open("w", buffering=1) # noqa: SIM115
|
|
|
|
def close(self):
|
|
self.journal.close()
|
|
shutil.rmtree(self.media_staging)
|
|
|
|
def actor(self, sender_id, is_outgoing=False):
|
|
if is_outgoing:
|
|
return "user"
|
|
if sender_id == self.sut_user_id:
|
|
return "bot"
|
|
return "other"
|
|
|
|
def append(self, kind, message_id=None, **fields):
|
|
if len(self.events) >= self.MAX_EVENTS:
|
|
self.truncated = True
|
|
return None
|
|
event = {
|
|
"seq": len(self.events) + 1,
|
|
"elapsedMs": int((time.monotonic() - self.started_at) * 1000),
|
|
"kind": kind,
|
|
"messageId": server_message_id(message_id),
|
|
**fields,
|
|
}
|
|
self.events.append(event)
|
|
self.journal.write(json.dumps(event, separators=(",", ":")) + "\n")
|
|
return event
|
|
|
|
def sender_id(self, message):
|
|
sender = message.get("sender_id") or {}
|
|
return sender.get("user_id") or sender.get("chat_id")
|
|
|
|
def reply_fields(self, message):
|
|
reply = message.get("reply_to") or {}
|
|
return {"replyToMessageId": server_message_id(reply.get("message_id") or message.get("reply_to_message_id"))}
|
|
|
|
def remember_buttons(self, message_id, markup):
|
|
flattened = []
|
|
public_buttons = []
|
|
for row in markup.get("rows") or markup.get("inline_keyboard") or []:
|
|
row_buttons = row.get("buttons") if isinstance(row, dict) else row
|
|
for button in row_buttons or []:
|
|
index = len(flattened)
|
|
button_type = button.get("type") or {}
|
|
flattened.append(button_type)
|
|
public_buttons.append(
|
|
{
|
|
"index": index,
|
|
"text": button.get("text", ""),
|
|
"type": str(button_type.get("@type", "")).removeprefix("inlineKeyboardButtonType"),
|
|
}
|
|
)
|
|
if flattened:
|
|
self.buttons[message_id] = flattened
|
|
else:
|
|
self.buttons.pop(message_id, None)
|
|
return public_buttons
|
|
|
|
def remember_message(self, message):
|
|
message_id = message.get("id")
|
|
if not isinstance(message_id, int):
|
|
return {}
|
|
sender_id = self.sender_id(message)
|
|
fields = {
|
|
"actor": self.actor(sender_id, bool(message.get("is_outgoing"))),
|
|
"isOutgoing": bool(message.get("is_outgoing")),
|
|
**self.reply_fields(message),
|
|
}
|
|
self.message_fields[message_id] = fields
|
|
public_buttons = []
|
|
if fields["actor"] != "other":
|
|
self.message_ids[server_message_id(message_id)] = message_id
|
|
if fields["isOutgoing"]:
|
|
self.sent_message_ids.add(message_id)
|
|
public_buttons = self.remember_buttons(message_id, message.get("reply_markup") or {})
|
|
return {**fields, "buttons": public_buttons}
|
|
|
|
def ingest(self, update):
|
|
message = update.get("message") or {}
|
|
update_chat_id = message.get("chat_id") if message else update.get("chat_id")
|
|
if int(update_chat_id or 0) != self.chat_id:
|
|
return
|
|
kind = update.get("@type")
|
|
if kind == "updateNewMessage":
|
|
if message.get("sending_state"):
|
|
return
|
|
if message.get("id") in self.recorded_message_ids:
|
|
return
|
|
content = message.get("content") or {}
|
|
self.recorded_message_ids.add(message.get("id"))
|
|
fields = self.remember_message(message)
|
|
if fields.get("actor") == "other":
|
|
return
|
|
self.append(
|
|
"message",
|
|
message.get("id"),
|
|
**fields,
|
|
contentType=content.get("@type", ""),
|
|
text=content_text(content),
|
|
)
|
|
elif kind == "updateMessageContent":
|
|
message_id = update.get("message_id")
|
|
content = update.get("new_content") or {}
|
|
fields = self.message_fields.get(message_id)
|
|
if not fields or fields.get("actor") == "other":
|
|
return
|
|
self.append(
|
|
"edit",
|
|
message_id,
|
|
**fields,
|
|
contentType=content.get("@type", ""),
|
|
text=content_text(content),
|
|
)
|
|
elif kind == "updateMessageEdited":
|
|
message_id = update.get("message_id")
|
|
fields = self.message_fields.get(message_id)
|
|
if not fields or fields.get("actor") == "other":
|
|
return
|
|
buttons = self.remember_buttons(message_id, update.get("reply_markup") or {})
|
|
self.append(
|
|
"edit-meta",
|
|
message_id,
|
|
**fields,
|
|
buttons=buttons,
|
|
editDate=update.get("edit_date"),
|
|
)
|
|
elif kind == "updateDeleteMessages" and not update.get("from_cache"):
|
|
for message_id in update.get("message_ids") or []:
|
|
fields = self.message_fields.get(message_id)
|
|
if not fields or fields.get("actor") == "other":
|
|
continue
|
|
self.append(
|
|
"delete",
|
|
message_id,
|
|
**fields,
|
|
isPermanent=bool(update.get("is_permanent")),
|
|
)
|
|
elif kind == "updateChatAction":
|
|
sender_id = self.sender_id({"sender_id": update.get("sender_id") or {}})
|
|
actor = self.actor(sender_id)
|
|
if actor == "other":
|
|
return
|
|
self.append(
|
|
"typing",
|
|
actor=actor,
|
|
action=str((update.get("action") or {}).get("@type", "")).removeprefix("chatAction"),
|
|
)
|
|
|
|
def pump(self, seconds):
|
|
deadline = time.monotonic() + seconds
|
|
while time.monotonic() < deadline:
|
|
update = self.client.next_update(timeout=min(0.2, max(0.0, deadline - time.monotonic())))
|
|
if update:
|
|
self.ingest(update)
|
|
|
|
def resolve_message_id(self, value):
|
|
message_id = self.message_ids.get(str(value))
|
|
if message_id is None:
|
|
raise DriverError(f"Message {value} was not observed in this session.")
|
|
return message_id
|
|
|
|
def resolve_media(self, value):
|
|
relative = Path(value)
|
|
if relative.is_absolute() or ".." in relative.parts:
|
|
raise DriverError("Media must be inside the Mantis output directory.")
|
|
media = self.media_root / relative
|
|
try:
|
|
descriptor = os.open(media, os.O_RDONLY | os.O_NOFOLLOW)
|
|
except OSError as error:
|
|
raise DriverError("Media must be a regular file inside the Mantis output directory.") from error
|
|
try:
|
|
opened = Path(os.path.realpath(f"/proc/self/fd/{descriptor}"))
|
|
try:
|
|
opened.relative_to(self.media_root)
|
|
except ValueError as error:
|
|
raise DriverError("Media must be inside the Mantis output directory.") from error
|
|
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
|
|
raise DriverError("Media must be a regular file no larger than 20 MiB.")
|
|
staging_dir = self.media_staging / f"upload-{secrets.token_hex(8)}"
|
|
staging_dir.mkdir(mode=0o700)
|
|
target = staging_dir / opened.name
|
|
target_descriptor = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
try:
|
|
copied = 0
|
|
with os.fdopen(os.dup(descriptor), "rb") as source, os.fdopen(target_descriptor, "wb") as destination:
|
|
while chunk := source.read(1024 * 1024):
|
|
copied += len(chunk)
|
|
if copied > 20 * 1024 * 1024:
|
|
raise DriverError("Media must be a regular file no larger than 20 MiB.")
|
|
destination.write(chunk)
|
|
except Exception:
|
|
shutil.rmtree(staging_dir)
|
|
raise
|
|
return str(target)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
def call(self, request):
|
|
command = request.get("command")
|
|
if command == "ping":
|
|
return {"ok": True, "cursor": len(self.events)}
|
|
if command == "events":
|
|
seconds = float(request.get("seconds") or 0)
|
|
if seconds < 0 or seconds > 60:
|
|
raise DriverError("Observation must be between 0 and 60 seconds.")
|
|
self.pump(seconds)
|
|
since = int(request.get("since") or 0)
|
|
if since < 0 or since > len(self.events):
|
|
raise DriverError("Observation cursor is outside this session's timeline.")
|
|
return {
|
|
"ok": True,
|
|
"cursor": len(self.events),
|
|
"events": self.events[since:],
|
|
}
|
|
if command == "send":
|
|
text, _ = apply_template(str(request.get("text") or ""), self.sut)
|
|
media = request.get("media")
|
|
if not text and not media:
|
|
raise DriverError("A message needs text or media.")
|
|
if len(text) > 4000:
|
|
raise DriverError("Message text exceeds 4000 characters.")
|
|
reply_to = request.get("replyTo")
|
|
reply_id = self.resolve_message_id(reply_to) if reply_to else None
|
|
staged_media = self.resolve_media(str(media)) if media else None
|
|
event_start = len(self.events)
|
|
try:
|
|
sent = self.driver.send_text(
|
|
self.chat_id,
|
|
text,
|
|
reply_to=reply_id,
|
|
file_path=staged_media,
|
|
)
|
|
finally:
|
|
if staged_media:
|
|
shutil.rmtree(Path(staged_media).parent)
|
|
fields = self.remember_message(sent)
|
|
self.recorded_message_ids.add(sent.get("id"))
|
|
sent_event = self.append(
|
|
"message",
|
|
sent.get("id"),
|
|
**fields,
|
|
contentType=(sent.get("content") or {}).get("@type", ""),
|
|
text=content_text(sent.get("content") or {}),
|
|
)
|
|
self.pump(0.2)
|
|
return {
|
|
"ok": True,
|
|
"cursor": len(self.events),
|
|
"events": self.events[event_start:],
|
|
"sent": sent_event,
|
|
}
|
|
if command == "delete":
|
|
message_id = self.resolve_message_id(request.get("messageId"))
|
|
if message_id not in self.sent_message_ids:
|
|
raise DriverError("Only user messages sent in this session can be deleted.")
|
|
self.client.request(
|
|
{
|
|
"@type": "deleteMessages",
|
|
"chat_id": self.chat_id,
|
|
"message_ids": [message_id],
|
|
"revoke": True,
|
|
}
|
|
)
|
|
self.pump(0.2)
|
|
return {"ok": True, "cursor": len(self.events)}
|
|
if command == "press":
|
|
message_id = self.resolve_message_id(request.get("messageId"))
|
|
index = int(request.get("button"))
|
|
buttons = self.buttons.get(message_id) or []
|
|
if index < 0 or index >= len(buttons):
|
|
raise DriverError("Button index was not observed on that message.")
|
|
button = buttons[index]
|
|
if button.get("@type") != "inlineKeyboardButtonTypeCallback":
|
|
raise DriverError("Only callback buttons can be pressed by this harness.")
|
|
self.client.request(
|
|
{
|
|
"@type": "getCallbackQueryAnswer",
|
|
"chat_id": self.chat_id,
|
|
"message_id": message_id,
|
|
"payload": {
|
|
"@type": "callbackQueryPayloadData",
|
|
"data": button.get("data", ""),
|
|
},
|
|
}
|
|
)
|
|
self.pump(0.2)
|
|
return {"ok": True, "cursor": len(self.events)}
|
|
if command == "shutdown":
|
|
self.pump(float(request.get("settleSeconds") or 0))
|
|
return {"ok": True, "cursor": len(self.events), "shutdown": True}
|
|
raise DriverError("Unknown observer command.")
|
|
|
|
|
|
def command_serve_session(args):
|
|
config, bot_config = load_config()
|
|
user_driver = UserDriver(config, bot_config)
|
|
user_driver.authorize(need_ready=True)
|
|
chat_id = user_driver.resolve_chat(args.chat)
|
|
# Authorization can replay updates received while this QA account was offline.
|
|
# The proof timeline begins only after this observer is ready for scenario actions.
|
|
for _ in range(UserObserver.MAX_EVENTS):
|
|
if not user_driver.client.next_update(timeout=0):
|
|
break
|
|
else:
|
|
raise DriverError("Telegram observer startup backlog exceeded its event budget.")
|
|
observer = UserObserver(
|
|
user_driver,
|
|
chat_id,
|
|
args.sut_user_id,
|
|
args.sut_username,
|
|
args.journal,
|
|
args.media_root,
|
|
)
|
|
socket_path = Path(args.socket)
|
|
socket_path.unlink(missing_ok=True)
|
|
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
server.bind(str(socket_path))
|
|
# The session root's default ACL grants mantis-sut access; group rw keeps
|
|
# that named ACL entry effective when the runner tightens this socket.
|
|
socket_path.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IWGRP)
|
|
server.listen(4)
|
|
server.settimeout(0.2)
|
|
shutdown = False
|
|
try:
|
|
while not shutdown:
|
|
update = observer.client.next_update(timeout=0)
|
|
if update:
|
|
observer.ingest(update)
|
|
continue
|
|
try:
|
|
connection, _ = server.accept()
|
|
except TimeoutError:
|
|
continue
|
|
with connection:
|
|
payload = b""
|
|
while b"\n" not in payload and len(payload) <= 65536:
|
|
chunk = connection.recv(65536 - len(payload) + 1)
|
|
if not chunk:
|
|
break
|
|
payload += chunk
|
|
try:
|
|
request = json.loads(payload.split(b"\n", 1)[0])
|
|
response = observer.call(request)
|
|
shutdown = bool(response.pop("shutdown", False))
|
|
response["truncated"] = observer.truncated
|
|
except (
|
|
DriverError,
|
|
ValueError,
|
|
TypeError,
|
|
json.JSONDecodeError,
|
|
) as error:
|
|
response = {"ok": False, "error": str(error)}
|
|
connection.sendall(json.dumps(response, separators=(",", ":")).encode() + b"\n")
|
|
finally:
|
|
observer.close()
|
|
server.close()
|
|
socket_path.unlink(missing_ok=True)
|
|
|
|
|
|
def command_serve(args):
|
|
pid_path = Path(args.pid_file)
|
|
pid_fd = os.open(pid_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(pid_fd, "w") as handle:
|
|
json.dump({"pid": os.getpid(), "pgid": os.getpgrp(), "socket": args.socket}, handle)
|
|
handle.write("\n")
|
|
try:
|
|
command_serve_session(args)
|
|
finally:
|
|
pid_path.unlink(missing_ok=True)
|
|
|
|
|
|
def command_terminate_observer(args):
|
|
pid_path = Path(args.pid_file)
|
|
try:
|
|
metadata = pid_path.lstat()
|
|
except FileNotFoundError:
|
|
return
|
|
if (
|
|
not stat.S_ISREG(metadata.st_mode)
|
|
or stat.S_IMODE(metadata.st_mode) != 0o600
|
|
or metadata.st_uid != os.getuid()
|
|
or metadata.st_nlink != 1
|
|
):
|
|
raise DriverError("Telegram observer pid file is not a private runner-owned file.")
|
|
value = json.loads(pid_path.read_text())
|
|
pid = int(value.get("pid") or 0)
|
|
pgid = int(value.get("pgid") or 0)
|
|
if value.get("socket") != args.socket or pid <= 0 or pgid <= 0 or pgid == os.getpgrp():
|
|
raise DriverError("Telegram observer pid file is invalid.")
|
|
try:
|
|
command_line = Path(f"/proc/{pid}/cmdline").read_bytes()
|
|
except FileNotFoundError:
|
|
pid_path.unlink(missing_ok=True)
|
|
return
|
|
if b"telegram-user-driver" not in command_line or args.socket.encode() not in command_line or b"serve" not in command_line:
|
|
raise DriverError("Telegram observer process identity changed before cleanup.")
|
|
try:
|
|
os.killpg(pgid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
process_stat = Path(f"/proc/{pid}/stat")
|
|
|
|
def running():
|
|
try:
|
|
return process_stat.read_text().rsplit(")", 1)[1].split()[0] != "Z"
|
|
except FileNotFoundError:
|
|
return False
|
|
|
|
deadline = time.monotonic() + 2
|
|
while running() and time.monotonic() < deadline:
|
|
time.sleep(0.05)
|
|
if running():
|
|
try:
|
|
os.killpg(pgid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
pid_path.unlink(missing_ok=True)
|
|
|
|
|
|
def apply_template(text, sut):
|
|
run = f"USER-E2E-{int(time.time())}"
|
|
username = sut.get("username", "")
|
|
return text.replace("{run}", run).replace("{sut}", username), run
|
|
|
|
|
|
def print_result(payload, as_json=False, output=""):
|
|
if output:
|
|
output_path = Path(output).expanduser()
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
|
|
if as_json:
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
else:
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
|
|
|
|
def command_configure(args):
|
|
config, _ = load_config()
|
|
changed = False
|
|
for arg_name, key in [
|
|
("api_id", "apiId"),
|
|
("api_hash", "apiHash"),
|
|
("tdlib_path", "tdlibPath"),
|
|
("chat", "defaultChatId"),
|
|
("sut_username", "sutUsername"),
|
|
("sut_id", "sutId"),
|
|
]:
|
|
value = getattr(args, arg_name)
|
|
if value:
|
|
config[key] = int(value) if key in {"apiId", "sutId"} else value
|
|
changed = True
|
|
if not changed:
|
|
raise DriverError("No config values passed.")
|
|
write_json_private(CONFIG_PATH, config)
|
|
print(f"Wrote {CONFIG_PATH}")
|
|
|
|
|
|
def command_doctor(args):
|
|
config, bot_config = load_config()
|
|
lib_path = find_tdjson(config)
|
|
tdjson_loads = False
|
|
tdjson_error = ""
|
|
if lib_path:
|
|
try:
|
|
ctypes.CDLL(str(lib_path))
|
|
tdjson_loads = True
|
|
except OSError as error:
|
|
tdjson_error = str(error)
|
|
checks = {
|
|
"configPath": str(CONFIG_PATH),
|
|
"tdjson": str(lib_path) if lib_path else "",
|
|
"tdjsonLoads": tdjson_loads,
|
|
"tdjsonError": tdjson_error,
|
|
"hasApiId": bool(env_or_config("TELEGRAM_USER_DRIVER_API_ID", config, "apiId")),
|
|
"hasApiHash": bool(env_or_config("TELEGRAM_USER_DRIVER_API_HASH", config, "apiHash")),
|
|
"hasDefaultChat": bool(default_chat(config, bot_config)),
|
|
"hasSutIdentity": bool(resolve_sut(config, bot_config).get("username")),
|
|
"hasTesterIdentity": bool(config.get("testerUserId")),
|
|
"testerUsername": config.get("testerUsername", ""),
|
|
"hasQrencode": bool(shutil.which("qrencode")),
|
|
}
|
|
checks["ok"] = bool(checks["tdjsonLoads"] and checks["hasApiId"] and checks["hasApiHash"])
|
|
if checks["ok"]:
|
|
checks["next"] = "run login --qr"
|
|
elif not checks["tdjsonLoads"]:
|
|
checks["next"] = "brew install tdlib"
|
|
else:
|
|
checks["next"] = "configure --api-id ... --api-hash ..."
|
|
print_result(checks, args.json, getattr(args, "output", ""))
|
|
if not checks["ok"]:
|
|
sys.exit(1)
|
|
|
|
|
|
def command_login(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(args)
|
|
me = driver.client.request({"@type": "getMe"})
|
|
save_tester_identity(config, me)
|
|
print_result({"ok": True, "user": public_user(me)}, args.json, getattr(args, "output", ""))
|
|
|
|
|
|
def command_status(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
ready = driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms), need_ready=False)
|
|
if not ready:
|
|
print_result(
|
|
{"ok": False, "authorized": False, "next": "login --qr"},
|
|
args.json,
|
|
getattr(args, "output", ""),
|
|
)
|
|
sys.exit(1)
|
|
me = driver.client.request({"@type": "getMe"})
|
|
save_tester_identity(config, me)
|
|
print_result(
|
|
{"ok": True, "authorized": True, "user": public_user(me)},
|
|
args.json,
|
|
getattr(args, "output", ""),
|
|
)
|
|
|
|
|
|
def command_confirm_qr(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
|
|
result = driver.client.request({"@type": "confirmQrCodeAuthentication", "link": args.link}, timeout=30)
|
|
print_result(
|
|
{
|
|
"ok": True,
|
|
"session": {
|
|
"id": result.get("id"),
|
|
"isCurrent": result.get("is_current"),
|
|
"isPasswordPending": result.get("is_password_pending"),
|
|
"applicationName": result.get("application_name"),
|
|
"deviceModel": result.get("device_model"),
|
|
"platform": result.get("platform"),
|
|
"systemVersion": result.get("system_version"),
|
|
},
|
|
},
|
|
args.json,
|
|
getattr(args, "output", ""),
|
|
)
|
|
|
|
|
|
def command_terminate_session(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
|
|
driver.client.request({"@type": "terminateSession", "session_id": int(args.session_id)}, timeout=30)
|
|
print_result(
|
|
{"ok": True, "sessionId": args.session_id},
|
|
args.json,
|
|
getattr(args, "output", ""),
|
|
)
|
|
|
|
|
|
def command_terminate_desktop_sessions(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
|
|
result = driver.client.request({"@type": "getActiveSessions"}, timeout=30)
|
|
terminated = []
|
|
for session in result.get("sessions", []):
|
|
if session.get("is_current"):
|
|
continue
|
|
if session.get("application_name") != "Telegram Desktop":
|
|
continue
|
|
session_id = session.get("id")
|
|
if session_id is None:
|
|
continue
|
|
driver.client.request({"@type": "terminateSession", "session_id": int(session_id)}, timeout=30)
|
|
terminated.append({"id": session_id, "applicationName": session.get("application_name")})
|
|
print_result({"ok": True, "terminated": terminated}, args.json, getattr(args, "output", ""))
|
|
|
|
|
|
def public_user(user):
|
|
return {
|
|
"id": user.get("id"),
|
|
"firstName": user.get("first_name"),
|
|
"lastName": user.get("last_name"),
|
|
"username": user.get("username"),
|
|
"isBot": user.get("type", {}).get("@type") == "userTypeBot",
|
|
}
|
|
|
|
|
|
def save_tester_identity(config, user):
|
|
config["testerUserId"] = user.get("id")
|
|
config["testerUsername"] = user.get("username") or ""
|
|
config["testerFirstName"] = user.get("first_name") or ""
|
|
config["testerLastName"] = user.get("last_name") or ""
|
|
write_json_private(CONFIG_PATH, config)
|
|
|
|
|
|
def command_send(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
|
|
chat_id = driver.resolve_chat(args.chat)
|
|
text, _run = apply_template(args.text, resolve_sut(config, bot_config))
|
|
sent = driver.send_text(chat_id, text, args.reply_to, args.thread_id)
|
|
print_result(
|
|
{"ok": True, "sent": normalize_message(sent)},
|
|
args.json,
|
|
getattr(args, "output", ""),
|
|
)
|
|
|
|
|
|
def command_wait(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
|
|
chat_id = driver.resolve_chat(args.chat)
|
|
message, observed = driver.wait_for_message(chat_id, args, args.after_message_id)
|
|
print_result(
|
|
{
|
|
"ok": bool(message),
|
|
"message": message,
|
|
"observedCount": len(observed),
|
|
"observed": [] if message else observed[-10:],
|
|
},
|
|
args.json,
|
|
getattr(args, "output", ""),
|
|
)
|
|
if not message:
|
|
sys.exit(1)
|
|
|
|
|
|
def command_probe(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
|
|
sut = resolve_sut(config, bot_config)
|
|
chat_id = driver.resolve_chat(args.chat)
|
|
text, run = apply_template(args.text, sut)
|
|
sent = driver.send_text(chat_id, text, args.reply_to)
|
|
wait_args = argparse.Namespace(
|
|
expect=args.expect,
|
|
from_bot=args.from_bot,
|
|
reply_to=sent["id"] if args.require_reply else None,
|
|
thread_id=args.thread_id,
|
|
timeout_ms=args.timeout_ms,
|
|
)
|
|
message, observed = driver.wait_for_message(chat_id, wait_args, sent["id"])
|
|
result = {
|
|
"ok": bool(message),
|
|
"chatId": chat_id,
|
|
"run": run,
|
|
"sent": normalize_message(sent),
|
|
"reply": message,
|
|
"observedCount": len(observed),
|
|
"observed": [] if message else observed[-10:],
|
|
}
|
|
print_result(result, args.json, getattr(args, "output", ""))
|
|
if not message:
|
|
sys.exit(1)
|
|
|
|
|
|
def command_transcript(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
|
|
chat_id = driver.resolve_chat(args.chat)
|
|
history = driver.client.request(
|
|
{
|
|
"@type": "getChatHistory",
|
|
"chat_id": chat_id,
|
|
"from_message_id": 0,
|
|
"offset": 0,
|
|
"limit": args.limit,
|
|
"only_local": False,
|
|
}
|
|
)
|
|
messages = [normalize_message(message) for message in history.get("messages", [])]
|
|
print_result(
|
|
{"ok": True, "chatId": chat_id, "messages": messages},
|
|
args.json,
|
|
getattr(args, "output", ""),
|
|
)
|
|
|
|
|
|
def command_chats(args):
|
|
config, bot_config = load_config()
|
|
driver = UserDriver(config, bot_config)
|
|
driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
|
|
chats = driver.client.request(
|
|
{
|
|
"@type": "getChats",
|
|
"chat_list": {"@type": "chatListMain"},
|
|
"limit": args.limit,
|
|
},
|
|
timeout=20,
|
|
)
|
|
seen = set()
|
|
rows = []
|
|
for chat_id in chats.get("chat_ids", []):
|
|
chat = driver.client.request({"@type": "getChat", "chat_id": chat_id}, timeout=10)
|
|
rows.append(public_chat(chat, "main"))
|
|
seen.add(int(chat["id"]))
|
|
configured = default_chat(config, bot_config)
|
|
if configured and configured.lstrip("-").isdigit():
|
|
try:
|
|
chat = driver.client.request({"@type": "getChat", "chat_id": int(configured)}, timeout=10)
|
|
if int(chat["id"]) not in seen:
|
|
rows.append(public_chat(chat, "configured"))
|
|
except DriverError:
|
|
pass
|
|
print_result(
|
|
{"ok": True, "configuredChat": configured, "chats": rows},
|
|
args.json,
|
|
getattr(args, "output", ""),
|
|
)
|
|
|
|
|
|
def public_chat(chat, source):
|
|
return {
|
|
"id": chat.get("id"),
|
|
"title": chat.get("title"),
|
|
"type": (chat.get("type") or {}).get("@type"),
|
|
"unreadCount": chat.get("unread_count"),
|
|
"source": source,
|
|
}
|
|
|
|
|
|
def add_common(parser):
|
|
parser.add_argument("--json", action="store_true")
|
|
parser.add_argument("--output", default="")
|
|
parser.add_argument("--timeout-ms", type=int, default=120000)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Telegram real-user E2E driver backed by TDLib.")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
configure = sub.add_parser("configure")
|
|
configure.add_argument("--api-id")
|
|
configure.add_argument("--api-hash")
|
|
configure.add_argument("--tdlib-path")
|
|
configure.add_argument("--chat")
|
|
configure.add_argument("--sut-username")
|
|
configure.add_argument("--sut-id")
|
|
configure.set_defaults(func=command_configure)
|
|
|
|
doctor = sub.add_parser("doctor")
|
|
doctor.add_argument("--json", action="store_true")
|
|
doctor.add_argument("--output", default="")
|
|
doctor.set_defaults(func=command_doctor)
|
|
|
|
login = sub.add_parser("login")
|
|
add_common(login)
|
|
login.add_argument("--qr", action="store_true", default=True)
|
|
login.add_argument("--phone", default="")
|
|
login.add_argument("--code", default="")
|
|
login.add_argument("--password", default="")
|
|
login.set_defaults(func=command_login)
|
|
|
|
status = sub.add_parser("status")
|
|
add_common(status)
|
|
status.set_defaults(func=command_status)
|
|
|
|
confirm_qr = sub.add_parser("confirm-qr")
|
|
add_common(confirm_qr)
|
|
confirm_qr.add_argument("--link", required=True)
|
|
confirm_qr.set_defaults(func=command_confirm_qr)
|
|
|
|
terminate_session = sub.add_parser("terminate-session")
|
|
add_common(terminate_session)
|
|
terminate_session.add_argument("--session-id", required=True)
|
|
terminate_session.set_defaults(func=command_terminate_session)
|
|
|
|
terminate_desktop_sessions = sub.add_parser("terminate-desktop-sessions")
|
|
add_common(terminate_desktop_sessions)
|
|
terminate_desktop_sessions.set_defaults(func=command_terminate_desktop_sessions)
|
|
|
|
send = sub.add_parser("send")
|
|
add_common(send)
|
|
send.add_argument("--chat", default="")
|
|
send.add_argument("--text", required=True)
|
|
send.add_argument("--reply-to")
|
|
send.add_argument("--thread-id", type=int, default=0)
|
|
send.set_defaults(func=command_send)
|
|
|
|
wait = sub.add_parser("wait")
|
|
add_common(wait)
|
|
wait.add_argument("--chat", default="")
|
|
wait.add_argument("--expect", action="append", default=[])
|
|
wait.add_argument("--from-bot", default="")
|
|
wait.add_argument("--reply-to")
|
|
wait.add_argument("--thread-id", type=int, default=0)
|
|
wait.add_argument("--after-message-id", type=int, default=0)
|
|
wait.set_defaults(func=command_wait)
|
|
|
|
probe = sub.add_parser("probe")
|
|
add_common(probe)
|
|
probe.add_argument("--chat", default="")
|
|
probe.add_argument("--text", default="@{sut} Reply exactly: USER-E2E-{run}")
|
|
probe.add_argument("--expect", action="append", default=[])
|
|
probe.add_argument("--from-bot", default="")
|
|
probe.add_argument("--reply-to")
|
|
probe.add_argument("--thread-id", type=int, default=0)
|
|
probe.add_argument("--require-reply", action="store_true", default=True)
|
|
probe.add_argument("--any-sut-reply", dest="require_reply", action="store_false")
|
|
probe.set_defaults(func=command_probe)
|
|
|
|
transcript = sub.add_parser("transcript")
|
|
add_common(transcript)
|
|
transcript.add_argument("--chat", default="")
|
|
transcript.add_argument("--limit", type=int, default=20)
|
|
transcript.set_defaults(func=command_transcript)
|
|
|
|
chats = sub.add_parser("chats")
|
|
add_common(chats)
|
|
chats.add_argument("--limit", type=int, default=50)
|
|
chats.set_defaults(func=command_chats)
|
|
|
|
serve = sub.add_parser("serve", help=argparse.SUPPRESS)
|
|
serve.add_argument("--chat", required=True)
|
|
serve.add_argument("--sut-user-id", required=True, type=int)
|
|
serve.add_argument("--sut-username", required=True)
|
|
serve.add_argument("--socket", required=True)
|
|
serve.add_argument("--pid-file", required=True)
|
|
serve.add_argument("--journal", required=True)
|
|
serve.add_argument("--media-root", required=True)
|
|
serve.set_defaults(func=command_serve)
|
|
|
|
terminate_observer = sub.add_parser("terminate-observer", help=argparse.SUPPRESS)
|
|
terminate_observer.add_argument("--pid-file", required=True)
|
|
terminate_observer.add_argument("--socket", required=True)
|
|
terminate_observer.set_defaults(func=command_terminate_observer)
|
|
|
|
args = parser.parse_args()
|
|
try:
|
|
args.func(args)
|
|
except DriverError as error:
|
|
print(str(error), file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|