bivashy/Blitz-Proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix: Include both IPv4 and IPv6 URIs in normalsub output

Browse Source
This commit is contained in:
Whispering Wind
2025-02-22 13:31:02 +03:30
committed by GitHub
parent 54aa154465
commit f0697d56e6
1 changed files with 25 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
core/scripts/normalsub/normalsub.py
View File

@ -135,7 +135,6 @@ async def get_template_context(username, user_agent):
ipv4_uri, ipv6_uri = get_uris(username)
sub_link = f"https://{DOMAIN}:{PORT}/sub/normal/{username}"
# Generate QR codes only if URIs are available
ipv4_qrcode = generate_qrcode_base64(ipv4_uri) if ipv4_uri else None
ipv6_qrcode = generate_qrcode_base64(ipv6_uri) if ipv6_uri else None
sublink_qrcode = generate_qrcode_base64(sub_link)
@ -187,7 +186,7 @@ def get_user_info(username):
def get_user_uri(username, user_agent):
"""
Returns the URI for the user, adapting the output based on the User-Agent.
Returns the URI for the user, adapting the output based on the User-Agent. Returns BOTH IPv4 and IPv6.
"""
try:
user_info = get_user_info(username)
@ -205,33 +204,41 @@ def get_user_uri(username, user_agent):
else:
expiration_timestamp = 0
# Get URI
ipv4_uri, ipv6_uri = get_uris(username)
# Choose the appropriate URI based on availability. Prioritize IPv6 if available.
output_uri = ipv6_uri if ipv6_uri else (ipv4_uri if ipv4_uri else "No URI available")
output_lines = []
if ipv4_uri:
output_lines.append(ipv4_uri)
if ipv6_uri:
output_lines.append(ipv6_uri)
if "v2ray" in user_agent and "ng" in user_agent:
match = re.search(r'pinSHA256=sha256/([^&]+)', output_uri)
if match:
base64_pin = match.group(1)
try:
decoded_pin = base64.b64decode(base64_pin)
hex_pin = ':'.join(['{:02X}'.format(byte) for byte in decoded_pin])
if not output_lines:
return "No URI available"
processed_uris = []
for uri in output_lines:
if "v2ray" in user_agent and "ng" in user_agent:
match = re.search(r'pinSHA256=sha256/([^&]+)', uri)
if match:
base64_pin = match.group(1)
try:
decoded_pin = base64.b64decode(base64_pin)
hex_pin = ':'.join(['{:02X}'.format(byte) for byte in decoded_pin])
uri = uri.replace(f'pinSHA256=sha256/{base64_pin}', f'pinSHA256={hex_pin}')
except Exception as e:
print(f"Error processing pinSHA256: {e}")
processed_uris.append(uri)
output_uri = output_uri.replace(f'pinSHA256=sha256/{base64_pin}', f'pinSHA256={hex_pin}')
except Exception as e:
print(f"Error processing pinSHA256: {e}")
subscription_info = (
f"//subscription-userinfo: upload={upload}; download={download}; total={total}; expire={expiration_timestamp}\n"
)
profile_lines = f"//profile-title: {username}-Hysteria2 🚀\n//profile-update-interval: 1\n"
output = profile_lines + subscription_info + output_uri
output = profile_lines + subscription_info + "\n".join(processed_uris)
return output
except subprocess.CalledProcessError:
raise RuntimeError("Failed to get URI or user info.")
except json.JSONDecodeError:
@ -254,7 +261,6 @@ def get_uris(username):
safe_command = [shlex.quote(arg) for arg in command]
output = subprocess.check_output(safe_command).decode().strip()
# Use regex to find IPv4 and IPv6 URIs, handling cases where they might not exist.
ipv4_match = re.search(r'IPv4:\s*(.*)', output)
ipv6_match = re.search(r'IPv6:\s*(.*)', output)
@ -286,4 +292,4 @@ if __name__ == '__main__':
ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
ssl_context.set_ciphers('AES256+EECDH:AES256+EDH')
web.run_app(app, port=PORT, ssl_context=ssl_context)
web.run_app(app, port=PORT, ssl_context=ssl_context)