import http.server import urllib.request import json import sqlite3 import time import threading import traceback DB_PATH = '/etc/x-ui/x-ui.db' AUTH_PORT = 2088 HY2_STATS_URL = 'http://127.0.0.1:9999/traffic?clear=1' HY2_ONLINE_URL = 'http://127.0.0.1:9999/online' HY2_KICK_URL = 'http://127.0.0.1:9999/kick' HY2_SECRET = 'oracle_traffic_secret_2026' def get_client_status(email): try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute("SELECT enable, up, down, total, expiry_time FROM client_traffics WHERE email=?;", (email,)) traf = c.fetchone() conn.close() if traf: return { 'enable': traf[0], 'up': traf[1] or 0, 'down': traf[2] or 0, 'total': traf[3] or 0, 'expiry_time': traf[4] or 0 } except Exception: pass return None def get_client_by_token(auth_str): if auth_str == 'oracle_hy2_pass_2026': return {'email': 'admin', 'total': 0, 'enable': 1, 'expiry_time': 0} try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() found_email = None # 1. Search in clients table c.execute("SELECT email, enable, sub_id, uuid, password FROM clients;") for row in c.fetchall(): email, enable, sub_id, uuid, password = row if auth_str in (sub_id, uuid, password, email): found_email = email break # 2. Search in inbounds JSON settings if not found if not found_email: c.execute("SELECT settings FROM inbounds;") for row in c.fetchall(): try: settings = json.loads(row[0]) for cl in settings.get('clients', []): if (cl.get('subId') == auth_str or cl.get('id') == auth_str or cl.get('password') == auth_str or cl.get('email') == auth_str): found_email = cl.get('email') break except Exception: pass if found_email: break conn.close() if not found_email: return None status = get_client_status(found_email) if status: return { 'email': found_email, 'enable': status['enable'], 'up': status['up'], 'down': status['down'], 'total': status['total'], 'expiry_time': status['expiry_time'] } return {'email': found_email, 'enable': 1, 'up': 0, 'down': 0, 'total': 0, 'expiry_time': 0} except Exception as e: traceback.print_exc() return None def kick_user(user_id): try: req = urllib.request.Request( HY2_KICK_URL, data=json.dumps([user_id]).encode('utf-8'), headers={'Authorization': HY2_SECRET, 'Content-Type': 'application/json'} ) urllib.request.urlopen(req, timeout=3) print(f"[HY2] Successfully kicked user: {user_id}") except Exception as e: print(f"[HY2] Error kicking user {user_id}: {e}") class Hy2AuthHandler(http.server.BaseHTTPRequestHandler): def do_POST(self): try: length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode('utf-8') data = json.loads(body) auth_str = data.get('auth', '').strip() client = get_client_by_token(auth_str) if not client: self.send_json({'ok': False, 'msg': 'User not found or invalid token'}) return email = client['email'] if email != 'admin': # Check enable if client.get('enable') == 0: self.send_json({'ok': False, 'msg': 'User disabled in 3X-UI'}) return # Check expiry now_ms = int(time.time() * 1000) if client.get('expiry_time', 0) > 0 and client['expiry_time'] < now_ms: self.send_json({'ok': False, 'msg': 'Account expired'}) return # Check total quota total = client.get('total', 0) if total > 0 and (client.get('up', 0) + client.get('down', 0)) >= total: self.send_json({'ok': False, 'msg': 'Traffic quota exceeded'}) return print(f"[HY2 Auth] Approved user: {email}") self.send_json({'ok': True, 'id': email}) except Exception as e: traceback.print_exc() self.send_json({'ok': False, 'msg': str(e)}) def send_json(self, d): content = json.dumps(d).encode('utf-8') self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(content))) self.end_headers() self.wfile.write(content) def log_message(self, format, *args): return def traffic_sync_worker(): print("[HY2 Sync] Traffic and online watcher started...") while True: time.sleep(2) # 1. Check currently online users and kick disabled/expired ones try: req = urllib.request.Request( HY2_ONLINE_URL, headers={'Authorization': HY2_SECRET} ) resp = urllib.request.urlopen(req, timeout=2).read().decode('utf-8') online = json.loads(resp) if online: for user_id in online.keys(): if user_id == 'admin': continue status = get_client_status(user_id) if not status: continue # Kick if disabled if status.get('enable') == 0: print(f"[HY2 Enforce] User {user_id} is disabled in 3X-UI. Kicking immediately...") kick_user(user_id) continue # Kick if quota exceeded total = status.get('total', 0) if total > 0 and (status.get('up', 0) + status.get('down', 0)) >= total: print(f"[HY2 Enforce] User {user_id} exceeded quota. Kicking immediately...") kick_user(user_id) continue # Kick if expired now_ms = int(time.time() * 1000) if status.get('expiry_time', 0) > 0 and status['expiry_time'] < now_ms: print(f"[HY2 Enforce] User {user_id} expired. Kicking immediately...") kick_user(user_id) continue except Exception: pass # 2. Sync traffic stats from Hysteria try: req = urllib.request.Request( HY2_STATS_URL, headers={'Authorization': HY2_SECRET} ) resp = urllib.request.urlopen(req, timeout=2).read().decode('utf-8') stats = json.loads(resp) if stats: conn = sqlite3.connect(DB_PATH) c = conn.cursor() for user_id, delta in stats.items(): if user_id == 'admin': continue delta_tx = 0 delta_rx = 0 if isinstance(delta, dict): delta_tx = delta.get('tx', 0) delta_rx = delta.get('rx', 0) elif isinstance(delta, (int, float)): delta_rx = int(delta) if delta_tx == 0 and delta_rx == 0: continue c.execute( "UPDATE client_traffics SET up = up + ?, down = down + ? WHERE email = ?;", (delta_tx, delta_rx, user_id) ) c.execute("SELECT enable, up, down, total FROM client_traffics WHERE email = ?;", (user_id,)) row = c.fetchone() if row: enable, up, down, total = row if total > 0 and (up + down) >= total and enable == 1: print(f"[HY2 Quota] User {user_id} exceeded quota! Total: {total}, Used: {up+down}. Disabling...") c.execute("UPDATE client_traffics SET enable = 0 WHERE email = ?;", (user_id,)) kick_user(user_id) conn.commit() conn.close() except Exception: pass if __name__ == '__main__': t = threading.Thread(target=traffic_sync_worker, daemon=True) t.start() server = http.server.ThreadingHTTPServer(('127.0.0.1', AUTH_PORT), Hy2AuthHandler) print(f"HY2 Bridge listening on 127.0.0.1:{AUTH_PORT}...") server.serve_forever()