diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 80c22fd..8f27dac 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -20,7 +20,7 @@ jobs: with: python-version: "3.12" - name: Install Ruff - run: pip install ruff + run: pip install ruff==0.17.2 - name: Ruff format check run: ruff format --check . - name: Ruff lint check diff --git a/blockbuster/blockbuster.py b/blockbuster/blockbuster.py index 97cf7e7..58153e1 100644 --- a/blockbuster/blockbuster.py +++ b/blockbuster/blockbuster.py @@ -2,22 +2,23 @@ # A python padding oracle vulnerability exploitation tool # By Paul Mueller (@paulmmueller) -import socket -import httpx -import sys -import urllib.parse -import binascii import argparse -import os.path -from os import path +import asyncio +import base64 +import binascii import configparser import json -import validators +import os +import os.path import pickle +import socket +import sys import time -import base64 -import os -import asyncio +import urllib.parse +from os import path + +import httpx +import validators # httpx does not emit SSL warnings by default when verify=False @@ -34,16 +35,16 @@ def encode_multipart(fields): body = ( "".join( - "--%s\r\n" - 'Content-Disposition: form-data; name="%s"\r\n' + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="{field}"\r\n' "\r\n" - "%s\r\n" % (boundary, field, value) + f"{value}\r\n" for field, value in fields.items() ) - + "--%s--\r\n" % boundary + + f"--{boundary}--\r\n" ) - content_type = "multipart/form-data; boundary=%s" % boundary + content_type = f"multipart/form-data; boundary={boundary}" return body, content_type @@ -55,12 +56,15 @@ def split_by_n(seq, n): seq = seq[n:] +class BlockFailedError(Exception): + pass + + # append message to log files def writeToLog(message): ts = str(time.time()) - f = open("blockbuster.log", "a") - f.write(f"{ts}:{message}\n") - f.close() + with open("blockbuster.log", "a") as f: + f.write(f"{ts}:{message}\n") # / and + b64 characters are problematic if they are not URL encoded @@ -93,10 +97,14 @@ def saveState(job): byte_suffix = ( f"-BYTE_{bytes_done}" if bytes_done > 0 and currentBlockStr != "FINAL" else "" ) - outputFileName = f"blockbuster-state-{job.name}-BLOCK_{currentBlockStr}{byte_suffix}-{str(int(ts))}.pkl" - pickleOut = open(outputFileName, "wb") - pickle.dump(job, pickleOut) - pickleOut.close() + outputFileName = f"blockbuster-state-{job.name}-BLOCK_{currentBlockStr}{byte_suffix}-{int(ts)}.pkl" + with open(outputFileName, "wb") as pickleOut: + pickle.dump(job, pickleOut) + + +def loadState(fileName): + with open(fileName, "rb") as pickleFile: + return pickle.load(pickleFile) # add padding to the end of the string @@ -113,7 +121,7 @@ def paddify(string, blocksize): # if we fall right on a block boundary, we need a full block of padding if padding_length == 0: padding_length = blocksize - for i in range(0, padding_length): + for i in range(padding_length): temp_group = temp_group + chr(padding_length) groups_storage.append(temp_group) paddedstring = "".join(groups_storage) @@ -157,10 +165,10 @@ def __init__( print("[*]Initializing job....") self.name = name print(f"\nJob name: {self.name}") - print(f"[+]Blocksize: {str(blocksize)}") + print(f"[+]Blocksize: {blocksize}") self.blocksize = blocksize self.mode = mode - print(f"\n[+]Mode: {str(mode)}") + print(f"\n[+]Mode: {mode}") self.debug = debug if self.debug: print("[+]Debug Mode ON\n") @@ -827,7 +835,7 @@ def printProgress(self): print("##################################") try: print("".join(self.solvedBlocks.values())) - except Exception: + except TypeError: print(b"".join(self.solvedBlocks.values())) print("##################################") @@ -856,7 +864,7 @@ def decryptBlockFail(self, padding_array, tempTokenBytes): "\n[!]ERROR! No characters produced valid padding! This must mean there was previously an irrecoverable error!" ) print("*************************************************\n") - raise Exception( + raise BlockFailedError( "Block failed to decrypt/encrypt, likely a random network error." ) # sys.exit(2) @@ -931,7 +939,7 @@ async def encryptBlock(self): (self.blocks[self.currentBlock][currentbyte]) ^ currenti ) print( - f"[+]crypto value of this char in the next (previous) block is: {str(currentcrypto)}\n" + f"[+]crypto value of this char in the next (previous) block is: {currentcrypto}\n" ) solved_crypto[currentbyte] = currentcrypto padding_num += 1 @@ -964,7 +972,7 @@ async def encryptBlock(self): currentcrypto = (self.blocks[self.currentBlock][currentbyte]) ^ currenti print( - f"[+]crypto value of this char in the next (previous) block is: {str(currentcrypto)}\n" + f"[+]crypto value of this char in the next (previous) block is: {currentcrypto}\n" ) solved_crypto[currentbyte] = currentcrypto @@ -1034,7 +1042,7 @@ async def decryptBlock(self): print(f"[+] Verified! Skipping byte {currentbyte}") solved_intermediates[currentbyte] = currenti currentreal = (previousBlock[currentbyte]) ^ currenti - print(f"[+]real value of last char is: {str(currentreal)}\n") + print(f"[+]real value of last char is: {currentreal}\n") solved_reals[currentbyte] = currentreal padding_num += 1 currentbyte -= 1 @@ -1079,7 +1087,7 @@ async def decryptBlock(self): try: blockresult.decode() - except Exception: + except UnicodeDecodeError: blockresult.decode("latin1") print("[*] Block contains non-UTF-8 bytes, using latin1 decoding") writeToLog(f"[!]BLOCK SOLVED: {blockresult}") @@ -1092,7 +1100,7 @@ async def nextBlock(self): try: result = await self.decryptBlock() - except Exception as e: + except Exception as e: # noqa: BLE001 - any failure (network, padding) aborts just this block writeToLog( f"[!] decryption of block {self.currentBlock} failed. Error message: {e}" ) @@ -1103,7 +1111,7 @@ async def nextBlock(self): if self.mode == "encrypt": try: result = await self.encryptBlock() - except Exception as e: + except Exception as e: # noqa: BLE001 - any failure (network, padding) aborts just this block writeToLog( f"[!] encryption of block {self.currentBlock} failed. Error message: {e}" ) @@ -1176,8 +1184,8 @@ def decryptInit(self): # Get the block count and save it to the instance print(actualBlocks) - print(int(len(actualBlocks))) - self.blockCount = int((len(actualBlocks) / self.blocksize)) + print(len(actualBlocks)) + self.blockCount = int(len(actualBlocks) / self.blocksize) # if the mode is 'firstblock' we need to remove the first block and assign it as the IV if self.ivMode == "firstblock": @@ -1202,7 +1210,7 @@ def decryptInit(self): print("*************************************************\n") # iterate through the block array and separate the blocks - for x in range(0, self.blockCount): + for x in range(self.blockCount): # take the next block off and add it to self.blocks self.blocks.append(actualBlocks[0 : self.blocksize]) @@ -1277,10 +1285,10 @@ def encryptInit(self): actualBlocks = self.bytemap[:] # Get the block count and save it to the instance - self.blockCount = int((len(self.bytemap) / self.blocksize)) + self.blockCount = int(len(self.bytemap) / self.blocksize) # iterate through the block array and separate the blocks - for x in range(0, self.blockCount): + for x in range(self.blockCount): # take the next block off and add it to self.blocks self.blocks.append(actualBlocks[0 : self.blocksize]) @@ -1354,9 +1362,7 @@ async def async_main(): f"\n[!]RESTORE MODE INTIATED. Attempting to restart job from file {args.restore}" ) - pickleFile = open(args.restore, "rb") - job = pickle.load(pickleFile) - pickleFile.close() + job = loadState(args.restore) # Recreate the httpx client since it can't be pickled job.initialize_client() if not hasattr(job, "preseeded_intermediates"): @@ -1433,31 +1439,26 @@ async def async_main(): ) # Validate POST format - if httpMethod == "POST": - if postFormat == "form-urlencoded": - pass - - elif postFormat == "multipart": - pass - - elif postFormat == "json": - pass - else: - handleError( - "[x]CONFIG ERROR: When httpMethod is POST postFormat must be 'form-urlencoded', 'multipart', or 'json'" - ) + if httpMethod == "POST" and postFormat not in ( + "form-urlencoded", + "multipart", + "json", + ): + handleError( + "[x]CONFIG ERROR: When httpMethod is POST postFormat must be 'form-urlencoded', 'multipart', or 'json'" + ) # validate proxy IP if httpProxyIp: try: socket.inet_aton(httpProxyIp) - except socket.error: + except OSError: handleError("[x]CONFIG ERROR: proxy ip is not a valid IP address.") # validate proxy port if httpProxyPort: try: httpProxyPort = int(httpProxyPort) - except Exception: + except ValueError: handleError("[x]CONFIG ERROR: proxy port is not valid INT") if not (httpProxyPort <= 65535): @@ -1466,7 +1467,7 @@ async def async_main(): # validate block size try: blocksize = int(blocksize) - except Exception: + except ValueError: handleError("[x]CONFIG ERROR: blocksize must be INT.") if not validators.url(URL): @@ -1499,11 +1500,10 @@ async def async_main(): # anchorBlock mode requires anchorCiphertext anchorCiphertext = config["default"].get("anchorCiphertext", "") - if ivMode == "anchorBlock": - if not anchorCiphertext: - handleError( - "[x]CONFIG ERROR: anchorCiphertext is required when ivMode is anchorBlock" - ) + if ivMode == "anchorBlock" and not anchorCiphertext: + handleError( + "[x]CONFIG ERROR: anchorCiphertext is required when ivMode is anchorBlock" + ) # Initialize Job object job = Job( @@ -1596,7 +1596,7 @@ async def async_main(): # if we just completed an encrypt operation, we need to reverse the order, join the pieces, and base64 if job.mode == "encrypt": - for xxx in range(0, len(job.solvedBlocks.values())): + for xxx in range(len(job.solvedBlocks.values())): # combine all of the blocks into one decimal list joinedCrypto = b"".join(reversed(list(job.solvedBlocks.values()))) @@ -1633,7 +1633,7 @@ async def async_main(): stripped = combined try: plaintext = stripped.decode(job.plaintextEncoding) - except Exception: + except (UnicodeDecodeError, LookupError): plaintext = stripped.decode("latin1") print(f"\n{'=' * 50}") print(f"[+] Decrypted plaintext: {plaintext}") diff --git a/tests/conftest.py b/tests/conftest.py index e84df17..0de937e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -63,35 +63,35 @@ def check_padding(self, ciphertext_with_iv: bytes) -> bool: # Job factory # --------------------------------------------------------------------------- -JOB_DEFAULTS = dict( - blocksize=16, - mode="decrypt", - debug=False, - sourceString="", - name="test", - ivMode="firstblock", - URL="http://testserver/oracle", - httpMethod="GET", - additionalParameters={}, - httpProxyOn=False, - httpProxyIp="127.0.0.1", - httpProxyPort=8080, - headers={"User-Agent": "pytest"}, - iv=[0] * 16, - oracleMode="negative", - oracleText="Invalid padding", - vulnerableParameter="token", - inputMode="parameter", - cookies={}, - encodingMode="base64", - postFormat="form-urlencoded", - followRedirects=False, - concurrency=5, - redirectDelay=0, - confirmations=0, - plaintextEncoding="utf-8", - anchorCiphertext="", -) +JOB_DEFAULTS = { + "blocksize": 16, + "mode": "decrypt", + "debug": False, + "sourceString": "", + "name": "test", + "ivMode": "firstblock", + "URL": "http://testserver/oracle", + "httpMethod": "GET", + "additionalParameters": {}, + "httpProxyOn": False, + "httpProxyIp": "127.0.0.1", + "httpProxyPort": 8080, + "headers": {"User-Agent": "pytest"}, + "iv": [0] * 16, + "oracleMode": "negative", + "oracleText": "Invalid padding", + "vulnerableParameter": "token", + "inputMode": "parameter", + "cookies": {}, + "encodingMode": "base64", + "postFormat": "form-urlencoded", + "followRedirects": False, + "concurrency": 5, + "redirectDelay": 0, + "confirmations": 0, + "plaintextEncoding": "utf-8", + "anchorCiphertext": "", +} def make_job(**overrides) -> Job: diff --git a/tests/test_block_orchestration.py b/tests/test_block_orchestration.py index fd9a3ec..f6d9355 100644 --- a/tests/test_block_orchestration.py +++ b/tests/test_block_orchestration.py @@ -1,7 +1,7 @@ """Tests for nextBlock, block retry logic, and encrypt sanity check.""" from types import SimpleNamespace -from unittest.mock import patch, AsyncMock +from unittest.mock import AsyncMock, patch import pytest @@ -68,11 +68,13 @@ async def test_successful_encrypt_with_passing_sanity_check(self): def mock_request(token): return SimpleNamespace(text="OK", status_code=200) - with patch.object( - j, "encryptBlock", new_callable=AsyncMock, return_value=block_result + with ( + patch.object( + j, "encryptBlock", new_callable=AsyncMock, return_value=block_result + ), + patch.object(j, "makeRequest", side_effect=mock_request), ): - with patch.object(j, "makeRequest", side_effect=mock_request): - result = await j.nextBlock() + result = await j.nextBlock() assert result == 0 assert j.solvedBlocks[0] == block_result @@ -94,11 +96,13 @@ def mock_request(token): # Sanity check fails — the oracle text IS found (negative mode: pass = text absent) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object( - j, "encryptBlock", new_callable=AsyncMock, return_value=block_result + with ( + patch.object( + j, "encryptBlock", new_callable=AsyncMock, return_value=block_result + ), + patch.object(j, "makeRequest", side_effect=mock_request), ): - with patch.object(j, "makeRequest", side_effect=mock_request): - result = await j.nextBlock() + result = await j.nextBlock() assert result == 1 assert 0 not in j.solvedBlocks # Block was backed out @@ -149,18 +153,20 @@ async def mock_next_block(): max_block_retries = 3 block_failures = 0 - with patch.object(j, "nextBlock", side_effect=mock_next_block): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - if result == 0: - block_failures = 0 - j.currentBlock += 1 - j._clear_byte_progress() - else: - block_failures += 1 - if block_failures >= max_block_retries: - break + with ( + patch.object(j, "nextBlock", side_effect=mock_next_block), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + if result == 0: + block_failures = 0 + j.currentBlock += 1 + j._clear_byte_progress() + else: + block_failures += 1 + if block_failures >= max_block_retries: + break assert j.currentBlock == 1 assert call_count == 2 @@ -220,11 +226,13 @@ def mock_request(token): captured_tokens.append(token) return SimpleNamespace(text="OK", status_code=200) - with patch.object( - j, "encryptBlock", new_callable=AsyncMock, return_value=block_result + with ( + patch.object( + j, "encryptBlock", new_callable=AsyncMock, return_value=block_result + ), + patch.object(j, "makeRequest", side_effect=mock_request), ): - with patch.object(j, "makeRequest", side_effect=mock_request): - result = await j.nextBlock() + result = await j.nextBlock() assert result == 0 # The sanity check token should contain the IV appended diff --git a/tests/test_crypto_logic.py b/tests/test_crypto_logic.py index 356af65..53ebb31 100644 --- a/tests/test_crypto_logic.py +++ b/tests/test_crypto_logic.py @@ -10,13 +10,12 @@ import io import sys from types import SimpleNamespace -from unittest.mock import patch, AsyncMock +from unittest.mock import AsyncMock, patch import pytest from Crypto.Cipher import AES -from tests.conftest import make_job, PaddingOracle, AES_KEY, AES_IV, ORACLE - +from tests.conftest import AES_IV, AES_KEY, ORACLE, PaddingOracle, make_job # --------------------------------------------------------------------------- # Helpers: compute expected intermediate values for known key @@ -422,9 +421,11 @@ async def test_dispatches_to_sync_for_low_concurrency(self): j = make_job(concurrency=1) j.initialize_client() - with patch.object(j, "solveByteSync", return_value=(42, None)) as mock_sync: - with patch.object(j, "solveByteAsync") as mock_async: - result = await j.solveByte(15, 1, {}, [0] * 16, False) + with ( + patch.object(j, "solveByteSync", return_value=(42, None)) as mock_sync, + patch.object(j, "solveByteAsync") as mock_async, + ): + result = await j.solveByte(15, 1, {}, [0] * 16, False) mock_sync.assert_called_once() mock_async.assert_not_called() @@ -435,11 +436,13 @@ async def test_dispatches_to_async_for_high_concurrency(self): j = make_job(concurrency=10) j.initialize_client() - with patch.object(j, "solveByteSync") as mock_sync: - with patch.object( + with ( + patch.object(j, "solveByteSync") as mock_sync, + patch.object( j, "solveByteAsync", new_callable=AsyncMock, return_value=(42, None) - ) as mock_async: - await j.solveByte(15, 1, {}, [0] * 16, False) + ) as mock_async, + ): + await j.solveByte(15, 1, {}, [0] * 16, False) mock_async.assert_called_once() mock_sync.assert_not_called() @@ -526,9 +529,11 @@ def mock_oracle_request(token, progress=None): sys.stdout = old_stdout j.initialize_client() - with patch.object(j, "makeRequest", side_effect=mock_oracle_request): - with patch("blockbuster.blockbuster.saveState"): - result = await j.decryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle_request), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.decryptBlock() # The decrypted block should be the first 16 bytes of the padded plaintext # Our plaintext is 16 bytes so block 0 = plaintext, block 1 = padding @@ -580,9 +585,11 @@ def mock_oracle_request(token, progress=None): return SimpleNamespace(text="OK", status_code=200) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_oracle_request): - with patch("blockbuster.blockbuster.saveState"): - result = await j.encryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle_request), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.encryptBlock() # Result should be blocksize bytes of ciphertext assert len(result) == 16 @@ -649,9 +656,11 @@ def mock_oracle(token, progress=None): return SimpleNamespace(text="OK", status_code=200) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.decryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.decryptBlock() assert result == plaintext @@ -704,16 +713,20 @@ def mock_oracle(token, progress=None): j2 = copy.deepcopy(j) j2.initialize_client() - with patch.object(j2, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - await j2.encryptBlock() + with ( + patch.object(j2, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + await j2.encryptBlock() # Now test resume: pre-seed byte 15 as solved # We need to figure out the intermediate for byte 15 # Run the first byte only and capture the intermediate - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.encryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.encryptBlock() assert len(result) == 16 @@ -766,9 +779,11 @@ def mock_oracle(token, progress=None): return SimpleNamespace(text="OK", status_code=200) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.decryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.decryptBlock() assert result == plaintext @@ -811,9 +826,11 @@ def mock_oracle(token, progress=None): return SimpleNamespace(text="OK", status_code=200) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.decryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.decryptBlock() # Should still get correct result via fallback assert result == plaintext @@ -860,9 +877,11 @@ def mock_oracle(token, progress=None): return SimpleNamespace(text="Invalid padding", status_code=200) # No preseeded values - just verify encrypt works - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.encryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.encryptBlock() assert len(result) == 16 @@ -933,9 +952,11 @@ def mock_oracle(token, progress=None): return SimpleNamespace(text="OK", status_code=200) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.decryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.decryptBlock() # The raw bytes should be recovered assert result == raw_plaintext @@ -1232,9 +1253,11 @@ def mock_oracle(token, progress=None): return SimpleNamespace(text="OK", status_code=200) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.decryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.decryptBlock() # First block with unknown IV will be XOR'd with zeros instead of real IV # So we get intermediate values XOR zeros = intermediate values (garbled) @@ -1283,9 +1306,11 @@ def mock_oracle(token, progress=None): return SimpleNamespace(text="OK", status_code=200) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.encryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.encryptBlock() assert len(result) == 16 @@ -1336,9 +1361,11 @@ def mock_oracle(token, progress=None): j2 = copy.deepcopy(j) j2.initialize_client() - with patch.object(j2, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - await j2.encryptBlock() + with ( + patch.object(j2, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + await j2.encryptBlock() # Now re-solve with preseeded byte 15 # We need to discover the intermediate value @@ -1356,9 +1383,11 @@ def mock_oracle(token, progress=None): # Test with wrong value - should still produce valid result via fallback j.preseeded_intermediates = {15: 0x00} # Almost certainly wrong - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result = await j.encryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result = await j.encryptBlock() assert len(result) == 16 @@ -1404,17 +1433,21 @@ def mock_oracle(token, progress=None): return SimpleNamespace(text="Invalid padding", status_code=200) # Solve block 0 - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result0 = await j.encryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result0 = await j.encryptBlock() j.solvedBlocks[0] = result0 j.currentBlock = 1 # Solve block 1 — uses solvedBlocks[0] as previousBlock - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - result1 = await j.encryptBlock() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + result1 = await j.encryptBlock() assert len(result1) == 16 diff --git a/tests/test_encoding.py b/tests/test_encoding.py index b020e3e..7855c25 100644 --- a/tests/test_encoding.py +++ b/tests/test_encoding.py @@ -5,7 +5,6 @@ import sys import urllib.parse - from tests.conftest import make_job @@ -119,13 +118,13 @@ def _init_job( blocksize=16, iv=None, ): - kw = dict( - sourceString=source_string, - mode="decrypt", - encodingMode=encoding_mode, - ivMode=iv_mode, - blocksize=blocksize, - ) + kw = { + "sourceString": source_string, + "mode": "decrypt", + "encodingMode": encoding_mode, + "ivMode": iv_mode, + "blocksize": blocksize, + } if iv is not None: kw["iv"] = iv j = make_job(**kw) diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py index 637d069..7570386 100644 --- a/tests/test_end_to_end.py +++ b/tests/test_end_to_end.py @@ -13,7 +13,7 @@ import pytest -from tests.conftest import make_job, PaddingOracle, AES_KEY, AES_IV, ORACLE +from tests.conftest import AES_IV, AES_KEY, ORACLE, PaddingOracle, make_job def make_mock_oracle( @@ -81,13 +81,15 @@ async def test_decrypt_short_plaintext_base64_firstblock(self): mock_oracle = make_mock_oracle(ORACLE) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0, f"Block {j.currentBlock} failed" - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0, f"Block {j.currentBlock} failed" + j.currentBlock += 1 + j._clear_byte_progress() # Reconstruct plaintext with PKCS#7 stripping combined = b"".join(j.solvedBlocks[i] for i in sorted(j.solvedBlocks.keys())) @@ -128,13 +130,15 @@ async def test_decrypt_exact_block_boundary(self): mock_oracle = make_mock_oracle(ORACLE) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0 - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0 + j.currentBlock += 1 + j._clear_byte_progress() combined = b"".join(j.solvedBlocks[i] for i in sorted(j.solvedBlocks.keys())) pad_len = combined[-1] @@ -176,13 +180,15 @@ async def test_decrypt_multi_block(self): mock_oracle = make_mock_oracle(ORACLE) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0 - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0 + j.currentBlock += 1 + j._clear_byte_progress() combined = b"".join(j.solvedBlocks[i] for i in sorted(j.solvedBlocks.keys())) pad_len = combined[-1] @@ -223,13 +229,15 @@ async def test_decrypt_hex_encoding(self): mock_oracle = make_mock_oracle(ORACLE, encoding_mode="hex") - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0 - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0 + j.currentBlock += 1 + j._clear_byte_progress() combined = b"".join(j.solvedBlocks[i] for i in sorted(j.solvedBlocks.keys())) pad_len = combined[-1] @@ -276,13 +284,15 @@ async def test_decrypt_knownIV_mode(self): # The oracle just checks the raw bytes we send mock_oracle = make_mock_oracle(ORACLE) - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0 - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0 + j.currentBlock += 1 + j._clear_byte_progress() combined = b"".join(j.solvedBlocks[i] for i in sorted(j.solvedBlocks.keys())) pad_len = combined[-1] @@ -332,13 +342,15 @@ def mock_request(token, progress=None): return SimpleNamespace(text="Decryption successful", status_code=200) return SimpleNamespace(text="Error occurred", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_request): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0 - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequest", side_effect=mock_request), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0 + j.currentBlock += 1 + j._clear_byte_progress() combined = b"".join(j.solvedBlocks[i] for i in sorted(j.solvedBlocks.keys())) pad_len = combined[-1] @@ -392,13 +404,15 @@ def mock_request_sync(token, progress=None): async def mock_request_async(token, progress=None): return mock_request_sync(token, progress) - with patch.object(j, "makeRequestAsync", side_effect=mock_request_async): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0 - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequestAsync", side_effect=mock_request_async), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0 + j.currentBlock += 1 + j._clear_byte_progress() combined = b"".join(j.solvedBlocks[i] for i in sorted(j.solvedBlocks.keys())) pad_len = combined[-1] @@ -458,13 +472,15 @@ def mock_request(token, progress=None): return SimpleNamespace(text="OK", status_code=200) return SimpleNamespace(text="Invalid padding", status_code=200) - with patch.object(j, "makeRequest", side_effect=mock_request): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0, f"Block {j.currentBlock} failed" - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequest", side_effect=mock_request), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0, f"Block {j.currentBlock} failed" + j.currentBlock += 1 + j._clear_byte_progress() # Verify the final assembled ciphertext is oracle-valid joined = b"".join(reversed(list(j.solvedBlocks.values()))) @@ -516,24 +532,28 @@ async def test_resume_mid_block(self): j2 = copy.deepcopy(j) j2.initialize_client() - with patch.object(j2, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - while j2.currentBlock < j2.blockCount: - result = await j2.nextBlock() - assert result == 0 - j2.currentBlock += 1 - j2._clear_byte_progress() + with ( + patch.object(j2, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + while j2.currentBlock < j2.blockCount: + result = await j2.nextBlock() + assert result == 0 + j2.currentBlock += 1 + j2._clear_byte_progress() expected = b"".join(j2.solvedBlocks[i] for i in sorted(j2.solvedBlocks.keys())) # Now run with j but verify it produces the same result - with patch.object(j, "makeRequest", side_effect=mock_oracle): - with patch("blockbuster.blockbuster.saveState"): - while j.currentBlock < j.blockCount: - result = await j.nextBlock() - assert result == 0 - j.currentBlock += 1 - j._clear_byte_progress() + with ( + patch.object(j, "makeRequest", side_effect=mock_oracle), + patch("blockbuster.blockbuster.saveState"), + ): + while j.currentBlock < j.blockCount: + result = await j.nextBlock() + assert result == 0 + j.currentBlock += 1 + j._clear_byte_progress() actual = b"".join(j.solvedBlocks[i] for i in sorted(j.solvedBlocks.keys())) assert actual == expected diff --git a/tests/test_job_state.py b/tests/test_job_state.py index 0c234dc..f375665 100644 --- a/tests/test_job_state.py +++ b/tests/test_job_state.py @@ -3,7 +3,6 @@ import pickle from unittest.mock import patch - from tests.conftest import make_job diff --git a/tests/test_padding.py b/tests/test_padding.py index 0c123a5..d03c6fc 100644 --- a/tests/test_padding.py +++ b/tests/test_padding.py @@ -3,7 +3,6 @@ import io import sys - from blockbuster.blockbuster import paddify from tests.conftest import make_job @@ -104,7 +103,7 @@ def test_blocks_are_reversed(self): assert len(j.blocks) == 3 # blocks[0] should be the LAST plaintext block (padding block) # blocks[-1] should be the FIRST plaintext block - padded = "A".encode() * 32 + bytes([16] * 16) + padded = b"A" * 32 + bytes([16] * 16) first_block = list(padded[:16]) last_block = list(padded[32:48]) assert list(j.blocks[0]) == last_block diff --git a/tests/test_request_building.py b/tests/test_request_building.py index fa56548..679b96f 100644 --- a/tests/test_request_building.py +++ b/tests/test_request_building.py @@ -74,7 +74,7 @@ def test_form_urlencoded_parameter(self): postFormat="form-urlencoded", vulnerableParameter="tok", ) - method, url, kwargs = j._buildRequest("AAAA") + method, _url, kwargs = j._buildRequest("AAAA") assert method == "POST" assert kwargs["data"]["tok"] == "AAAA" assert "application/x-www-form-urlencoded" in kwargs["headers"]["Content-Type"] @@ -108,7 +108,7 @@ def test_multipart_parameter(self): def test_querystring_in_post(self): j = self._job(inputMode="querystring", postFormat="form-urlencoded") - _, url, kwargs = j._buildRequest("CT_IN_QS") + _, url, _kwargs = j._buildRequest("CT_IN_QS") # Token goes in URL, not body assert "?CT_IN_QS" in url diff --git a/tests/test_request_http.py b/tests/test_request_http.py index b0c596d..75d547b 100644 --- a/tests/test_request_http.py +++ b/tests/test_request_http.py @@ -1,6 +1,6 @@ """Tests for HTTP retry logic, redirect following, and async requests.""" -from unittest.mock import patch, AsyncMock +from unittest.mock import AsyncMock, patch import httpx import pytest @@ -24,9 +24,11 @@ def fake_get(url, **kwargs): raise httpx.ConnectError("Connection refused") return httpx.Response(200, text="OK") - with patch.object(j.client, "get", side_effect=fake_get): - with patch("time.sleep"): # skip actual waits - result = j.makeRequest("test_token") + with ( + patch.object(j.client, "get", side_effect=fake_get), + patch("time.sleep"), # skip actual waits + ): + result = j.makeRequest("test_token") assert result.status_code == 200 assert call_count == 3 @@ -38,10 +40,12 @@ def test_raises_after_max_retries(self): def always_fail(url, **kwargs): raise httpx.ConnectError("Connection refused") - with patch.object(j.client, "get", side_effect=always_fail): - with patch("time.sleep"): - with pytest.raises(httpx.ConnectError): - j.makeRequest("test_token") + with ( + patch.object(j.client, "get", side_effect=always_fail), + patch("time.sleep"), + pytest.raises(httpx.ConnectError), + ): + j.makeRequest("test_token") def test_retry_backoff_times(self): j = make_job() @@ -57,9 +61,11 @@ def fake_get(url, **kwargs): raise httpx.TimeoutException("timeout") return httpx.Response(200, text="OK") - with patch.object(j.client, "get", side_effect=fake_get): - with patch("time.sleep", side_effect=lambda s: sleep_calls.append(s)): - j.makeRequest("test_token") + with ( + patch.object(j.client, "get", side_effect=fake_get), + patch("time.sleep", side_effect=lambda s: sleep_calls.append(s)), + ): + j.makeRequest("test_token") # Backoff: 2**0=1, 2**1=2 assert sleep_calls == [1, 2] @@ -82,9 +88,11 @@ async def fake_get(url, **kwargs): raise httpx.ConnectError("Connection refused") return httpx.Response(200, text="OK") - with patch.object(j.async_client, "get", side_effect=fake_get): - with patch("asyncio.sleep", new_callable=AsyncMock): - result = await j.makeRequestAsync("test_token") + with ( + patch.object(j.async_client, "get", side_effect=fake_get), + patch("asyncio.sleep", new_callable=AsyncMock), + ): + result = await j.makeRequestAsync("test_token") assert result.status_code == 200 assert call_count == 3 @@ -97,10 +105,12 @@ async def test_raises_after_max_retries(self): async def always_fail(url, **kwargs): raise httpx.ConnectError("Connection refused") - with patch.object(j.async_client, "get", side_effect=always_fail): - with patch("asyncio.sleep", new_callable=AsyncMock): - with pytest.raises(httpx.ConnectError): - await j.makeRequestAsync("test_token") + with ( + patch.object(j.async_client, "get", side_effect=always_fail), + patch("asyncio.sleep", new_callable=AsyncMock), + pytest.raises(httpx.ConnectError), + ): + await j.makeRequestAsync("test_token") class TestRedirectFollowing: @@ -148,9 +158,11 @@ def test_redirect_delay(self): final_resp = httpx.Response(200, text="OK") sleep_calls = [] - with patch.object(j.client, "get", return_value=final_resp): - with patch("time.sleep", side_effect=lambda s: sleep_calls.append(s)): - j._followRedirect(redirect_resp, {}) + with ( + patch.object(j.client, "get", return_value=final_resp), + patch("time.sleep", side_effect=lambda s: sleep_calls.append(s)), + ): + j._followRedirect(redirect_resp, {}) assert sleep_calls == [0.5] @@ -221,11 +233,13 @@ async def test_redirect_delay_uses_async_sleep(self): async def fake_sleep(s): sleep_calls.append(s) - with patch.object( - j.async_client, "get", new_callable=AsyncMock, return_value=final_resp + with ( + patch.object( + j.async_client, "get", new_callable=AsyncMock, return_value=final_resp + ), + patch("asyncio.sleep", side_effect=fake_sleep), ): - with patch("asyncio.sleep", side_effect=fake_sleep): - await j._followRedirectAsync(redirect_resp, {}) + await j._followRedirectAsync(redirect_resp, {}) assert sleep_calls == [1.0] @@ -411,9 +425,11 @@ def test_get_with_follow_redirects(self): request=httpx.Request("GET", "http://testserver/oracle?token=x"), ) - with patch.object(j.client, "get", return_value=resp): - with patch.object(j, "_followRedirect", return_value=resp) as mock_redir: - j.makeRequest("tok") + with ( + patch.object(j.client, "get", return_value=resp), + patch.object(j, "_followRedirect", return_value=resp) as mock_redir, + ): + j.makeRequest("tok") mock_redir.assert_called_once() @@ -432,9 +448,11 @@ def fake_get(url, **kwargs): return httpx.Response(200, text="OK") progress = [0, 0] - with patch.object(j.client, "get", side_effect=fake_get): - with patch("time.sleep"): - j.makeRequest("tok", progress=progress) + with ( + patch.object(j.client, "get", side_effect=fake_get), + patch("time.sleep"), + ): + j.makeRequest("tok", progress=progress) assert progress[1] == 2 # Two retries @@ -472,11 +490,13 @@ async def test_follow_redirects_async(self): async def fake_get(url, **kwargs): return resp - with patch.object(j.async_client, "get", side_effect=fake_get): - with patch.object( + with ( + patch.object(j.async_client, "get", side_effect=fake_get), + patch.object( j, "_followRedirectAsync", new_callable=AsyncMock, return_value=resp - ) as mock_redir: - await j.makeRequestAsync("tok") + ) as mock_redir, + ): + await j.makeRequestAsync("tok") mock_redir.assert_called_once() @@ -495,8 +515,10 @@ async def fake_get(url, **kwargs): return httpx.Response(200, text="OK") progress = [0, 0] - with patch.object(j.async_client, "get", side_effect=fake_get): - with patch("asyncio.sleep", new_callable=AsyncMock): - await j.makeRequestAsync("tok", progress=progress) + with ( + patch.object(j.async_client, "get", side_effect=fake_get), + patch("asyncio.sleep", new_callable=AsyncMock), + ): + await j.makeRequestAsync("tok", progress=progress) assert progress[1] == 1 diff --git a/tests/test_utility_functions.py b/tests/test_utility_functions.py index c93ac3c..ed4876e 100644 --- a/tests/test_utility_functions.py +++ b/tests/test_utility_functions.py @@ -6,16 +6,15 @@ import pytest from blockbuster.blockbuster import ( - makeCookieString, - encode_multipart, - split_by_n, - bytes_to_base64, b64urlEncode, + bytes_to_base64, + encode_multipart, handleError, + makeCookieString, saveState, + split_by_n, ) - # --------------------------------------------------------------------------- # makeCookieString # ---------------------------------------------------------------------------