Spaces:
Paused
Paused
File size: 8,359 Bytes
1c49bd5 d8cb92d 1c49bd5 d8cb92d 1c49bd5 d8cb92d 1c49bd5 d8cb92d 1c49bd5 d8cb92d 1c49bd5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 | """Safely activate, register, inspect, or remove the Telegram webhook.
The ``activate`` action is intended for a trusted workstation. It prompts
with :func:`getpass.getpass` when the bot token is not inherited by the
process, and never writes that token or generated secrets to disk.
"""
from __future__ import annotations
import argparse
import getpass
import json
import os
import secrets
import sys
import time
from typing import Any
import requests
REPO_ID = "Really-amin/Asset"
PUBLIC_BASE = "https://really-amin-asset.hf.space"
WEBHOOK_PATH = "/api/telegram/webhook"
ALLOWED_UPDATES = ["message", "callback_query"]
def _token() -> str:
value = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip()
if value:
return value
return getpass.getpass("Telegram Bot Token: ").strip()
def _call(token: str, method: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
"""Call Telegram without ever including the token in diagnostics."""
try:
response = requests.post(
f"https://api.telegram.org/bot{token}/{method}",
json=payload or {},
timeout=(5, 15),
)
try:
body = response.json()
except ValueError:
body = {"ok": False, "description": "invalid Telegram response"}
if not isinstance(body, dict):
return {"ok": False, "description": "invalid Telegram response"}
body["_status"] = response.status_code
return body
except requests.RequestException:
return {"ok": False, "description": "Telegram request failed"}
def _safe_info(body: dict[str, Any]) -> dict[str, Any]:
result = body.get("result") if isinstance(body.get("result"), dict) else {}
return {
"ok": bool(body.get("ok")),
"status": body.get("_status"),
"url_configured": bool(result.get("url")),
"url_path": (str(result.get("url", "")).split(".hf.space", 1)[-1] if result.get("url") else ""),
"pending_update_count": result.get("pending_update_count", 0),
"allowed_updates": result.get("allowed_updates", []),
"last_error": bool(result.get("last_error_message")),
"description": body.get("description", ""),
}
def _hf_api():
from huggingface_hub import HfApi
token = os.environ.get("HF_TOKEN", "").strip() or None
return HfApi(token=token)
def _wait_running(api: Any, timeout: int = 900) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
try:
runtime = api.get_space_runtime(REPO_ID)
stage = str(getattr(runtime, "stage", "") or "").upper()
if stage == "RUNNING":
return True
if stage in {"RUNTIME_ERROR", "BUILD_ERROR", "ERROR"}:
return False
except Exception:
pass
time.sleep(10)
return False
def _configure_space(api: Any, webhook_secret: str, bootstrap_secret: str, bot_token: str) -> None:
# The token is updated in the Space Secret but is never printed or saved.
api.add_space_secret(REPO_ID, "TELEGRAM_BOT_TOKEN", bot_token)
api.add_space_secret(REPO_ID, "TELEGRAM_WEBHOOK_SECRET", webhook_secret)
api.add_space_secret(REPO_ID, "TELEGRAM_BOOTSTRAP_SECRET", bootstrap_secret)
for key, value in {
"TELEGRAM_ENABLED": "true",
"TELEGRAM_MODE": "webhook",
"TELEGRAM_PUBLIC_BASE_URL": PUBLIC_BASE,
"TELEGRAM_WEBHOOK_PATH": WEBHOOK_PATH,
}.items():
api.add_space_variable(REPO_ID, key, value)
def _activate(args: argparse.Namespace, bot_token: str) -> int:
if not bot_token:
print("A Telegram Bot Token is required.", file=sys.stderr)
return 2
webhook_secret = secrets.token_urlsafe(32)
bootstrap_secret = secrets.token_urlsafe(32)
try:
api = _hf_api()
_configure_space(api, webhook_secret, bootstrap_secret, bot_token)
print("Space secrets and webhook-mode variables configured.")
if not _wait_running(api):
print("Space did not reach RUNNING after configuration.", file=sys.stderr)
return 1
me = _call(bot_token, "getMe")
if not me.get("ok"):
print("getMe failed; activation stopped.", file=sys.stderr)
return 1
user = me.get("result") if isinstance(me.get("result"), dict) else {}
username = str(user.get("username") or "(username unavailable)")
webhook = _call(bot_token, "setWebhook", {
"url": PUBLIC_BASE + WEBHOOK_PATH,
"secret_token": webhook_secret,
"allowed_updates": ALLOWED_UPDATES,
"drop_pending_updates": bool(args.drop_pending),
})
if not webhook.get("ok"):
print("setWebhook failed; activation stopped.", file=sys.stderr)
return 1
info = _call(bot_token, "getWebhookInfo")
print(json.dumps({"bot_username": username, "webhook": _safe_info(info)}, sort_keys=True))
print(f"One-time owner claim: /claim {bootstrap_secret}")
print("Send that command in a private chat, then press Enter here.")
input()
# The bootstrap secret is consumed by the Space after a successful claim.
status_url = PUBLIC_BASE + "/api/telegram/bootstrap/status"
claimed = False
for _ in range(30):
try:
response = requests.get(status_url, headers={"X-Telegram-Bot-Api-Secret-Token": webhook_secret}, timeout=(5, 10))
data = response.json() if response.headers.get("content-type", "").startswith("application/json") else {}
claimed = bool(isinstance(data, dict) and data.get("ownerClaimed"))
except requests.RequestException:
pass
if claimed:
break
time.sleep(2)
if claimed:
try:
api.delete_space_secret(REPO_ID, "TELEGRAM_BOOTSTRAP_SECRET")
except Exception:
print("Owner claimed; remove the bootstrap secret from Space settings manually.")
print("Owner claim confirmed; one-time bootstrap secret removed.")
else:
print("Owner claim was not confirmed before timeout; bootstrap remains available for one valid claim.")
return 0
finally:
# Remove references before returning; no token is written by this tool.
bot_token = ""
webhook_secret = ""
bootstrap_secret = ""
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("action", choices=("activate", "register", "info", "remove"))
parser.add_argument("--drop-pending", action="store_true")
args = parser.parse_args()
if args.action == "activate":
token = _token()
if not token:
print("A Telegram Bot Token is required.", file=sys.stderr)
return 2
try:
return _activate(args, token)
finally:
token = ""
token = _token()
if not token:
print("A Telegram Bot Token is required.", file=sys.stderr)
return 2
try:
if args.action == "info":
print(json.dumps(_safe_info(_call(token, "getWebhookInfo")), sort_keys=True)); return 0
if args.action == "remove":
body = _call(token, "deleteWebhook", {"drop_pending_updates": args.drop_pending})
print(json.dumps({"ok": bool(body.get("ok")), "status": body.get("_status"), "description": body.get("description", "")}, sort_keys=True)); return 0 if body.get("ok") else 1
public = os.environ.get("TELEGRAM_PUBLIC_BASE_URL", "").rstrip("/") or PUBLIC_BASE
path = os.environ.get("TELEGRAM_WEBHOOK_PATH", WEBHOOK_PATH)
secret = os.environ.get("TELEGRAM_WEBHOOK_SECRET", "")
if not secret:
print("TELEGRAM_WEBHOOK_SECRET is required for register.", file=sys.stderr); return 2
body = _call(token, "setWebhook", {"url": public + path, "secret_token": secret, "allowed_updates": ALLOWED_UPDATES, "drop_pending_updates": args.drop_pending})
print(json.dumps({"ok": bool(body.get("ok")), "status": body.get("_status"), "description": body.get("description", "")}, sort_keys=True)); return 0 if body.get("ok") else 1
finally:
token = ""
if __name__ == "__main__":
raise SystemExit(main())
|