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

feat: Enhance v2rayNG compatibility for pinSHA256 conversion

Browse Source 
Converting the pinSHA256 value from Base64 to colon-separated hexadecimal format when a v2rayNG User-Agent is detected.
This commit is contained in:
Whispering Wind
2025-02-13 20:54:39 +03:30
committed by GitHub
parent 66fd6bf1d2
commit b307ff7675
1 changed files with 19 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4,7 +4,9 @@ import subprocess
import time import time
import re import re
import shlex import shlex
import json # Import the json module import json
import base64
import hashlib
from aiohttp import web from aiohttp import web
from aiohttp.web_middlewares import middleware from aiohttp.web_middlewares import middleware
from dotenv import load_dotenv from dotenv import load_dotenv
@ -51,8 +53,9 @@ async def handle(request):
if not username: if not username:
return web.Response(status=400, text="Error: Missing 'username' parameter.") return web.Response(status=400, text="Error: Missing 'username' parameter.")
uri_output = get_user_uri(username) user_agent = request.headers.get('User-Agent', '').lower()
uri_output = get_user_uri(username, user_agent)
return web.Response(text=uri_output, content_type='text/plain') return web.Response(text=uri_output, content_type='text/plain')
except ValueError as e: except ValueError as e:
@ -61,7 +64,7 @@ async def handle(request):
print(f"Internal Server Error: {str(e)}") print(f"Internal Server Error: {str(e)}")
return web.Response(status=500, text="Error: Internal server error.") return web.Response(status=500, text="Error: Internal server error.")
def get_user_uri(username): def get_user_uri(username, user_agent):
try: try:
user_info_command = [ user_info_command = [
'python3', 'python3',
@ -72,8 +75,6 @@ def get_user_uri(username):
safe_user_info_command = [shlex.quote(arg) for arg in user_info_command] safe_user_info_command = [shlex.quote(arg) for arg in user_info_command]
user_info_output = subprocess.check_output(safe_user_info_command).decode() user_info_output = subprocess.check_output(safe_user_info_command).decode()
user_info = json.loads(user_info_output) user_info = json.loads(user_info_output)
upload = user_info.get('upload_bytes', 0) upload = user_info.get('upload_bytes', 0)
download = user_info.get('download_bytes', 0) download = user_info.get('download_bytes', 0)
total = user_info.get('max_download_bytes', 0) total = user_info.get('max_download_bytes', 0)
@ -101,6 +102,18 @@ def get_user_uri(username):
output = re.sub(r'IPv4:\s*', '', output) output = re.sub(r'IPv4:\s*', '', output)
output = re.sub(r'IPv6:\s*', '', output) output = re.sub(r'IPv6:\s*', '', output)
if "v2ray" in user_agent and "ng" in user_agent:
match = re.search(r'pinSHA256=sha256/([^&]+)', output)
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])
output = output.replace(f'pinSHA256=sha256/{base64_pin}', f'pinSHA256={hex_pin}')
except Exception as e:
print(f"Error processing pinSHA256: {e}")
subscription_info = ( subscription_info = (
f"//subscription-userinfo: upload={upload}; download={download}; total={total}; expire={expiration_timestamp}\n" f"//subscription-userinfo: upload={upload}; download={download}; total={total}; expire={expiration_timestamp}\n"
) )