more cipher tests

shadowsocks/crypto/ctypes_openssl.py

@@ -151,44 +151,12 @@ ciphers = {
 
 
 def run_method(method):
-    from os import urandom
+    from shadowsocks.crypto import util
-    import random
-    import time
 
-    BLOCK_SIZE = 16384
-    rounds = 1 * 1024
-    plain = urandom(BLOCK_SIZE * rounds)
-    # import M2Crypto.EVP
-    # cipher = M2Crypto.EVP.Cipher('aes_128_cfb', 'k' * 32, 'i' * 16, 1,
-    #                key_as_bytes=0, d='md5', salt=None, i=1,
-    #                padding=1)
-    # decipher = M2Crypto.EVP.Cipher('aes_128_cfb', 'k' * 32, 'i' * 16, 0,
-    #                key_as_bytes=0, d='md5', salt=None, i=1,
-    #                padding=1)
     cipher = CtypesCrypto(method, b'k' * 32, b'i' * 16, 1)
     decipher = CtypesCrypto(method, b'k' * 32, b'i' * 16, 0)
 
-    # cipher = Salsa20Cipher('salsa20-ctr', 'k' * 32, 'i' * 8, 1)
+    util.run_cipher(cipher, decipher)
-    # decipher = Salsa20Cipher('salsa20-ctr', 'k' * 32, 'i' * 8, 1)
-    results = []
-    pos = 0
-    print('openssl test start')
-    start = time.time()
-    while pos < len(plain):
-        l = random.randint(100, 32768)
-        c = cipher.update(plain[pos:pos + l])
-        results.append(c)
-        pos += l
-    pos = 0
-    c = b''.join(results)
-    results = []
-    while pos < len(plain):
-        l = random.randint(100, 32768)
-        results.append(decipher.update(c[pos:pos + l]))
-        pos += l
-    end = time.time()
-    print('speed: %d bytes/s' % (BLOCK_SIZE * rounds / (end - start)))
-    assert b''.join(results) == plain
 
 
 def test_aes_128_cfb():

shadowsocks/crypto/m2.py

@@ -68,3 +68,32 @@ if has_m2:
     }
 else:
     ciphers = {}
+
+
+def run_method(method):
+    from shadowsocks.crypto import util
+
+    cipher = create_cipher(method, b'k' * 32, b'i' * 16, 1)
+    decipher = create_cipher(method, b'k' * 32, b'i' * 16, 0)
+
+    util.run_cipher(cipher, decipher)
+
+
+def test_aes_128_cfb():
+    run_method(b'aes-128-cfb')
+
+
+def test_aes_256_cfb():
+    run_method(b'aes-256-cfb')
+
+
+def test_bf_cfb():
+    run_method(b'bf-cfb')
+
+
+def test_rc4():
+    run_method(b'rc4')
+
+
+if __name__ == '__main__':
+    test_aes_128_cfb()

shadowsocks/crypto/rc4_md5.py

@@ -55,3 +55,16 @@ def create_cipher(alg, key, iv, op, key_as_bytes=0, d=None, salt=None,
 ciphers = {
     b'rc4-md5': (16, 16, create_cipher),
 }
+
+
+def test():
+    from shadowsocks.crypto import util
+
+    cipher = create_cipher(b'salsa20-ctr', b'k' * 32, b'i' * 8, 1)
+    decipher = create_cipher(b'salsa20-ctr', b'k' * 32, b'i' * 8, 1)
+
+    util.run_cipher(cipher, decipher)
+
+
+if __name__ == '__main__':
+    test()

shadowsocks/crypto/salsa20_ctr.py

@@ -129,40 +129,12 @@ ciphers = {
 
 
 def test():
-    from os import urandom
+    from shadowsocks.crypto import util
-    import random
-
-    rounds = 1 * 1024
-    plain = urandom(BLOCK_SIZE * rounds)
-    # import M2Crypto.EVP
-    # cipher = M2Crypto.EVP.Cipher('aes_128_cfb', 'k' * 32, 'i' * 16, 1,
-    #                key_as_bytes=0, d='md5', salt=None, i=1,
-    #                padding=1)
-    # decipher = M2Crypto.EVP.Cipher('aes_128_cfb', 'k' * 32, 'i' * 16, 0,
-    #                key_as_bytes=0, d='md5', salt=None, i=1,
-    #                padding=1)
 
     cipher = Salsa20Cipher(b'salsa20-ctr', b'k' * 32, b'i' * 8, 1)
     decipher = Salsa20Cipher(b'salsa20-ctr', b'k' * 32, b'i' * 8, 1)
-    results = []
+
-    pos = 0
+    util.run_cipher(cipher, decipher)
-    print('salsa20 test start')
-    start = time.time()
-    while pos < len(plain):
-        l = random.randint(100, 32768)
-        c = cipher.update(plain[pos:pos + l])
-        results.append(c)
-        pos += l
-    pos = 0
-    c = b''.join(results)
-    results = []
-    while pos < len(plain):
-        l = random.randint(100, 32768)
-        results.append(decipher.update(c[pos:pos + l]))
-        pos += l
-    end = time.time()
-    print('speed: %d bytes/s' % (BLOCK_SIZE * rounds / (end - start)))
-    assert b''.join(results) == plain
 
 
 if __name__ == '__main__':

shadowsocks/crypto/util.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python
+
+# Copyright (c) 2014 clowwindy
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+
+def run_cipher(cipher, decipher):
+    from os import urandom
+    import random
+    import time
+
+    BLOCK_SIZE = 16384
+    rounds = 1 * 1024
+    plain = urandom(BLOCK_SIZE * rounds)
+
+    results = []
+    pos = 0
+    print('test start')
+    start = time.time()
+    while pos < len(plain):
+        l = random.randint(100, 32768)
+        c = cipher.update(plain[pos:pos + l])
+        results.append(c)
+        pos += l
+    pos = 0
+    c = b''.join(results)
+    results = []
+    while pos < len(plain):
+        l = random.randint(100, 32768)
+        results.append(decipher.update(c[pos:pos + l]))
+        pos += l
+    end = time.time()
+    print('speed: %d bytes/s' % (BLOCK_SIZE * rounds / (end - start)))
+    assert b''.join(results) == plain