hybrid unifier

6 minutos de lectura

Se nos proporciona un proyecto de Flask con el servidor que tiene la flag:

$ tree
.
├── Dockerfile
├── build-docker.sh
├── challenge
│   ├── README.pdf
│   └── application
│       ├── app.py
│       ├── crypto
│       │   └── session.py
│       ├── requirements.txt
│       └── views.py
└── flag.txt

4 directories, 8 files

También se nos proporciona un documento PDF con una especie de documentación de API de los endpoints del servidor. Esta es la parte relevante:

Endpoints
For the following endpoints, only the POST method is allowed.
/api/request-session-parameters You can use this endpoint to obtain the Diffie Hellman parameters. You will need them for initializing the secure session.
/api/init-session You can use this endpoint to establish a secure session with the server. You receive the server’s Diffie Hellman public key and you send the server yours.
/api/request-challenge You can use this endpoint to request an encrypted challenge from the server. It is encrypted with the shared session key. This challenge is required for authentication to interact with the /api/dashboard endpoint
/api/dashboard You can use this endpoint to send the action you want to the server. The available actions at the moment are flag and about. Keep in mind that you need to send an encrypted packet as the communication is E2E encrypted.
The /api/request-challenge and /api/dashboard endpoints can be accessed only after a secure session is established. For this, you need to first interact with the /api/init-session endpoint.

Por lo tanto, el objetivo de este reto es programar un cliente que solicite la flag de la API del servidor. Tendremos que comprender cómo funciona el protocolo de cifrado para implementarlo desde cero.

Análisis del código fuente

Hay dos archivos relevantes: views.py y crypto/session.py. El primero se utiliza para administrar los endpoints de la API, mientras que el segundo contiene la implementación del protocolo criptográfico.

El servidor crea un nuevo objeto SecureSession al inicio y expone la información pública en /api/request-session-parameters:

session = SecureSession(384)

# Step 1. Request the Diffie Hellman parameters
@bp.route('/api/request-session-parameters', methods=['POST'])
def get_session_parameters():
    return jsonify({'g': hex(session.g), 'p': hex(session.p)})

Diffie-Hellman

Estos parámetros corresponden a un protocolo de intercambio clave de Diffie-Hellman:

class SecureSession:
    def __init__(self, bits):
        self.bits = bits
        self.g = 2
        self.p = getPrime(self.bits)
        self.compute_server_public_key()
        self.reset_challenge()
        self.initialized = False

    def compute_server_public_key(self):
        self.a = randbelow(self.p)
        self.server_public_key = pow(self.g, self.a, self.p)
    
    def reset_challenge(self):
        self.challenge = os.urandom(24)

En resumen, el servidor elige un número primo público y un generador . La clave privada se elige al azar tal que . Entonces, la clave pública es .

En este punto, un cliente debe tomar los parámetros del protocolo y , calcular una clave privada y su clave pública asociada . El cliente envía el valor al servidor, y en este punto, ambos comparten un secreto :

Ahora, tanto el cliente como el servidor pueden usar el valor compartido para derivar una clave para un algoritmo de cifrado simétrico, como AES.

Esto es seguro porque se basa en el Problema de Logaritmo Discreto, que es computacionalmente difícil de resolver si los parámetros se eligen correctamente. Un atacante tendría que hallar dado .

Protocolo

Como se dijo en la documentación anterior, debemos comenzar con /api/init-session:

# Step 2. Initialize a secure session with the server by sending your Diffie Hellman public key
@bp.route('/api/init-session', methods=['POST'])
def init_session():
    if session.initialized:
        return jsonify({'status_code': 400, 'error': 'A secure session has already been established.'})

    data = request.json
    if 'client_public_key' not in data:
        return jsonify({'status_code': 400, 'error': 'You need to send the client public key.'})
    
    client_public_key = data['client_public_key']
    session.establish_session_key(client_public_key)
    session.initialized = True
    return jsonify({'status_code': 200, 'success': 'A secure session was successfully established. There will be E2E encryption for the rest of the communication.', 'server_public_key': hex(session.server_public_key)})

Aquí, debemos enviar nuestra clave pública (el valor del ejemplo anterior).

El servidor creará una clave de sesión usando establish_session_key:

    def establish_session_key(self, client_public_key):
        key = pow(client_public_key, self.a, self.p)
        self.session_key = sha256(str(key).encode()).digest()

Este método calcula , que es el secreto compartido, y usa SHA256 para derivar una clave de sesión de 32 bytes.

Ahora, el servidor requiere que resolvamos un desafío:

