475d511345
Refactored normalsub.py to use the user's password as the identifier in the subscription URL path instead of the username, enhancing privacy by making user enumeration harder.
218 lines
8.0 KiB
Python
218 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
|
|
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 init_paths import *
|
|
from paths import *
|
|
|
|
def load_env_file(env_file: str) -> Dict[str, str]:
|
|
"""Load environment variables from a file into a dictionary."""
|
|
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_hysteria2_env() -> Dict[str, str]:
|
|
"""Load Hysteria2 environment variables."""
|
|
return load_env_file(CONFIG_ENV)
|
|
|
|
def load_hysteria2_ips() -> Tuple[str, str, str]:
|
|
"""Load Hysteria2 IPv4 and IPv6 addresses from environment."""
|
|
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]:
|
|
"""Get domain and port from SingBox config."""
|
|
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]:
|
|
"""Get domain, port, and subpath from Normal-SUB config."""
|
|
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:
|
|
"""Check if a systemd service is active."""
|
|
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) -> str:
|
|
"""Generate Hysteria2 URI for the given parameters."""
|
|
uri_base = f"hy2://{username}%3A{auth_password}@{ip}:{port}"
|
|
|
|
if ip_version == 6 and re.match(r'^[0-9a-fA-F:]+$', ip):
|
|
uri_base = f"hy2://{username}%3A{auth_password}@[{ip}]:{port}"
|
|
|
|
params = []
|
|
|
|
if obfs_password:
|
|
params.append(f"obfs=salamander&obfs-password={obfs_password}")
|
|
|
|
if sha256:
|
|
params.append(f"pinSHA256={sha256}")
|
|
|
|
insecure_value = "1" if insecure else "0"
|
|
params.append(f"insecure={insecure_value}&sni={sni}")
|
|
|
|
params_str = "&".join(params)
|
|
return f"{uri_base}?{params_str}#{username}-IPv{ip_version}"
|
|
|
|
def generate_qr_code(uri: str) -> List[str]:
|
|
"""Generate terminal-friendly ASCII QR code using pure Python."""
|
|
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:
|
|
"""Center text in the given width."""
|
|
return text.center(width)
|
|
|
|
def get_terminal_width() -> int:
|
|
"""Get terminal width."""
|
|
try:
|
|
return os.get_terminal_size().columns
|
|
except (AttributeError, OSError):
|
|
return 80
|
|
|
|
def show_uri(args: argparse.Namespace) -> None:
|
|
"""Show URI and optional QR codes for the given username."""
|
|
if not os.path.exists(USERS_FILE):
|
|
print(f"\033[0;31mError:\033[0m Config file {USERS_FILE} not found.")
|
|
return
|
|
|
|
if not is_service_active("hysteria-server.service"):
|
|
print("\033[0;31mError:\033[0m Hysteria2 is not active.")
|
|
return
|
|
|
|
with open(CONFIG_FILE, 'r') as f:
|
|
config = json.load(f)
|
|
|
|
with open(USERS_FILE, 'r') as f:
|
|
users = json.load(f)
|
|
|
|
if args.username not in users:
|
|
print("Invalid username. Please try again.")
|
|
return
|
|
|
|
auth_password = users[args.username]["password"]
|
|
port = config["listen"].split(":")[1] if ":" in config["listen"] else config["listen"]
|
|
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()
|
|
available_ip4 = ip4 and ip4 != "None"
|
|
available_ip6 = ip6 and ip6 != "None"
|
|
|
|
uri_ipv4 = None
|
|
uri_ipv6 = None
|
|
|
|
if args.all:
|
|
if available_ip4:
|
|
uri_ipv4 = generate_uri(args.username, auth_password, ip4, port,
|
|
obfs_password, sha256, sni, 4, insecure)
|
|
print(f"\nIPv4:\n{uri_ipv4}\n")
|
|
|
|
if available_ip6:
|
|
uri_ipv6 = generate_uri(args.username, auth_password, ip6, port,
|
|
obfs_password, sha256, sni, 6, insecure)
|
|
print(f"\nIPv6:\n{uri_ipv6}\n")
|
|
else:
|
|
if args.ip_version == 4 and available_ip4:
|
|
uri_ipv4 = generate_uri(args.username, auth_password, ip4, port,
|
|
obfs_password, sha256, sni, 4, insecure)
|
|
print(f"\nIPv4:\n{uri_ipv4}\n")
|
|
elif args.ip_version == 6 and available_ip6:
|
|
uri_ipv6 = generate_uri(args.username, auth_password, ip6, port,
|
|
obfs_password, sha256, sni, 6, insecure)
|
|
print(f"\nIPv6:\n{uri_ipv6}\n")
|
|
else:
|
|
print("Invalid IP version or no available IP for the requested version.")
|
|
return
|
|
|
|
if args.qrcode:
|
|
terminal_width = get_terminal_width()
|
|
|
|
if uri_ipv4:
|
|
qr_code = generate_qr_code(uri_ipv4)
|
|
print("\nIPv4 QR Code:\n")
|
|
for line in qr_code:
|
|
print(center_text(line, terminal_width))
|
|
|
|
if uri_ipv6:
|
|
qr_code = generate_qr_code(uri_ipv6)
|
|
print("\nIPv6 QR Code:\n")
|
|
for line in qr_code:
|
|
print(center_text(line, 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():
|
|
"""Main function to parse arguments and show URIs."""
|
|
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() |