app.py

"""
╔══════════════════════════════════════════════════════════════╗
║         LINUX SYSTEM SERVER — by BU THE ZYZ STUDIO          ║
║    🖥️  سيرفر نظام لينكس — يُتحكم فيه عبر السيرفر الأول     ║
╚══════════════════════════════════════════════════════════════╝

هذا الملف خاص بـ: سيرفر 2 (سيرفر نظام Linux)
This file belongs to: SERVER 2 (Linux System Server)
"""

import os
import subprocess
import json
import threading
import time
import shutil
import signal
from pathlib import Path
from flask import Flask, request, jsonify, Response, stream_with_context
from flask_cors import CORS

app = Flask(__name__)
CORS(app, resources={r"/api/*": {"origins": "*"}})

# ══════════════════════════════════════════
#  🔐 Security Token — رمز الأمان
#  Set as environment variable: LINUX_API_TOKEN
#  عيّن هذا المتغير في إعدادات Hugging Face Space
# ══════════════════════════════════════════
API_TOKEN = os.environ.get("LINUX_API_TOKEN", "zyz-linux-secret-2025")

# ══════════════════════════════════════════
#  📁 Workspace Directory
# ══════════════════════════════════════════
WORKSPACE = "/workspace"
os.makedirs(WORKSPACE, exist_ok=True)

# Active processes store (for killing long-running processes)
active_processes = {}
process_lock = threading.Lock()


# ══════════════════════════════════════════
#  🔒 Auth Helper
# ══════════════════════════════════════════
def verify_token(req) -> bool:
    """التحقق من رمز الأمان في الطلب"""
    # Check header first
    token_header = req.headers.get("X-API-Token", "")
    if token_header == API_TOKEN:
        return True
    # Check JSON body
    if req.is_json:
        body = req.get_json(silent=True) or {}
        if body.get("token", "") == API_TOKEN:
            return True
    return False


# ══════════════════════════════════════════
#  ✅ Status Endpoint
# ══════════════════════════════════════════
@app.route("/")
def root():
    return jsonify({
        "name": "Linux System Server",
        "version": "1.0.0",
        "by": "BU THE ZYZ STUDIO",
        "status": "online",
        "note": "Use /api/status with your token to verify authentication."
    })


@app.route("/api/status")
def status():
    """حالة السيرفر — يعمل بدون توثيق للتحقق من الاتصال"""
    try:
        uname = subprocess.run("uname -a", shell=True, capture_output=True, text=True, timeout=5)
        disk   = subprocess.run("df -h /workspace 2>/dev/null || df -h /", shell=True, capture_output=True, text=True, timeout=5)
        mem    = subprocess.run("free -h", shell=True, capture_output=True, text=True, timeout=5)
        python = subprocess.run("python3 --version", shell=True, capture_output=True, text=True, timeout=5)
    except Exception:
        uname = disk = mem = python = type("R", (), {"stdout": "N/A", "stderr": ""})()

    authenticated = verify_token(request)
    return jsonify({
        "status": "online",
        "authenticated": authenticated,
        "system": uname.stdout.strip(),
        "disk": disk.stdout.strip(),
        "memory": mem.stdout.strip(),
        "python": python.stdout.strip(),
        "workspace": WORKSPACE,
        "timestamp": time.time()
    })


