kobalt/gotelegram_pro
1
0
Fork 0
mirror of https://github.com/anten-ka/gotelegram_pro.git synced 2026-05-19 17:56:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

v2.5.0: add key disable switches and pro UI polish

Browse Source
This commit is contained in:
Виталий Литвинов
2026-04-25 09:44:53 +03:00
parent 6b89c3ea81
commit e54778c08c
17 changed files with 683 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
admin-web/server.py
View File

@@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
GoTelegram local web admin.
goTelegram Pro local web admin.
The service is intentionally bound to 127.0.0.1:1984. Operators reach it
through an SSH tunnel; it must never be exposed directly on the public network.
@@ -37,6 +37,7 @@ CURRENT_STATS = Path(os.getenv("GOTELEGRAM_STATS_CURRENT", "/run/gotelegram/stat
BACKUP_DIR = Path(os.getenv("GOTELEGRAM_BACKUP_DIR", "/opt/gotelegram/backups"))
INSTALL_DIR = Path(os.getenv("GOTELEGRAM_DIR", "/opt/gotelegram"))
BOT_DIR = Path(os.getenv("GOTELEGRAM_BOT_DIR", "/opt/gotelegram-bot"))
DISABLED_USERS_FILE = Path(os.getenv("GOTELEGRAM_DISABLED_USERS", "/opt/gotelegram/disabled_users.json"))
HOST = os.getenv("GOTELEGRAM_ADMIN_HOST", "127.0.0.1")
PORT = int(os.getenv("GOTELEGRAM_ADMIN_PORT", "1984"))
@@ -152,6 +153,44 @@ def read_telemt_users() -> dict[str, str]:
return users
def read_disabled_users() -> dict[str, str]:
raw = load_json(DISABLED_USERS_FILE, {}) or {}
if not isinstance(raw, dict):
return {}
users = raw.get("users") if isinstance(raw.get("users"), dict) else raw
if not isinstance(users, dict):
return {}
clean: dict[str, str] = {}
for name, secret in users.items():
if name in {"version", "updated_at"}:
continue
name_s = str(name).strip()
secret_s = str(secret or "").strip()
if USER_RE.match(name_s) and secret_s:
clean[name_s] = secret_s
return clean
def write_disabled_users(users: dict[str, str]) -> None:
payload = {
"version": 1,
"updated_at": utc_now(),
"users": {name: users[name] for name in sorted(users)},
}
save_json(DISABLED_USERS_FILE, payload)
def read_user_records() -> dict[str, dict[str, Any]]:
active = read_telemt_users()
disabled = read_disabled_users()
records: dict[str, dict[str, Any]] = {}
for name, secret in disabled.items():
records[name] = {"secret": secret, "enabled": False}
for name, secret in active.items():
records[name] = {"secret": secret, "enabled": True}
return records
def _ordered_user_lines(users: dict[str, str]) -> list[str]:
names = []
if "main" in users:
@@ -462,14 +501,15 @@ def read_log_payload(service: str) -> dict[str, Any]:
}
def user_payload(name: str, secret: str, include_runtime: bool = False) -> dict[str, Any]:
def user_payload(name: str, secret: str, enabled: bool = True, include_runtime: bool = False) -> dict[str, Any]:
item: dict[str, Any] = {
"name": name,
"secret": secret,
"link": proxy_link(secret),
"main": name == "main",
"enabled": bool(enabled),
}
if include_runtime:
if include_runtime and enabled:
item["runtime"] = telemt_api(f"/v1/users/{urllib.parse.quote(name, safe='')}")
return item
@@ -477,7 +517,7 @@ def user_payload(name: str, secret: str, include_runtime: bool = False) -> dict[
def overview_payload() -> dict[str, Any]:
config = load_json(GOTELEGRAM_CONFIG, {}) or {}
language = read_language(config)
users = read_telemt_users()
users = read_user_records()
current = load_json(CURRENT_STATS, {}) or {}
history = load_stats_history()
summary = telemt_api("/v1/stats/summary")
@@ -506,7 +546,7 @@ def overview_payload() -> dict[str, Any]:
class AdminHandler(BaseHTTPRequestHandler):
server_version = "GoTelegramAdmin/2.5.0"
server_version = "goTelegramProAdmin/2.5.0"
def log_message(self, fmt: str, *args: Any) -> None:
print("%s - %s" % (self.address_string(), fmt % args))
@@ -542,15 +582,20 @@ class AdminHandler(BaseHTTPRequestHandler):
if path == "/api/overview":
self.send_json({"ok": True, "data": overview_payload()})
elif path == "/api/users":
users = read_telemt_users()
self.send_json({"ok": True, "data": [user_payload(k, v) for k, v in sorted(users.items())]})
users = read_user_records()
items = []
for name in sorted(users, key=lambda item: (item != "main", item)):
record = users[name]
items.append(user_payload(name, record["secret"], record["enabled"]))
self.send_json({"ok": True, "data": items})
elif path.startswith("/api/users/"):
name = urllib.parse.unquote(path[len("/api/users/"):])
users = read_telemt_users()
users = read_user_records()
if name not in users:
self.send_error_json(404, "user not found")
return
self.send_json({"ok": True, "data": user_payload(name, users[name], include_runtime=True)})
record = users[name]
self.send_json({"ok": True, "data": user_payload(name, record["secret"], record["enabled"], include_runtime=True)})
elif path == "/api/backups":
self.send_json({"ok": True, "data": list_backups()})
elif path == "/api/stats":
@@ -586,10 +631,11 @@ class AdminHandler(BaseHTTPRequestHandler):
if not USER_RE.match(name):
self.send_error_json(400, "invalid user name")
return
users = read_telemt_users()
if name in users:
records = read_user_records()
if name in records:
self.send_error_json(409, "user already exists")
return
users = read_telemt_users()
seed = f"{name}:{time.time()}:{secrets.token_hex(32)}".encode()
secret = hashlib.sha256(seed).hexdigest()[:32]
users[name] = secret
@@ -599,7 +645,37 @@ class AdminHandler(BaseHTTPRequestHandler):
self.send_error_json(500, f"failed to save config: {exc}")
return
restarted = restart_service("telemt")
self.send_json({"ok": True, "data": user_payload(name, secret), "restarted": restarted})
self.send_json({"ok": True, "data": user_payload(name, secret, True), "restarted": restarted})
elif path.startswith("/api/users/") and path.endswith("/enabled"):
name = urllib.parse.unquote(path[len("/api/users/"):-len("/enabled")])
if name == "main":
self.send_error_json(400, "main user cannot be disabled")
return
enabled = bool(body.get("enabled"))
active = read_telemt_users()
disabled = read_disabled_users()
records = read_user_records()
if name not in records:
self.send_error_json(404, "user not found")
return
if enabled:
secret = disabled.pop(name, records[name]["secret"])
active[name] = secret
else:
secret = active.pop(name, records[name]["secret"])
disabled[name] = secret
try:
if enabled:
write_telemt_users(active)
write_disabled_users(disabled)
else:
write_disabled_users(disabled)
write_telemt_users(active)
except Exception as exc:
self.send_error_json(500, f"failed to save config: {exc}")
return
restarted = restart_service("telemt")
self.send_json({"ok": True, "data": user_payload(name, secret, enabled), "restarted": restarted})
elif path == "/api/backups":
ok, result = create_backup()
self.send_json({"ok": ok, "data": {"path": result, "backups": list_backups()}}, 200 if ok else 500)
@@ -640,13 +716,17 @@ class AdminHandler(BaseHTTPRequestHandler):
if name == "main":
self.send_error_json(400, "main user cannot be deleted")
return
users = read_telemt_users()
if name not in users:
active = read_telemt_users()
disabled = read_disabled_users()
records = read_user_records()
if name not in records:
self.send_error_json(404, "user not found")
return
users.pop(name, None)
active.pop(name, None)
disabled.pop(name, None)
try:
write_telemt_users(users)
write_telemt_users(active)
write_disabled_users(disabled)
except Exception as exc:
self.send_error_json(500, f"failed to save config: {exc}")
return
@@ -702,7 +782,7 @@ def main() -> None:
if not STATIC_DIR.exists():
raise SystemExit(f"static dir not found: {STATIC_DIR}")
httpd = ThreadingHTTPServer((HOST, PORT), AdminHandler)
print(f"GoTelegram admin listening on http://{HOST}:{PORT}")
print(f"goTelegram Pro admin listening on http://{HOST}:{PORT}")
httpd.serve_forever()