TzX Team - Future Coding 2999

from fastapi import FastAPI, Request, Depends, HTTPException, status from fastapi.responses import HTMLResponse from fastapi.security import HTTPBasic, HTTPBasicCredentials import uvicorn import asyncio import socket import threading import logging from rich.console import Console from rich.panel import Panel import struct import select import secrets # إعداد تسجيل الأخطاء logging.basicConfig( filename='socks5.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) console = Console() app = FastAPI(title="TzX Team SOCKS5 Server") # إعداد المصادقة الأساسية security = HTTPBasic() USERNAME = "admin" # يمكنك تغييره PASSWORD = "TzX2999" # يمكنك تغييره # التحقق من اسم المستخدم وكلمة المرور def verify_credentials(credentials: HTTPBasicCredentials = Depends(security)): correct_username = secrets.compare_digest(credentials.username, USERNAME) correct_password = secrets.compare_digest(credentials.password, PASSWORD) if not (correct_username and correct_password): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="اسم المستخدم أو كلمة المرور غير صحيحة", headers={"WWW-Authenticate": "Basic"}, ) return credentials.username # حالة الخادوم server_running = False server_thread = None server_instance = None # HTML للواجهة الفاخرة html_content = """ TzX Team - Future Coding 2999

TzX Team

SOCKS5 Server: Stopped

الأوامر:

الاتصال بالخادوم: استخدم عميل SOCKS5 على 127.0.0.1:1081
إعداد العميل: ضبط البروكسي على SOCKS5 مع العنوان 127.0.0.1 والمنفذ 1081
مراقبة السجلات: تحقق من ملف socks5.log لتتبع الاتصالات
إيقاف الخادوم: اضغط على زر "إيقاف الخادوم" أو Ctrl+C في الطرفية

""" class SOCKS5Server: def __init__(self, host='127.0.0.1', port=1081, max_connections=100): self.host = host self.port = port self.max_connections = max_connections self.server_socket = None self.running = False def setup_socket(self): self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.bind((self.host, self.port)) self.server_socket.listen(self.max_connections) console.print(Panel(f"SOCKS5 Server started on {self.host}:{self.port} | TzX Team", style="bold cyan")) async def handle_auth(self, client_socket, addr): try: loop = asyncio.get_running_loop() data = await loop.sock_recv(client_socket, 1024) if not data or data[0] != 0x05: return False nmethods = data[1] methods = data[2:2 + nmethods] if 0x00 not in methods: await loop.sock_sendall(client_socket, b'\x05\xff') return False await loop.sock_sendall(client_socket, b'\x05\x00') return True except Exception as e: logging.error(f"Client {addr}: Auth error - {e}") return False async def handle_request(self, client_socket, addr): try: loop = asyncio.get_running_loop() data = await loop.sock_recv(client_socket, 1024) if not data or data[0] != 0x05: return False cmd, _, atype = data[1:4] if cmd != 0x01: await loop.sock_sendall(client_socket, b'\x05\x07\x00\x01\x00\x00\x00\x00\x00\x00') return False if atype == 0x01: # IPv4 dest_addr = socket.inet_ntoa(data[4:8]) dest_port = struct.unpack('>H', data[8:10])[0] elif atype == 0x03: # Domain name addr_len = data[4] dest_addr = data[5:5 + addr_len].decode() dest_port = struct.unpack('>H', data[5 + addr_len:7 + addr_len])[0] else: await loop.sock_sendall(client_socket, b'\x05\x08\x00\x01\x00\x00\x00\x00\x00\x00') return False target_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: target_socket.connect((dest_addr, dest_port)) await loop.sock_sendall(client_socket, b'\x05\x00\x00\x01' + socket.inet_aton('0.0.0.0') + b'\x00\x00') return target_socket except Exception as e: await loop.sock_sendall(client_socket, b'\x05\x04\x00\x01\x00\x00\x00\x00\x00\x00') logging.error(f"Client {addr}: Connection to {dest_addr}:{dest_port} failed - {e}") return False except Exception as e: logging.error(f"Client {addr}: Request error - {e}") return False async def relay_data(self, client_socket, target_socket, addr): try: loop = asyncio.get_running_loop() while True: r, _, _ = await loop.run_in_executor(None, select.select, [client_socket, target_socket], [], []) if client_socket in r: data = await loop.sock_recv(client_socket, 4096) if not data: break await loop.sock_sendall(target_socket, data) if target_socket in r: data = await loop.sock_recv(target_socket, 4096) if not data: break await loop.sock_sendall(client_socket, data) except Exception as e: logging.error(f"Client {addr}: Relay error - {e}") finally: client_socket.close() target_socket.close() async def handle_client(self, client_socket, addr): try: if not await self.handle_auth(client_socket, addr): return target_socket = await self.handle_request(client_socket, addr) if target_socket: console.print(f"[green]Client {addr}: Connected to target | TzX Team[/green]") await self.relay_data(client_socket, target_socket, addr) else: console.print(f"[red]Client {addr}: Connection failed | TzX Team[/red]") except Exception as e: logging.error(f"Client {addr}: General error - {e}") finally: client_socket.close() async def accept_connections(self): loop = asyncio.get_running_loop() while self.running: try: client_socket, addr = await loop.sock_accept(self.server_socket) loop.create_task(self.handle_client(client_socket, addr)) except Exception as e: if self.running: logging.error(f"Accept error: {e}") def start(self): if not self.running: self.running = True self.setup_socket() asyncio.run(self.accept_connections()) def stop(self): if self.running: self.running = False if self.server_socket: self.server_socket.close() console.print("[yellow]SOCKS5 Server stopped | TzX Team[/yellow]") @app.get("/", response_class=HTMLResponse) async def root(username: str = Depends(verify_credentials)): return HTMLResponse(content=html_content) @app.post("/start_server") async def start_server(username: str = Depends(verify_credentials)): global server_running, server_thread, server_instance if not server_running: server_instance = SOCKS5Server() server_thread = threading.Thread(target=server_instance.start, daemon=True) server_thread.start() server_running = True return {"message": "SOCKS5 Server: Running | Port: 1081"} return {"message": "SOCKS5 Server: Already Running"} @app.post("/stop_server") async def stop_server(username: str = Depends(verify_credentials)): global server_running, server_thread, server_instance if server_running: server_instance.stop() server_running = False server_thread = None server_instance = None return {"message": "SOCKS5 Server: Stopped"} return {"message": "SOCKS5 Server: Already Stopped"} if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=1080, log_level="error")