Telegram Server Manager

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Simple FastAPI server for Hugging Face Spaces This keeps the space alive and provides a basic web interface """ from fastapi import FastAPI, Request, HTTPException, Form, UploadFile, File from fastapi.responses import HTMLResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware import uvicorn import os import logging import subprocess import configparser import tempfile import traceback # Configure logging logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO ) logger = logging.getLogger(__name__) app = FastAPI() # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Load config for authentication def load_config(): """Load configuration file""" config = configparser.ConfigParser() if os.path.exists("config"): config.read("config") return config return None def verify_admin(chat_id: str): """Verify if the provided chat_id matches admin""" config = load_config() if config and "SecretConfig" in config: admin_cid = config["SecretConfig"].get("admincid", "") return str(chat_id) == str(admin_cid) # If no config, allow access for demo purposes in restricted environments # WARNING: This is insecure in production. Always use a config file with proper admin_id logger.warning("No config file found - running in insecure demo mode") return True @app.get("/", response_class=HTMLResponse) async def root(): """Root endpoint that returns a professional web interface""" # Check if config file exists config_exists = os.path.exists("config") return f""" Telegram Server Manager - Web Console

⚠️ Demo Mode: Config file not found. Some features may be restricted. Create a config file to enable full functionality.

' if not config_exists else ''}

💻 Command Executor

💡 Tip: Run any shell command directly from the browser

Command:

Output will appear here...

🐍 Python Evaluator

💡 Tip: Execute Python code with full system access

Python Code:

Output will appear here...

📁 Python File Executor

💡 Tip: Upload and run Python files

Upload Python File (.py):

Output will appear here...

📊 Quick System Info

Click a button to see system info...

🔒 Security Notice:

⚠️ IMPORTANT: This web interface provides full system access and is designed for trusted environments only.

In production environments:

Use strong authentication (configure admin chat ID in config file)
Deploy behind a reverse proxy with HTTPS
Implement rate limiting and input validation
Monitor and log all command executions
Restrict network access to trusted IPs only

Note: This tool intentionally allows shell command execution and Python eval - features equivalent to the Telegram bot. Use responsibly.

""" @app.post("/api/execute") async def execute_command(request: Request): """Execute a shell command""" try: data = await request.json() command = data.get("command", "").strip() admin_id = data.get("admin_id", "") if not command: return JSONResponse({"success": False, "error": "No command provided"}) # Verify admin (basic check) if not verify_admin(admin_id): return JSONResponse({"success": False, "error": "Unauthorized"}) logger.info(f"Executing command: {command}") # Execute command with timeout try: result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=30 ) output = result.stdout if result.stdout else result.stderr if not output: output = "Command executed successfully (no output)" return JSONResponse({ "success": True, "output": output, "return_code": result.returncode }) except subprocess.TimeoutExpired: return JSONResponse({ "success": False, "error": "Command timed out after 30 seconds" }) except Exception as e: return JSONResponse({ "success": False, "error": f"Execution error: {str(e)}" }) except Exception as e: logger.error(f"Error in execute_command: {e}") return JSONResponse({ "success": False, "error": f"Server error: {str(e)}" }) @app.post("/api/eval") async def evaluate_python(request: Request): """Execute Python code""" try: data = await request.json() code = data.get("code", "").strip() admin_id = data.get("admin_id", "") if not code: return JSONResponse({"success": False, "error": "No code provided"}) # Verify admin if not verify_admin(admin_id): return JSONResponse({"success": False, "error": "Unauthorized"}) logger.info(f"Executing Python code (length: {len(code)})") try: # Create a namespace for execution namespace = { '__builtins__': __builtins__, 'os': os, 'subprocess': subprocess, } # Capture stdout import sys from io import StringIO old_stdout = sys.stdout sys.stdout = StringIO() try: # Try to evaluate as expression first try: result = eval(code, namespace) if result is not None: print(result) except SyntaxError: # If it fails, execute as statement exec(code, namespace) output = sys.stdout.getvalue() if not output: output = "Code executed successfully (no output)" return JSONResponse({ "success": True, "output": output }) finally: sys.stdout = old_stdout except Exception as e: return JSONResponse({ "success": False, "error": f"{type(e).__name__}: {str(e)}\\n{traceback.format_exc()}" }) except Exception as e: logger.error(f"Error in evaluate_python: {e}") return JSONResponse({ "success": False, "error": f"Server error: {str(e)}" }) @app.post("/api/run-file") async def run_python_file(file: UploadFile = File(...), admin_id: str = Form(...)): """Upload and execute a Python file""" try: # Verify admin if not verify_admin(admin_id): return JSONResponse({"success": False, "error": "Unauthorized"}) # Verify file type if not file.filename.endswith('.py'): return JSONResponse({ "success": False, "error": "Only .py files are allowed" }) logger.info(f"Executing uploaded file: {file.filename}") # Save file to temp location with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as tmp: content = await file.read() tmp.write(content.decode('utf-8')) tmp_path = tmp.name try: # Execute the file result = subprocess.run( ['python3', tmp_path], # Use python3 from PATH instead of hardcoded path capture_output=True, text=True, timeout=30 ) output = result.stdout if result.stdout else result.stderr if not output: output = "File executed successfully (no output)" return JSONResponse({ "success": True, "output": output, "return_code": result.returncode, "filename": file.filename }) except subprocess.TimeoutExpired: return JSONResponse({ "success": False, "error": "Execution timed out after 30 seconds" }) finally: # Clean up temp file try: os.unlink(tmp_path) except (FileNotFoundError, PermissionError) as e: logger.warning(f"Could not delete temp file {tmp_path}: {e}") except Exception as e: logger.error(f"Error in run_python_file: {e}") return JSONResponse({ "success": False, "error": f"Server error: {str(e)}" }) @app.get("/health") async def health(): """Health check endpoint""" return {"status": "healthy", "bot": "running"} @app.get("/api/status") async def status(): """Status endpoint""" return { "status": "online", "service": "Telegram Server Manager Bot", "port": 7860, "web_console": "enabled" } if __name__ == "__main__": # Run on port 7860 for Hugging Face Spaces port = int(os.environ.get("PORT", 7860)) logger.info(f"Starting web server on port {port}...") uvicorn.run(app, host="0.0.0.0", port=port)

🚀 Telegram Server Manager

💻 Command Executor

🐍 Python Evaluator

📁 Python File Executor

📊 Quick System Info

🔒 Security Notice: