hybrid unifier

6 minutes to read

We are given a Flask project of the server that has the 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

We are also given a PDF document with a kind of API documentation of the server’s endpoints. This is the relevant part:

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.

Therefore, the objective of this challenge is to code a client that requests the flag from the server API. We will need to understand how the encryption protocol works in order to implement it from scratch.

Source code analysis

There are two relevant files: views.py and crypto/session.py. The former is used to manage API endpoints, whereas the latter contains the cryptographic protocol implementation.

The server creates a new SecureSession object at the start, and exposes the public information in /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

These parameters correspond to a Diffie-Hellman Key-Exchange protocol:

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)

In short, the server chooses a public prime number and a generator . The private key is chosen randomly with . Then, the public key is .

At this point, a client must take the protocol parameters and , compute a private key and its associated public key . The client sends the value to the server, and at this point, they both share a secret :

Now, both client and server can use the shared to derive a key for a symmetric encryption algorithm, such as AES.

This is safe because it relies on the Discrete Logarithm Problem, which is computationally hard to solve if the parameters are correctly chosen. An adversary would have to solve for given .

Protocol

As said on the previous documentation, we need to start with /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)})

Here, we must send our public key (the value of the previous example).

The server will create a session key using 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()

This method computes , which is the shared secret, and uses SHA256 to derive a 32-byte session key.

Now, the server requires us to solve a challenge:

# 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()})

This is a way to validate that there are no errors on the communication protocol. This is 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)

If the key exchange worked correctly, we should be able to derive the session key from the shared secret and decrypt the challenge.

The server verifies the challenge in /api/dashboard before performing the action specified in 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.'))

If everything is correct, we should be able to send a flag command in order to get the flag.

Solution

We only need to follow the previous steps:

We take the protocol parameters:

# 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)

We compute a private key and send the public key:

# 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()

We derive the session key from the shared secret and request a challenge:

# 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)

We send the decrypted challenge and request the 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

At this point, if we execute the script, we will get the flag:

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

The full script can be found in here: solve.py.