add chacha20-ietf-poly1305 c version

Author: qwj

pproxy/__doc__.py

@@ -1,5 +1,5 @@
 __title__       = "pproxy"
-__version__     = "2.0.0"
+__version__     = "2.0.1"
 __license__     = "MIT"
 __description__ = "Proxy server that can tunnel among remote servers by regex rules."
 __keywords__    = "proxy socks http shadowsocks shadowsocksr ssr redirect pf tunnel cipher ssl udp"

pproxy/cipher.py

@@ -164,6 +164,19 @@ class AES_192_GCM_Cipher(AES_256_GCM_Cipher):
 class AES_128_GCM_Cipher(AES_256_GCM_Cipher):
     KEY_LENGTH = IV_LENGTH = 16
 
+class ChaCha20_IETF_POLY1305_Cipher(AEADCipher):
+    KEY_LENGTH = 32
+    IV_LENGTH = 32
+    NONCE_LENGTH = 12
+    TAG_LENGTH = 16
+    def decrypt_and_verify(self, buffer, tag):
+        return self.cipher_new(self.nonce).decrypt_and_verify(buffer, tag)
+    def encrypt_and_digest(self, buffer):
+        return self.cipher_new(self.nonce).encrypt_and_digest(buffer)
+    def setup(self):
+        from Crypto.Cipher import ChaCha20_Poly1305
+        self.cipher_new = lambda nonce: ChaCha20_Poly1305.new(key=self.key, nonce=nonce)
+
 class BF_CFB_Cipher(BaseCipher):
     KEY_LENGTH = 16
     IV_LENGTH = 8

pproxy/server.py

@@ -394,8 +394,6 @@ async def test_url(url, rserver):
     port = int(port) if port else 80 if url.scheme == 'http' else 443
     initbuf = f'GET {url.path or "/"} HTTP/1.1\r\nHost: {host_name}\r\nUser-Agent: pproxy-{__version__}\r\nConnection: close\r\n\r\n'.encode()
     for roption in rserver:
-        if roption.direct:
-            continue
         print(f'============ {roption.bind} ============')
         try:
             reader, writer = await roption.open_connection(host_name, port, None, None)
@@ -438,7 +436,7 @@ def main():
     parser.add_argument('--version', action='version', version=f'%(prog)s {__version__}')
     args = parser.parse_args()
     if args.test:
-        asyncio.run(test_url(args.test, args.rserver))
+        asyncio.get_event_loop().run_until_complete(test_url(args.test, args.rserver))
         return
     if not args.listen and not args.ulisten:
         args.listen.append(ProxyURI.compile_relay('http+socks4+socks5://:8080/'))

setup.py

@@ -37,7 +37,7 @@ def find_value(name):
     ],
     extras_require      = {
         'accelerated': [
-            'pycryptodome',
+            'pycryptodome >= 3.7.2',
         ],
     },
     install_requires    = [],

tests/api_client.py

@@ -0,0 +1,19 @@
+import asyncio
+import pproxy
+
+async def test_tcp():
+    conn = pproxy.Connection('ss://chacha20:123@127.0.0.1:12345')
+    reader, writer = await conn.tcp_connect('google.com', 80)
+    writer.write(b'GET / HTTP/1.1\r\n\r\n')
+    data = await reader.read(1024*16)
+    print(data.decode())
+
+async def test_udp():
+    conn = pproxy.Connection('ss://chacha20:123@127.0.0.1:12345')
+    answer = asyncio.Future()
+    await conn.udp_sendto('8.8.8.8', 53, b'hello', answer.set_result)
+    await answer
+    print(answer.result())
+
+asyncio.run(test_tcp())
+asyncio.run(test_udp())

tests/cipher_compare.py

@@ -0,0 +1,17 @@
+import os, time
+from pproxy.cipher import AES_256_CFB_Cipher as A
+from pproxy.cipherpy import AES_256_CFB_Cipher as B
+from pproxy.cipher import ChaCha20_Cipher as C
+from pproxy.cipherpy import ChaCha20_Cipher as D
+from pproxy.cipherpy import Camellia_256_CFB_Cipher as E
+
+TO_TEST = (A, B, C, D, E)
+
+for X in TO_TEST:
+    t = time.perf_counter()
+    for i in range(10):
+        c = X(os.urandom(X.KEY_LENGTH))
+        c.setup_iv()
+        for j in range(100):
+            c.encrypt(os.urandom(1024))
+    print(time.perf_counter()-t)

tests/cipher_speed.py