# ══════════════════════════════════════════
#  ⚡ Execute Command — تنفيذ أوامر الترمينال
# ══════════════════════════════════════════
@app.route("/api/execute", methods=["POST"])
def execute_command():
    """
    تنفيذ أمر ترمينال واحد والحصول على النتيجة كاملة
    Execute a single terminal command and get full output
    """
    if not verify_token(request):
        return jsonify({"error": "Unauthorized — Invalid or missing API token"}), 401

    data    = request.get_json(silent=True) or {}
    command = data.get("command", "").strip()
    cwd     = data.get("cwd", WORKSPACE)
    timeout = min(int(data.get("timeout", 30)), 300)  # max 5 minutes

    if not command:
        return jsonify({"error": "No command provided"}), 400

    # Validate working directory
    if not os.path.exists(cwd):
        cwd = WORKSPACE

    try:
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            cwd=cwd,
            timeout=timeout,
            env={**os.environ, "HOME": "/root", "TERM": "xterm-256color"}
        )
        return jsonify({
            "success": True,
            "stdout":     result.stdout,
            "stderr":     result.stderr,
            "returncode": result.returncode,
            "command":    command,
            "cwd":        cwd
        })
    except subprocess.TimeoutExpired:
        return jsonify({
            "success": False,
            "error": f"Command timed out after {timeout} seconds",
            "returncode": -1
        }), 408
    except Exception as e:
        return jsonify({
            "success": False,
            "error": str(e),
            "returncode": -1
        }), 500


