Files
amnezia-bot/awg/awg-decode.py
T
2025-02-27 23:41:43 +03:00

141 строка
4.4 KiB
Python

import sys
import struct
import zlib
import base64
import argparse
import socket
import ipaddress
import re
def qCompress(data, level=-1):
compressed = zlib.compress(data, level)
header = struct.pack('>I', len(data))
return header + compressed
def qUncompress(data):
if len(data) < 4:
return b''
uncompressed_size = struct.unpack('>I', data[:4])[0]
compressed_data = data[4:]
try:
uncompressed_data = zlib.decompress(compressed_data)
except zlib.error:
return b''
if len(uncompressed_data) != uncompressed_size:
return b''
return uncompressed_data
def base64url_encode(data):
encoded = base64.urlsafe_b64encode(data)
return encoded.rstrip(b'=')
def base64url_decode(data):
padding_needed = (4 - len(data) % 4) % 4
data += b'=' * padding_needed
return base64.urlsafe_b64decode(data)
def is_ip_address(address):
try:
ipaddress.ip_address(address)
return True
except ValueError:
return False
def resolve_dns_to_ip(dns_name):
try:
ip_address = socket.gethostbyname(dns_name)
return ip_address
except socket.gaierror:
return None
def process_conf_data(data):
def replace_endpoint(match):
full_line = match.group(0)
prefix = match.group(1)
address = match.group(2)
port = match.group(3)
suffix = match.group(4)
if not is_ip_address(address):
resolved_ip = resolve_dns_to_ip(address)
if resolved_ip:
print(f"Resolved DNS '{address}' to IP '{resolved_ip}'", file=sys.stderr)
return f"{prefix}{resolved_ip}:{port}{suffix}"
else:
print(f"Error: Could not resolve DNS name '{address}'", file=sys.stderr)
sys.exit(1)
else:
return full_line
pattern = r'^(.*Endpoint\s*=\s*)([^\s:]+)(?::(\d+))(.*)$'
return re.sub(pattern, replace_endpoint, data, flags=re.MULTILINE)
def encode(data):
data_bytes = data.encode('utf-8')
compressed = qCompress(data_bytes, level=8)
base64_encoded = base64url_encode(compressed)
s = 'vpn://' + base64_encoded.decode('ascii')
return s
def decode(s):
data = s.replace('vpn://', '')
data_bytes = data.encode('ascii')
compressed = base64url_decode(data_bytes)
uncompressed = qUncompress(compressed)
if uncompressed:
result = uncompressed
else:
result = compressed
return result.decode('utf-8')
def main():
parser = argparse.ArgumentParser(description='Encode and decode VPN configuration files to/from vpn:// format.')
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('-e', '--encode', action='store_true', help='Encode a .conf file to vpn:// format.')
group.add_argument('-d', '--decode', action='store_true', help='Decode a vpn:// string to configuration data.')
parser.add_argument('input', help='Input file for encoding or vpn:// string for decoding.')
parser.add_argument('-o', '--output', help='Output file. If not specified, output will be printed to console.')
args = parser.parse_args()
if args.encode:
try:
with open(args.input, 'r', encoding='utf-8') as f:
data = f.read()
except FileNotFoundError:
print(f'Error: File {args.input} not found.')
sys.exit(1)
except Exception as e:
print(f'Error reading file {args.input}: {e}')
sys.exit(1)
processed_data = process_conf_data(data)
encoded_string = encode(processed_data)
if args.output:
try:
with open(args.output, 'w', encoding='utf-8') as f:
f.write(encoded_string)
print(f'Encoded vpn:// string written to {args.output}')
except Exception as e:
print(f'Error writing to file {args.output}: {e}')
else:
print(encoded_string)
elif args.decode:
vpn_string = args.input
decoded_data = decode(vpn_string)
if args.output:
try:
with open(args.output, 'w', encoding='utf-8') as f:
f.write(decoded_data)
print(f'Decoded configuration data written to {args.output}')
except Exception as e:
print(f'Error writing to file {args.output}: {e}')
else:
print(decoded_data)
if __name__ == '__main__':
main()