@@ -0,0 +1,30 @@
+import os, time, sys, os
+from pproxy.cipher import MAP
+from pproxy.cipherpy import MAP as MAP_PY
+
+def test_cipher(A, size=32*1024, repeat=128):
+    for i in range(repeat):
+        key = os.urandom(A.KEY_LENGTH)
+        iv = os.urandom(A.IV_LENGTH)
+        a = A(key)
+        a.setup_iv(iv)
+        s = os.urandom(size)
+        s2 = a.encrypt(s)
+        a = A(key, True)
+        a.setup_iv(iv)
+        s4 = a.decrypt(s2)
+        assert s == s4
+
+cipher = sys.argv[1] if len(sys.argv) > 1 else None
+
+if cipher and cipher.endswith('-py'):
+    A = MAP_PY.get(cipher[:-3])
+else:
+    A = MAP.get(cipher)
+if A:
+    t = time.perf_counter()
+    test_cipher(A)
+    print(cipher, time.perf_counter()-t)
+else:
+    print('unknown cipher', cipher)
+

tests/cipher_verify.py

@@ -0,0 +1,103 @@
+import os, time, sys, pickle, os
+from pproxy.cipher import MAP
+from pproxy.cipherpy import MAP as MAP_PY
+
+def test_both_cipher(A, B, size=4*1024, repeat=16):
+    print('Testing', B.__name__, '...')
+    t1 = t2 = 0
+    for i in range(repeat):
+        assert A.KEY_LENGTH == B.KEY_LENGTH and A.IV_LENGTH == B.IV_LENGTH
+        key = os.urandom(A.KEY_LENGTH)
+        iv = os.urandom(A.IV_LENGTH)
+        t = time.perf_counter()
+        a = A(key)
+        a.setup_iv(iv)
+        t1 += time.perf_counter() - t
+        t = time.perf_counter()
+        b = B(key)
+        b.setup_iv(iv)
+        t2 += time.perf_counter() - t
+        s = os.urandom(size)
+        t = time.perf_counter()
+        s2 = a.encrypt(s)
+        t1 += time.perf_counter() - t
+        t = time.perf_counter()
+        s3 = b.encrypt(s)
+        t2 += time.perf_counter() - t
+        assert s2 == s3
+
+        t = time.perf_counter()
+        a = A(key, True)
+        a.setup_iv(iv)
+        t1 += time.perf_counter() - t
+        t = time.perf_counter()
+        b = B(key, True)
+        b.setup_iv(iv)
+        t2 += time.perf_counter() - t
+        t = time.perf_counter()
+        s4 = a.decrypt(s2)
+        t1 += time.perf_counter() - t
+        t = time.perf_counter()
+        s5 = b.decrypt(s2)
+        t2 += time.perf_counter() - t
+        assert s4 == s5 == s
+
+    print('Passed', t1, t2)
+
+def test_cipher(A, data, size=4*1024, repeat=16):
+    if A.__name__ not in data:
+        if input('Correct now? (Y/n)').upper() != 'Y':
+            return
+        d = []
+        for i in range(repeat):
+            key = os.urandom(A.KEY_LENGTH)
+            iv = os.urandom(A.IV_LENGTH)
+            a = A(key)
+            a.setup_iv(iv)
+            s = os.urandom(size)
+            s2 = a.encrypt(s)
+            a = A(key, True)
+            a.setup_iv(iv)
+            s4 = a.decrypt(s2)
+            assert s == s4
+            d.append((key, iv, s, s2))
+        data[A.__name__] = d
+        print('Saved correct data')
+    else:
+        t = time.perf_counter()
+        print('Testing', A.__name__, '...')
+        for key, iv, s, s2 in data[A.__name__]:
+            a = A(key)
+            a.setup_iv(iv)
+            s3 = a.encrypt(s)
+            assert s2 == s3
+            a = A(key, True)
+            a.setup_iv(iv)
+            s4 = a.decrypt(s2)
+            assert s == s4
+        print('Passed', time.perf_counter()-t)
+
+
+cipher = sys.argv[1] if len(sys.argv) > 1 else None
+data = pickle.load(open('.cipherdata', 'rb')) if os.path.exists('.cipherdata') else {}
+
+if cipher is None:
+    print('Testing all ciphers')
+
+    for cipher, B in sorted(MAP_PY.items()):
+        A = MAP.get(cipher)
+        if A:
+            test_both_cipher(A, B)
+        elif B.__name__ in data:
+            test_cipher(B, data)
+else:
+    B = MAP_PY[cipher]
+    A = MAP.get(cipher)
+    if A:
+        test_both_cipher(A, B)
+    else:
+        test_cipher(B, data)
+
+
+pickle.dump(data, open('.cipherdata', 'wb'))
+