# ══════════════════════════════════════════
#  📡 Stream Execute — تنفيذ مع بث النتائج
# ══════════════════════════════════════════
@app.route("/api/stream", methods=["POST"])
def stream_execute():
    """
    تنفيذ أمر مع بث المخرجات سطراً بسطر (مفيد للأوامر الطويلة)
    Stream command output line by line (useful for long-running commands)
    """
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401

    data    = request.get_json(silent=True) or {}
    command = data.get("command", "").strip()
    cwd     = data.get("cwd", WORKSPACE)
    pid_key = data.get("pid_key", f"proc_{int(time.time())}")

    if not command:
        return jsonify({"error": "No command provided"}), 400

    if not os.path.exists(cwd):
        cwd = WORKSPACE

    @stream_with_context
    def generate():
        try:
            process = subprocess.Popen(
                command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                cwd=cwd,
                env={**os.environ, "HOME": "/root", "TERM": "xterm-256color"},
                bufsize=1
            )
            with process_lock:
                active_processes[pid_key] = process

            yield f"data: {json.dumps({'type': 'start', 'pid': process.pid, 'command': command})}\n\n"

            for line in iter(process.stdout.readline, ""):
                yield f"data: {json.dumps({'type': 'output', 'line': line})}\n\n"

            process.wait()
            with process_lock:
                active_processes.pop(pid_key, None)

            yield f"data: {json.dumps({'type': 'done', 'returncode': process.returncode})}\n\n"
            yield "data: [DONE]\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"

    return Response(
        generate(),
        mimetype="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )


# ══════════════════════════════════════════
#  🛑 Kill Process — إيقاف عملية جارية
# ══════════════════════════════════════════
@app.route("/api/kill", methods=["POST"])
def kill_process():
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401
    data = request.get_json(silent=True) or {}
    pid_key = data.get("pid_key", "")
    with process_lock:
        proc = active_processes.get(pid_key)
    if proc:
        try:
            proc.terminate()
            time.sleep(0.5)
            if proc.poll() is None:
                proc.kill()
            return jsonify({"success": True, "message": "Process terminated"})
        except Exception as e:
            return jsonify({"error": str(e)}), 500
    return jsonify({"error": "Process not found"}), 404


# ══════════════════════════════════════════
#  📁 File System Operations
# ══════════════════════════════════════════
@app.route("/api/files/list", methods=["GET", "POST"])
def list_files():
    """قائمة الملفات في مسار معين"""
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401

    if request.method == "POST":
        data = request.get_json(silent=True) or {}
        path = data.get("path", WORKSPACE)
    else:
        path = request.args.get("path", WORKSPACE)

    try:
        items = []
        for name in sorted(os.listdir(path)):
            full = os.path.join(path, name)
            try:
                stat = os.stat(full)
                items.append({
                    "name":     name,
                    "path":     full,
                    "is_dir":   os.path.isdir(full),
                    "size":     stat.st_size,
                    "modified": stat.st_mtime
                })
            except Exception:
                pass
        return jsonify({"success": True, "path": path, "items": items})
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/files/read", methods=["POST"])
def read_file():
    """قراءة محتوى ملف"""
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401
    data = request.get_json(silent=True) or {}
    path = data.get("path", "").strip()
    if not path:
        return jsonify({"error": "No path provided"}), 400
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            content = f.read()
        return jsonify({"success": True, "path": path, "content": content, "size": len(content)})
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/files/write", methods=["POST"])
def write_file():
    """كتابة محتوى إلى ملف"""
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401
    data    = request.get_json(silent=True) or {}
    path    = data.get("path", "").strip()
    content = data.get("content", "")
    if not path:
        return jsonify({"error": "No path provided"}), 400
    try:
        dir_ = os.path.dirname(path)
        if dir_:
            os.makedirs(dir_, exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return jsonify({"success": True, "path": path, "bytes_written": len(content.encode())})
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/files/delete", methods=["POST"])
def delete_file():
    """حذف ملف أو مجلد"""
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401
    data = request.get_json(silent=True) or {}
    path = data.get("path", "").strip()
    if not path:
        return jsonify({"error": "No path provided"}), 400
    try:
        if os.path.isdir(path):
            shutil.rmtree(path)
        else:
            os.remove(path)
        return jsonify({"success": True, "deleted": path})
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/files/mkdir", methods=["POST"])
def make_dir():
    """إنشاء مجلد جديد"""
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401
    data = request.get_json(silent=True) or {}
    path = data.get("path", "").strip()
    try:
        os.makedirs(path, exist_ok=True)
        return jsonify({"success": True, "created": path})
    except Exception as e:
        return jsonify({"error": str(e)}), 500


# ══════════════════════════════════════════
#  📦 Package Manager — إدارة الحزم
# ══════════════════════════════════════════
@app.route("/api/packages/install", methods=["POST"])
def install_package():
    """تثبيت حزمة Python أو apt"""
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401
    data    = request.get_json(silent=True) or {}
    package = data.get("package", "").strip()
    manager = data.get("manager", "pip")  # "pip" or "apt"
    if not package:
        return jsonify({"error": "No package name"}), 400
    try:
        if manager == "apt":
            cmd = f"apt-get install -y {package}"
        else:
            cmd = f"pip install {package}"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=120)
        return jsonify({
            "success": result.returncode == 0,
            "stdout": result.stdout,
            "stderr": result.stderr
        })
    except subprocess.TimeoutExpired:
        return jsonify({"error": "Installation timed out"}), 408
    except Exception as e:
        return jsonify({"error": str(e)}), 500


# ══════════════════════════════════════════
#  🔧 System Info — معلومات النظام
# ══════════════════════════════════════════
@app.route("/api/sysinfo", methods=["GET", "POST"])
def sysinfo():
    """معلومات تفصيلية عن النظام"""
    if not verify_token(request):
        return jsonify({"error": "Unauthorized"}), 401

    commands = {
        "os":       "cat /etc/os-release 2>/dev/null | head -5",
        "cpu":      "lscpu 2>/dev/null | grep -E 'Architecture|CPU\\(s\\)|Model name' | head -5",
        "memory":   "free -h",
        "disk":     "df -h",
        "network":  "ip addr show 2>/dev/null | grep 'inet ' | head -5",
        "python":   "python3 --version && pip3 list 2>/dev/null | head -20",
        "processes":"ps aux --sort=-%cpu 2>/dev/null | head -10",
        "uptime":   "uptime"
    }
    info = {}
    for key, cmd in commands.items():
        try:
            r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
            info[key] = r.stdout.strip()
        except Exception:
            info[key] = "N/A"

    return jsonify({"success": True, "info": info})


# ══════════════════════════════════════════
#  🚀 Run — تشغيل السيرفر
# ══════════════════════════════════════════
if __name__ == "__main__":
    print("=" * 60)
    print("  🖥️  Linux System Server — BU THE ZYZ STUDIO")
    print("  Port: 7860")
    print("  Token: Set LINUX_API_TOKEN env variable")
    print("=" * 60)
    app.run(host="0.0.0.0", port=7860, debug=False, threaded=True)