# Step 3. Request an encrypted challenge.
@bp.route('/api/request-challenge', methods=['POST'])
def request_challenge():
    if not session.initialized:
        return jsonify({'status_code': 400, 'error': 'A secure server-client session has to be established first.'})

    return jsonify({'encrypted_challenge': session.get_encrypted_challenge().decode()})

Esta es una forma de validar que no hay errores en el protocolo de comunicación. Esto es get_encrypted_challenge:

    def get_encrypted_challenge(self):
        iv = os.urandom(16)
        cipher = AES.new(self.session_key, AES.MODE_CBC, iv)
        encrypted_challenge = iv + cipher.encrypt(pad(self.challenge, 16))
        return be(encrypted_challenge)

Si el intercambio de claves funcionó correctamente, deberíamos poder obtener la clave de la sesión a partir del secreto compartido y descifrar el desafío.

El servidor verifica el desafío en /api/dashboard antes de realizar la acción especificada en packet_data:

# Step 4. Authenticate by responding to the challenge and send an encrypted packet with 'flag' as action to get the flag. 
@bp.route('/api/dashboard', methods=['POST'])
def access_secret():
    if not session.initialized:
        return jsonify({'status_code': 400, 'error': 'A secure server-client session has to be established first.'})

    data = request.json

    if 'challenge' not in data:
        return jsonify({'status_code': 400, 'error': 'You need to send the hash of the challenge.'})

    if 'packet_data' not in data:
        return jsonify({'status_code': 400, 'error': 'Empty packet.'})
        
    challenge_hash = data['challenge']
    if not session.validate_challenge(challenge_hash):
        return jsonify({'status_code': 401, 'error': 'Invalid challenge! Something wrong? You can visit /request-challenge to get a new challenge!'})

    encrypted_packet = data['packet_data']

    packet = session.decrypt_packet(encrypted_packet)
    if not 'packet_data' in packet:
        return jsonify({'status_code': 400, 'error': packet['error']})
    
    action = packet['packet_data']
    if action == 'flag':
        return jsonify(session.encrypt_packet(open('/flag.txt').read()))
    elif action == 'about':
        return jsonify(session.encrypt_packet('[+] Welcome to my custom API! You are currently Alpha testing my new E2E protocol.\nTo get the flag, all you have to do is to follow the protocol as intended. For any bugs, feel free to contact us :-] !'))
    else:
        return jsonify(session.encrypt_packet('[!] Unknown action.'))

Si todo es correcto, deberíamos poder enviar un comando flag para obtener la flag.

Solución

Solo necesitamos seguir los pasos anteriores:

Tomamos los parámetros del protocolo:

# Step 1
r = requests.post(f'http://{URL}/api/request-session-parameters')

g = int(r.json().get('g'), 16)
p = int(r.json().get('p'), 16)

Calculamos una clave privada y enviamos la clave pública:

# Step 2
b = randbelow(p)
client_public_key = pow(g, b, p)

r = requests.post(f'http://{URL}/api/init-session', json={
    'client_public_key': client_public_key
})

server_public_key = int(r.json().get('server_public_key'), 16)

key = pow(server_public_key, b, p)
session_key = sha256(str(key).encode()).digest()

Derivamos la clave de la sesión a partir del secreto compartido y solicitamos un desafío:

# Step 3
r = requests.post(f'http://{URL}/api/request-challenge')

encrypted_challenge = b64d(r.json().get('encrypted_challenge'))
iv, ct = encrypted_challenge[:16], encrypted_challenge[16:]

cipher = AES.new(session_key, AES.MODE_CBC, iv)
challenge = unpad(cipher.decrypt(ct), 16)

Enviamos el desafío descifrado y solicitamos la flag:

# Step 4
iv = os.urandom(16)
cipher = AES.new(session_key, AES.MODE_CBC, iv)
packet_data = b64e(iv + cipher.encrypt(pad(b'flag', 16))).decode()

r = requests.post(f'http://{URL}/api/dashboard', json={
    'challenge': sha256(challenge).hexdigest(),
    'packet_data': packet_data
})

packet_data = b64d(r.json().get('packet_data'))

iv, ct = packet_data[:16], packet_data[16:]
cipher = AES.new(session_key, AES.MODE_CBC, iv)
flag = unpad(cipher.decrypt(ct), 16).decode()
print(flag)

Flag

En este punto, si ejecutamos el script, obtendremos la flag:

$ python3 solve.py 94.237.54.116:52507
HTB{Welcome_to_the_fascinating_world_of_hybrid_cryptography!}

El script completo se puede encontrar aquí: solve.py.