From a7dc63bf9d65a66ea7aaa8e7727ab6b09d02b655 Mon Sep 17 00:00:00 2001 From: dnomd343 Date: Thu, 28 Jul 2022 15:49:15 +0800 Subject: [PATCH] style: CRLF to LF --- Basis/Functions.py | 130 +++++++++---------- ProxyTester/VMess.py | 300 +++++++++++++++++++++---------------------- 2 files changed, 215 insertions(+), 215 deletions(-) diff --git a/Basis/Functions.py b/Basis/Functions.py index 744d1a3..2e6e141 100644 --- a/Basis/Functions.py +++ b/Basis/Functions.py @@ -1,65 +1,65 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -import time -import psutil -import random -from Basis.Logger import logging - - -def genFlag(length: int = 12) -> str: # generate random task flag - flag = '' - for i in range(0, length): - tmp = random.randint(0, 15) - if tmp >= 10: - flag += chr(tmp + 87) # a ~ f - else: - flag += str(tmp) # 0 ~ 9 - logging.debug('generate new flag -> ' + flag) - return flag - - -def getAvailablePort(rangeStart: int = 41952, rangeEnd: int = 65535) -> int: # get a available port - if rangeStart > rangeEnd or rangeStart < 1 or rangeEnd > 65535: - raise RuntimeError('invalid port range') - while True: - port = random.randint(rangeStart, rangeEnd) # choose randomly - if checkPortStatus(port): - logging.debug('get new port -> %i' % port) - return port - time.sleep(0.1) # wait for 100ms - - -def checkPortStatus(port: int) -> bool: # check if the port is occupied - logging.debug('check status of port %i -> available' % port) - for connection in networkStatus(): # scan every connections - if connection['local']['port'] == port: # port occupied (whatever ipv4-tcp / ipv4-udp / ipv6-tcp / ipv6-udp) - logging.debug('check status of port %i -> occupied' % port) - return False - return True - - -def networkStatus() -> list: # get all network connections - result = [] - for connection in psutil.net_connections(): - if not connection.family.name.startswith('AF_INET'): # AF_INET / AF_INET6 - continue - if connection.type.name not in ['SOCK_STREAM', 'SOCK_DGRAM']: # TCP / UDP - continue - result.append({ - 'fd': connection.fd, - 'family': 'ipv6' if connection.family.name[-1] == '6' else 'ipv4', # ip version - 'type': 'tcp' if connection.type.name == 'SOCK_STREAM' else 'udp', # tcp or udp - 'local': { # local bind address - 'addr': connection.laddr.ip, - 'port': connection.laddr.port, - }, - 'remote': { # remote address - 'addr': connection.raddr.ip, - 'port': connection.raddr.port, - } if len(connection.raddr) != 0 else None, - 'status': connection.status, - 'pid': connection.pid, # process id - }) - logging.debug('get network status -> found %i connections' % len(result)) - return result +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import time +import psutil +import random +from Basis.Logger import logging + + +def genFlag(length: int = 12) -> str: # generate random task flag + flag = '' + for i in range(0, length): + tmp = random.randint(0, 15) + if tmp >= 10: + flag += chr(tmp + 87) # a ~ f + else: + flag += str(tmp) # 0 ~ 9 + logging.debug('generate new flag -> ' + flag) + return flag + + +def getAvailablePort(rangeStart: int = 41952, rangeEnd: int = 65535) -> int: # get a available port + if rangeStart > rangeEnd or rangeStart < 1 or rangeEnd > 65535: + raise RuntimeError('invalid port range') + while True: + port = random.randint(rangeStart, rangeEnd) # choose randomly + if checkPortStatus(port): + logging.debug('get new port -> %i' % port) + return port + time.sleep(0.1) # wait for 100ms + + +def checkPortStatus(port: int) -> bool: # check if the port is occupied + logging.debug('check status of port %i -> available' % port) + for connection in networkStatus(): # scan every connections + if connection['local']['port'] == port: # port occupied (whatever ipv4-tcp / ipv4-udp / ipv6-tcp / ipv6-udp) + logging.debug('check status of port %i -> occupied' % port) + return False + return True + + +def networkStatus() -> list: # get all network connections + result = [] + for connection in psutil.net_connections(): + if not connection.family.name.startswith('AF_INET'): # AF_INET / AF_INET6 + continue + if connection.type.name not in ['SOCK_STREAM', 'SOCK_DGRAM']: # TCP / UDP + continue + result.append({ + 'fd': connection.fd, + 'family': 'ipv6' if connection.family.name[-1] == '6' else 'ipv4', # ip version + 'type': 'tcp' if connection.type.name == 'SOCK_STREAM' else 'udp', # tcp or udp + 'local': { # local bind address + 'addr': connection.laddr.ip, + 'port': connection.laddr.port, + }, + 'remote': { # remote address + 'addr': connection.raddr.ip, + 'port': connection.raddr.port, + } if len(connection.raddr) != 0 else None, + 'status': connection.status, + 'pid': connection.pid, # process id + }) + logging.debug('get network status -> found %i connections' % len(result)) + return result diff --git a/ProxyTester/VMess.py b/ProxyTester/VMess.py index 55982f4..97d2467 100644 --- a/ProxyTester/VMess.py +++ b/ProxyTester/VMess.py @@ -1,150 +1,150 @@ -#!/usr/bin/python -# -*- coding:utf-8 -*- - -from ProxyTester import V2ray - -testConfig = {} - -vmessMethodList = [ - 'aes-128-gcm', - 'chacha20-poly1305', - 'auto', - 'none', - 'zero', -] - -def vmessBasicTest(method: str, alterId: int) -> dict: - inboundConfig = { - 'protocol': 'vmess', - 'listen': testConfig['bind'], - 'port': testConfig['port'], - 'settings': { - 'clients': [ - { - 'id': testConfig['id'], - 'alterId': alterId - } - ] - } - } - - caption = 'VMess method ' + method - if alterId == 0: - envVar = {} - caption += ' (AEAD)' - else: - envVar = { - 'v2ray.vmess.aead.forced': 'false' - } - caption += ' (alterId ' + str(alterId) + ')' - - return { - 'caption': caption, - 'proxy': { - 'type': 'vmess', - 'server': testConfig['addr'], - 'port': testConfig['port'], - 'method': method, - 'id': testConfig['id'], - 'aid': alterId - }, - 'server': { - 'startCommand': ['v2ray', '-c', testConfig['file']], - 'fileContent': V2ray.v2rayConfig(inboundConfig), - 'filePath': testConfig['file'], - 'envVar': envVar - }, - 'aider': None - } - -def loadVmessStream(streamInfo: dict) -> dict: - proxyInfo = { - 'type': 'vmess', - 'server': testConfig['addr'], - 'port': testConfig['port'], - 'id': testConfig['id'], - 'stream': streamInfo['client'] - } - inboundConfig = { - 'protocol': 'vmess', - 'listen': testConfig['bind'], - 'port': testConfig['port'], - 'settings': { - 'clients': [ - { - 'id': testConfig['id'] - } - ] - }, - 'streamSettings': streamInfo['server'] - } - return { - 'caption': 'VMess network ' + streamInfo['caption'], - 'proxy': proxyInfo, - 'server': { - 'startCommand': ['v2ray', '-c', testConfig['file']], - 'fileContent': V2ray.v2rayConfig(inboundConfig), - 'filePath': testConfig['file'], - 'envVar': {} - }, - 'aider': None - } - -def test(config: dict) -> list: - global testConfig - testConfig = config - testList = [] - - # Basic test - for method in vmessMethodList: # methods and AEAD/MD5+AES test - testList.append(vmessBasicTest(method, 0)) - testList.append(vmessBasicTest(method, 64)) - - # TCP stream - streamInfo = V2ray.loadTcpStream(False, '', '') - testList.append(loadVmessStream(streamInfo)) - streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) - testList.append(loadVmessStream(streamInfo)) - - streamInfo = V2ray.loadTcpStream(True, config['host'], '/') - testList.append(loadVmessStream(streamInfo)) - streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) - testList.append(loadVmessStream(streamInfo)) - - # mKCP stream - for obfs in V2ray.udpObfsList: - streamInfo = V2ray.loadKcpStream(config['passwd'], obfs) - testList.append(loadVmessStream(streamInfo)) - streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) - testList.append(loadVmessStream(streamInfo)) - - # WebSocket stream - streamInfo = V2ray.loadWsStream(config['host'], config['path'], False) - testList.append(loadVmessStream(streamInfo)) - streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) - testList.append(loadVmessStream(streamInfo)) - - streamInfo = V2ray.loadWsStream(config['host'], config['path'], True) - testList.append(loadVmessStream(streamInfo)) - streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) - testList.append(loadVmessStream(streamInfo)) - - # HTTP/2 stream - streamInfo = V2ray.loadH2Stream(config['host'], config['path']) - streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) - testList.append(loadVmessStream(streamInfo)) - - # QUIC stream - for method in V2ray.quicMethodList: - for obfs in V2ray.udpObfsList: - streamInfo = V2ray.loadQuicStream(method, config['passwd'], obfs) - streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) - testList.append(loadVmessStream(streamInfo)) - - # GRPC stream - streamInfo = V2ray.loadGrpcStream(config['service']) - testList.append(loadVmessStream(streamInfo)) - streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) - testList.append(loadVmessStream(streamInfo)) - - return testList +#!/usr/bin/python +# -*- coding:utf-8 -*- + +from ProxyTester import V2ray + +testConfig = {} + +vmessMethodList = [ + 'aes-128-gcm', + 'chacha20-poly1305', + 'auto', + 'none', + 'zero', +] + +def vmessBasicTest(method: str, alterId: int) -> dict: + inboundConfig = { + 'protocol': 'vmess', + 'listen': testConfig['bind'], + 'port': testConfig['port'], + 'settings': { + 'clients': [ + { + 'id': testConfig['id'], + 'alterId': alterId + } + ] + } + } + + caption = 'VMess method ' + method + if alterId == 0: + envVar = {} + caption += ' (AEAD)' + else: + envVar = { + 'v2ray.vmess.aead.forced': 'false' + } + caption += ' (alterId ' + str(alterId) + ')' + + return { + 'caption': caption, + 'proxy': { + 'type': 'vmess', + 'server': testConfig['addr'], + 'port': testConfig['port'], + 'method': method, + 'id': testConfig['id'], + 'aid': alterId + }, + 'server': { + 'startCommand': ['v2ray', '-c', testConfig['file']], + 'fileContent': V2ray.v2rayConfig(inboundConfig), + 'filePath': testConfig['file'], + 'envVar': envVar + }, + 'aider': None + } + +def loadVmessStream(streamInfo: dict) -> dict: + proxyInfo = { + 'type': 'vmess', + 'server': testConfig['addr'], + 'port': testConfig['port'], + 'id': testConfig['id'], + 'stream': streamInfo['client'] + } + inboundConfig = { + 'protocol': 'vmess', + 'listen': testConfig['bind'], + 'port': testConfig['port'], + 'settings': { + 'clients': [ + { + 'id': testConfig['id'] + } + ] + }, + 'streamSettings': streamInfo['server'] + } + return { + 'caption': 'VMess network ' + streamInfo['caption'], + 'proxy': proxyInfo, + 'server': { + 'startCommand': ['v2ray', '-c', testConfig['file']], + 'fileContent': V2ray.v2rayConfig(inboundConfig), + 'filePath': testConfig['file'], + 'envVar': {} + }, + 'aider': None + } + +def test(config: dict) -> list: + global testConfig + testConfig = config + testList = [] + + # Basic test + for method in vmessMethodList: # methods and AEAD/MD5+AES test + testList.append(vmessBasicTest(method, 0)) + testList.append(vmessBasicTest(method, 64)) + + # TCP stream + streamInfo = V2ray.loadTcpStream(False, '', '') + testList.append(loadVmessStream(streamInfo)) + streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) + testList.append(loadVmessStream(streamInfo)) + + streamInfo = V2ray.loadTcpStream(True, config['host'], '/') + testList.append(loadVmessStream(streamInfo)) + streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) + testList.append(loadVmessStream(streamInfo)) + + # mKCP stream + for obfs in V2ray.udpObfsList: + streamInfo = V2ray.loadKcpStream(config['passwd'], obfs) + testList.append(loadVmessStream(streamInfo)) + streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) + testList.append(loadVmessStream(streamInfo)) + + # WebSocket stream + streamInfo = V2ray.loadWsStream(config['host'], config['path'], False) + testList.append(loadVmessStream(streamInfo)) + streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) + testList.append(loadVmessStream(streamInfo)) + + streamInfo = V2ray.loadWsStream(config['host'], config['path'], True) + testList.append(loadVmessStream(streamInfo)) + streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) + testList.append(loadVmessStream(streamInfo)) + + # HTTP/2 stream + streamInfo = V2ray.loadH2Stream(config['host'], config['path']) + streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) + testList.append(loadVmessStream(streamInfo)) + + # QUIC stream + for method in V2ray.quicMethodList: + for obfs in V2ray.udpObfsList: + streamInfo = V2ray.loadQuicStream(method, config['passwd'], obfs) + streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) + testList.append(loadVmessStream(streamInfo)) + + # GRPC stream + streamInfo = V2ray.loadGrpcStream(config['service']) + testList.append(loadVmessStream(streamInfo)) + streamInfo = V2ray.addSecureConfig(streamInfo, config['cert'], config['key'], config['host']) + testList.append(loadVmessStream(streamInfo)) + + return testList