#!/usr/bin/env python3 import init_paths import os import sys import json import subprocess import argparse import re import qrcode from io import StringIO from typing import Tuple, Optional, Dict, List, Any from db.database import db from paths import * def load_env_file(env_file: str) -> Dict[str, str]: env_vars = {} if os.path.exists(env_file): with open(env_file, 'r') as f: for line in f: line = line.strip() if line and not line.startswith('#') and '=' in line: key, value = line.split('=', 1) env_vars[key] = value return env_vars def load_nodes() -> List[Dict[str, str]]: if NODES_JSON_PATH.exists(): try: with NODES_JSON_PATH.open("r") as f: content = f.read() if content: return json.loads(content) except (json.JSONDecodeError, IOError): pass return [] def load_hysteria2_env() -> Dict[str, str]: return load_env_file(CONFIG_ENV) def load_hysteria2_ips() -> Tuple[str, str, str]: env_vars = load_hysteria2_env() ip4 = env_vars.get('IP4', 'None') ip6 = env_vars.get('IP6', 'None') sni = env_vars.get('SNI', '') return ip4, ip6, sni def get_singbox_domain_and_port() -> Tuple[str, str]: env_vars = load_env_file(SINGBOX_ENV) domain = env_vars.get('HYSTERIA_DOMAIN', '') port = env_vars.get('HYSTERIA_PORT', '') return domain, port def get_normalsub_domain_and_port() -> Tuple[str, str, str]: env_vars = load_env_file(NORMALSUB_ENV) domain = env_vars.get('HYSTERIA_DOMAIN', '') port = env_vars.get('HYSTERIA_PORT', '') subpath = env_vars.get('SUBPATH', '') return domain, port, subpath def is_service_active(service_name: str) -> bool: try: result = subprocess.run( ['systemctl', 'is-active', '--quiet', service_name], check=False ) return result.returncode == 0 except Exception: return False def generate_uri(username: str, auth_password: str, ip: str, port: str, obfs_password: str, sha256: str, sni: str, ip_version: int, insecure: bool, fragment_tag: str) -> str: ip_part = f"[{ip}]" if ip_version == 6 and ':' in ip else ip uri_base = f"hy2://{username}:{auth_password}@{ip_part}:{port}" params = [] if obfs_password: params.append(f"obfs=salamander&obfs-password={obfs_password}") if sha256: params.append(f"pinSHA256={sha256}") if sni: params.append(f"sni={sni}") params.append(f"insecure={'1' if insecure else '0'}") query_string = "&".join(params) return f"{uri_base}?{query_string}#{fragment_tag}" def generate_qr_code(uri: str) -> List[str]: try: qr = qrcode.QRCode( version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=1, border=2, ) qr.add_data(uri) qr.make(fit=True) output = StringIO() qr.print_ascii(out=output, invert=True) return output.getvalue().splitlines() except Exception as e: return [f"Error generating QR code: {str(e)}"] def center_text(text: str, width: int) -> str: return text.center(width) def get_terminal_width() -> int: try: return os.get_terminal_size().columns except (AttributeError, OSError): return 80 def display_uri_and_qr(uri: str, label: str, args: argparse.Namespace, terminal_width: int): if not uri: return print(f"\n{label}:\n{uri}\n") if args.qrcode: print(f"{label} QR Code:\n") qr_code = generate_qr_code(uri) for line in qr_code: print(center_text(line, terminal_width)) def show_uri(args: argparse.Namespace) -> None: if db is None: print("\033[0;31mError:\033[0m Database connection failed.") return if not is_service_active("hysteria-server.service"): print("\033[0;31mError:\033[0m Hysteria2 is not active.") return try: with open(CONFIG_FILE, 'r') as f: config = json.load(f) except (FileNotFoundError, json.JSONDecodeError) as e: print(f"\033[0;31mError:\033[0m Could not load config file {CONFIG_FILE}. Details: {e}") return user_doc = db.get_user(args.username) if not user_doc: print(f"\033[0;31mError:\033[0m User '{args.username}' not found in the database.") return auth_password = user_doc["password"] port = config["listen"].split(":")[-1] sha256 = config.get("tls", {}).get("pinSHA256", "") obfs_password = config.get("obfs", {}).get("salamander", {}).get("password", "") insecure = config.get("tls", {}).get("insecure", True) ip4, ip6, sni = load_hysteria2_ips() nodes = load_nodes() terminal_width = get_terminal_width() if args.all or args.ip_version == 4: if ip4 and ip4 != "None": uri = generate_uri(args.username, auth_password, ip4, port, obfs_password, sha256, sni, 4, insecure, f"{args.username}-IPv4") display_uri_and_qr(uri, "IPv4", args, terminal_width) if args.all or args.ip_version == 6: if ip6 and ip6 != "None": uri = generate_uri(args.username, auth_password, ip6, port, obfs_password, sha256, sni, 6, insecure, f"{args.username}-IPv6") display_uri_and_qr(uri, "IPv6", args, terminal_width) for node in nodes: node_name = node.get("name") node_ip = node.get("ip") if not node_name or not node_ip: continue ip_v = 4 if '.' in node_ip else 6 if args.all or args.ip_version == ip_v: uri = generate_uri(args.username, auth_password, node_ip, port, obfs_password, sha256, sni, ip_v, insecure, f"{args.username}-{node_name}") display_uri_and_qr(uri, f"Node: {node_name} (IPv{ip_v})", args, terminal_width) if args.singbox and is_service_active("hysteria-singbox.service"): domain, port = get_singbox_domain_and_port() if domain and port: print(f"\nSingbox Sublink:\nhttps://{domain}:{port}/sub/singbox/{args.username}/{args.ip_version}#{args.username}\n") if args.normalsub and is_service_active("hysteria-normal-sub.service"): domain, port, subpath = get_normalsub_domain_and_port() if domain and port: print(f"\nNormal-SUB Sublink:\nhttps://{domain}:{port}/{subpath}/sub/normal/{auth_password}#Hysteria2\n") def main(): parser = argparse.ArgumentParser(description="Hysteria2 URI Generator") parser.add_argument("-u", "--username", help="Username to generate URI for") parser.add_argument("-qr", "--qrcode", action="store_true", help="Generate QR code") parser.add_argument("-ip", "--ip-version", type=int, default=4, choices=[4, 6], help="IP version (4 or 6)") parser.add_argument("-a", "--all", action="store_true", help="Show all available IPs") parser.add_argument("-s", "--singbox", action="store_true", help="Generate SingBox sublink") parser.add_argument("-n", "--normalsub", action="store_true", help="Generate Normal-SUB sublink") args = parser.parse_args() if not args.username: parser.print_help() sys.exit(1) show_uri(args) if __name__ == "__main__": main()