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")
