From 32c314da8628d8c7db2dfd2e083ff5c5c5f207e1 Mon Sep 17 00:00:00 2001 From: liquidsec Date: Fri, 23 Oct 2026 14:56:28 -0400 Subject: [PATCH] Fix ruff 0.17.2 lint errors and pin ruff in CI CI installed ruff unpinned, and 0.17.2 expanded the default rule set, breaking the lint job with 88 errors. Fix them all and pin ruff so new releases can't break CI again. - Merge nested `with` statements in tests (SIM117) - Use context managers for file I/O; add loadState() helper (SIM115, ASYNC230) - Narrow blind excepts to the specific exceptions; keep the intentional per-block catch in nextBlock() with a noqa (BLE001) - Add BlockFailedError instead of raising bare Exception (TRY002) - Import sorting, f-strings, and other autofixes --- .github/workflows/tests.yaml | 2 +- blockbuster/blockbuster.py | 130 ++++++++++++------------ tests/conftest.py | 58 +++++------ tests/test_block_orchestration.py | 58 ++++++----- tests/test_crypto_logic.py | 143 ++++++++++++++++---------- tests/test_encoding.py | 15 ++- tests/test_end_to_end.py | 162 +++++++++++++++++------------- tests/test_job_state.py | 1 - tests/test_padding.py | 3 +- tests/test_request_building.py | 4 +- tests/test_request_http.py | 98 +++++++++++------- tests/test_utility_functions.py | 9 +- 12 files changed, 381 insertions(+), 302 deletions(-) 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 # ---------------------------------------------------------------------------