update: enhance ipv6 functions

Basis/Functions.py

@@ -5,6 +5,7 @@ import time
 import psutil
 import random
 import hashlib
+from IPy import IP
 from Basis.Logger import logging
 
 
@@ -12,6 +13,13 @@ def md5Sum(data: str, encode: str = 'UTF-8') -> str:
     return hashlib.md5(data.encode(encoding = encode)).hexdigest()
 
 
+def ipFormat(ipAddr: str, v6Bracket: bool = False) -> str:
+    ip = IP(ipAddr)
+    if v6Bracket and ip.version() == 6:
+        return '[%s]' % str(ip)  # [IPv6]
+    return str(ip)  # IPv4 / IPV6
+
+
 def genFlag(length: int = 12) -> str:  # generate random task flag
     flag = ''
     for i in range(0, length):

Dockerfile

@@ -465,5 +465,5 @@ COPY --from=dnsproxy /tmp/dnsproxy /asset/usr/bin/
 FROM python:3.10-alpine3.16
 RUN \
   apk add --no-cache boost-program_options c-ares glib libev libsodium libstdc++ mbedtls pcre && \
-  pip3 --no-cache-dir install colorlog pysocks requests
+  pip3 --no-cache-dir install colorlog pysocks requests IPy
 COPY --from=asset /asset /

Tester/Plugin.py

@@ -7,18 +7,24 @@ import json
 from Basis.Logger import logging
 from Basis.Process import Process
 from Basis.Methods import plugins
-from Basis.Functions import genFlag, getAvailablePort
+from Basis.Functions import genFlag
+from Basis.Functions import ipFormat
+from Basis.Functions import getAvailablePort
 
 settings = {
     'serverBind': '127.0.0.1',
     'workDir': '/tmp/ProxyC',
+    'site': 'www.bing.com',
+    'host': '343.re',
+    'cert': '/etc/ssl/certs/343.re/fullchain.pem',
+    'key': '/etc/ssl/certs/343.re/privkey.pem',
 }
 
 pluginParams = {
-    'SITE': 'www.bing.com',
+    'SITE': settings['site'],
-    'HOST': '343.re',
+    'HOST': settings['host'],
-    'CERT': '/etc/ssl/certs/343.re/fullchain.pem',
+    'CERT': settings['cert'],
-    'KEY': '/etc/ssl/certs/343.re/privkey.pem',
+    'KEY': settings['key'],
     'PATH': '/' + genFlag(length = 6),
     'PASSWD': genFlag(length = 8),
 }
@@ -270,7 +276,7 @@ def ssInject(server: Process, pluginInfo: dict) -> Process:
         ssConfig = json.loads(server.file[0]['content'])  # modify origin config
         ssConfig.pop('plugin')  # remove plugin option
         ssConfig.pop('plugin_opts')
-        rabbitBind = ('[%s]' if ':' in ssConfig['server'] else '%s') % ssConfig['server']  # ipv4 / [ipv6]
+        rabbitBind = ipFormat(ssConfig['server'], v6Bracket = True)  # ipv4 / [ipv6]
         rabbitPort = ssConfig['server_port']
         ssConfig['server'] = '127.0.0.1'  # SIP003 use ipv4 localhost for communication
         ssConfig['server_port'] = int(pluginInfo['server']['param'])  # aka ${RABBIT_PORT}

Tester/Shadowsocks.py

@@ -15,7 +15,9 @@ from Basis.Functions import genFlag, getAvailablePort
 
 settings = {
     'serverBind': '127.0.0.1',
-    'clientBind': '127.0.0.1',  # aka socks5 address
+    'clientBind': '127.0.0.1',
+    # 'serverBind': '::1',
+    # 'clientBind': '::1',
     'workDir': '/tmp/ProxyC'
 }
 

Tester/ShadowsocksR.py

@@ -11,7 +11,9 @@ from Basis.Methods import ssrMethods, ssrProtocols, ssrObfuscations
 
 settings = {
     'serverBind': '127.0.0.1',
-    'clientBind': '127.0.0.1',  # aka socks5 address
+    'clientBind': '127.0.0.1',
+    # 'serverBind': '::1',
+    # 'clientBind': '::1',
     'workDir': '/tmp/ProxyC'
 }
 

test.py

@@ -7,6 +7,7 @@ from threading import Thread
 from Tester import Shadowsocks
 from Tester import ShadowsocksR
 from Basis.Logger import logging
+from Basis.Functions import ipFormat
 from Basis.Functions import checkPortStatus
 
 
@@ -31,19 +32,23 @@ def test(testObj: dict) -> None:
     time.sleep(1)
 
     errFlag = False
+    socks5 = '%s:%i' % (
+        ipFormat(testObj['socks']['addr'], v6Bracket = True),
+        testObj['socks']['port']
+    )
     try:
         request = requests.get(
             'http://iserv.scutbot.cn',
             proxies = {
-                'http': 'socks5://127.0.0.1:%i' % testObj['socks']['port'],
+                'http': 'socks5://' + socks5,
-                'https': 'socks5://127.0.0.1:%i' % testObj['socks']['port'],
+                'https': 'socks5://' + socks5,
             },
             timeout = 10
         )
         request.raise_for_status()
-        logging.info('socks5 127.0.0.1:%i -> ok' % testObj['socks']['port'])
+        logging.info('socks5 %s -> ok' % socks5)
     except Exception as exp:
-        logging.error('socks5 127.0.0.1:%i -> error' % testObj['socks']['port'])
+        logging.error('socks5 %s -> error' % socks5)
         logging.error('requests exception\n' + str(exp))
         errFlag = True
 
@@ -84,6 +89,7 @@ def runTest(testIter: iter, threadNum: int):
 
 
 ss = Shadowsocks.load(isExtra = True)
+# ss = Shadowsocks.load(isExtra = False)
 ssr = ShadowsocksR.load()
 logging.critical('test start')