Turn it into a Python module

Author: djosix

.gitignore

@@ -0,0 +1,6 @@
+dist/
+build/
+.vscode/
+.DS_Store
+*.pyc
+*.egg-info

README.md

@@ -1,6 +1,20 @@
 # padding_oracle.py
 
-Extremely fast threaded [padding oracle](http://server.maojui.me/Crypto/Padding_oracle_attack/) automation script for Python >= 3.7.
+Extremely fast threaded [padding oracle](http://server.maojui.me/Crypto/Padding_oracle_attack/) automation script for Python 3.
+
+## Install
+
+Installing from PyPI:
+
+```shell
+pip3 install -U padding_oracle
+```
+
+Or, installing from GitHub:
+
+```shell
+pip3 install -U git+https://github.com/djosix/padding_oracle.py.git
+```
 
 ## Performance
 

padding_oracle/__init__.py

@@ -0,0 +1,24 @@
+'''
+Copyright (c) 2020 Yuankui Lee
+
+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.
+'''
+
+from .padding_oracle import *
+from .encoding import *

padding_oracle/encoding.py

@@ -0,0 +1,67 @@
+'''
+Copyright (c) 2020 Yuankui Lee
+
+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.
+'''
+
+import base64
+import urllib.parse
+from typing import Union
+
+__all__ = [
+    'base64_encode', 'base64_decode',
+    'urlencode', 'urldecode'
+]
+
+
+def _to_bytes(data: Union[str, bytes]):
+    if isinstance(data, str):
+        data = data.encode()
+    assert isinstance(data, bytes)
+    return data
+
+
+def _to_str(data):
+    if isinstance(data, bytes):
+        data = data.decode()
+    elif isinstance(data, str):
+        pass
+    else:
+        data = str(data)
+    return data
+
+
+def base64_decode(data: Union[str, bytes]) -> bytes:
+    data = _to_bytes(data)
+    return base64.b64decode(data)
+
+
+def base64_encode(data: Union[str, bytes]) -> str:
+    data = _to_bytes(data)
+    return base64.b64encode(data).decode()
+
+
+def urlencode(data: Union[str, bytes]) -> str:
+    data = _to_bytes(data)
+    return urllib.parse.quote(data)
+
+
+def urldecode(data: str) -> bytes:
+    data = _to_str(data)
+    return urllib.parse.unquote_plus(data)

padding_oracle.py renamed to padding_oracle/padding_oracle.py

@@ -1,73 +1,57 @@
-import logging
-import base64
-import urllib.parse
+'''
+Copyright (c) 2020 Yuankui Lee
+
+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.
+'''
 
+import logging
 from typing import Union, Callable
 from concurrent.futures import ThreadPoolExecutor
 
+from .encoding import *
 
 __all__ = [
-    'base64_encode', 'base64_decode',
-    'urlencode', 'urldecode',
     'padding_oracle',
     'remove_padding'
 ]
 
 
-def _to_bytes(data: Union[str, bytes]):
-    if isinstance(data, str):
-        data = data.encode()
-    assert isinstance(data, bytes)
-    return data
-
-def _to_str(data):
-    if isinstance(data, bytes):
-        data = data.decode()
-    elif isinstance(data, str):
-        pass
-    else:
-        data = str(data)
-    return data
-
-
-def base64_decode(data: Union[str, bytes]) -> bytes:
-    data = _to_bytes(data)
-    return base64.b64decode(data)
-
-def base64_encode(data: Union[str, bytes]) -> str:
-    data = _to_bytes(data)
-    return base64.b64encode(data).decode()
-
-def urlencode(data: Union[str, bytes]) -> str:
-    data = _to_bytes(data)
-    return urllib.parse.quote(data)
-
-def urldecode(data: str) -> bytes:
-    data = _to_str(data)
-    return urllib.parse.unquote_plus(data)
-
 def remove_padding(data: Union[str, bytes]):
     data = _to_bytes(data)
     return data[:-data[-1]]
 
 
-
-
 def _dummy_oracle(cipher: bytes) -> bool:
     raise NotImplementedError('You must implement the oracle function')
 
 
 def padding_oracle(cipher: bytes,
                    block_size: int,
-                   oracle: Callable[[bytes], bool]=_dummy_oracle,
-                   num_threads: int=1,
-                   log_level: int=logging.INFO,
-                   null: bytes=b' ') -> bytes:
+                   oracle: Callable[[bytes], bool] = _dummy_oracle,
+                   num_threads: int = 1,
+                   log_level: int = logging.INFO,
+                   null: bytes = b' ') -> bytes:
     # Check the oracle function
     assert callable(oracle), 'the oracle function should be callable'
     assert oracle.__code__.co_argcount == 1, 'expect oracle function with only 1 argument'
     assert len(cipher) % block_size == 0, 'cipher length should be multiple of block size'
-    
+
     logger = logging.getLogger('padding_oracle')
     logger.setLevel(log_level)
     formatter = logging.Formatter('[%(asctime)s][%(levelname)s] %(message)s')
@@ -84,52 +68,54 @@ def _update_plaintext(i: int, c: bytes):
         logger.info('plaintext: {}'.format(b''.join(plaintext)))
 
     oracle_executor = ThreadPoolExecutor(max_workers=num_threads)
-    
+
     def _block_decrypt_task(i, prev: bytes, block: bytes):
         logger.debug('task={} prev={} block={}'.format(i, prev, block))
         guess_list = list(prev)
-        
+
         for j in range(1, block_size + 1):
             oracle_hits = []
             oracle_futures = {}
-            
+
             for k in range(256):
                 if i == len(blocks) - 1 and j == 1 and k == prev[-j]:
                     # skip the last padding byte if it is identical to the original cipher
                     continue
-                
+
                 test_list = guess_list.copy()
                 test_list[-j] = k
-                oracle_futures[k] = oracle_executor.submit(oracle, bytes(test_list) + block)
-            
+                oracle_futures[k] = oracle_executor.submit(
+                    oracle, bytes(test_list) + block)
+
             for k, future in oracle_futures.items():
                 if future.result():
                     oracle_hits.append(k)
-            
-            logger.debug('oracles at block[{}][{}] -> {}'.format(i, block_size - j, oracle_hits))
-                
+
+            logger.debug(
+                'oracles at block[{}][{}] -> {}'.format(i, block_size - j, oracle_hits))
+
             if len(oracle_hits) != 1:
                 logfmt = 'at block[{}][{}]: expect only one positive result, got {}. (skipped)'
                 logger.error(logfmt.format(i, block_size-j, len(oracle_hits)))
                 return
-            
+
             guess_list[-j] = oracle_hits[0]
-            
+
             p = guess_list[-j] ^ j ^ prev[-j]
             _update_plaintext(i * block_size - j, bytes([p]))
-            
+
             for n in range(j):
                 guess_list[-n-1] ^= j
                 guess_list[-n-1] ^= j + 1
-    
+
     blocks = []
-    
+
     for i in range(0, len(cipher), block_size):
         j = i + block_size
         blocks.append(cipher[i:j])
-    
+
     logger.debug('blocks: {}'.format(blocks))
-    
+
     with ThreadPoolExecutor() as executor:
         futures = []
         for i in reversed(range(1, len(blocks))):
@@ -138,7 +124,7 @@ def _block_decrypt_task(i, prev: bytes, block: bytes):
             futures.append(executor.submit(_block_decrypt_task, i, prev, block))
         for future in futures:
             future.result()
-    
+
     oracle_executor.shutdown()
-    
+
     return b''.join(plaintext)

setup.py

@@ -0,0 +1,23 @@
+import setuptools
+
+with open('README.md', 'r') as fh:
+    long_description = fh.read()
+
+setuptools.setup(
+    name='padding_oracle',
+    version='0.1.1',
+    author='Yuankui Lee',
+    author_email='[email protected]',
+    description='Threaded padding oracle automation.',
+    long_description=long_description,
+    long_description_content_type='text/markdown',
+    url='https://github.com/djosix/padding_oracle.py',
+    packages=setuptools.find_packages(),
+    classifiers=[
+        'Programming Language :: Python :: 3',
+        'License :: OSI Approved :: MIT License',
+        'Operating System :: OS Independent',
+        'Topic :: Security :: Cryptography'
+    ],
+    python_requires='>=3.5',
+)