Protocol Analysis 9: Oracle — metactf

Description

Final challenge. In this one, Alice will stay in her last recv-send state infinitely. Can you get the flag?

English summary: the service exposes a live certificate-based Alice/Bob protocol at https://protocols.live. Bob sends a flag protected inside nested ciphertexts for Alice, and the goal is to recover it without Alice's private key by exploiting the protocol logic.

Analysis

The relevant protocol behavior is:

1. Alice sends pubA, alice, and certA.
2. Bob replies with pubB, bob, certB, and an outer ciphertext encrypted to Alice.
3. That outer ciphertext contains two fields: the inner ciphertext of the flag encrypted to Alice, and Bob's name.
4. Alice's final receive/send state loops forever instead of terminating.

In practice, this turns Alice into a reusable decryption + re-encryption oracle.

If we send Alice a certified message under our own identity, she will:

• decrypt attacker-controlled ciphertext addressed to Alice;
• parse the resulting plaintext as d:<something>|n:<recipient>;
• decrypt the nested d: payload if needed as part of her normal state handling; and
• re-encrypt the resulting plaintext to the public key named in the attacker-supplied certified identity.

That behavior is fatal because Bob's secret is wrapped as nested ciphertexts for Alice. Since Alice never leaves her final state, we can ask her to peel one layer, then a second layer, and have each plaintext re-encrypted to Mallory.

Conceptually, Bob sends:

C_outer = Enc_A(d:Enc_A(flag)|n:bob)

So the challenge is really a chosen-ciphertext oracle attack caused by bad protocol state management.

Solution

1. Create a Mallory keypair and get a valid certificate for the identity mallory.
2. Start a fresh /model/9 instance, trigger Alice's opener, and forward it to Bob.
3. Capture Bob's outer ciphertext C_outer.
4. Encrypt d:C_outer|n:mallory to Alice and send it to Alice with Mallory's certified identity.
5. Alice decrypts Bob's outer blob and re-encrypts its plaintext to Mallory, revealing d:Enc_A(flag)|n:bob after Mallory decrypts the response.
6. Extract Enc_A(flag) from that plaintext.
7. Encrypt d:Enc_A(flag)|n:mallory to Alice and send it again with Mallory's certified identity.
8. Alice decrypts the inner flag ciphertext and re-encrypts the plaintext flag to Mallory.
9. Decrypt Alice's response with Mallory's private key to recover the flag.

The attack works because Alice's final state is both reusable and able to transform ciphertexts for Alice into ciphertexts for an attacker-chosen recipient. Bob's nested encryption only forces us to query the oracle twice.

#!/usr/bin/env python3
import json
import time

import requests


BASE = "https://protocols.live"
TIMEOUT = 60


def raw_post(path, body):
    try:
        return requests.post(
            f"{BASE}{path}",
            headers={"Content-Type": "application/json"},
            data=json.dumps(body),
            timeout=TIMEOUT,
        )
    except requests.exceptions.RequestException:
        return None


def api_post(path, body, retries=15):
    for _ in range(retries):
        r = raw_post(path, body)
        if r is None:
            time.sleep(2)
            continue
        if r.status_code == 200:
            data = r.json()
            return data.get("content", data)
        if "DB access error" in r.text or r.status_code >= 500:
            time.sleep(2)
            continue
        raise RuntimeError(f"{path} failed: HTTP {r.status_code} {r.text}")
    raise RuntimeError(f"{path} retries exhausted")


def parse(msg):
    if not msg:
        return []
    return [tuple(item.split(":", 1)) for item in msg.split("|")]


def first(items, typ):
    return next(v for t, v in items if t == typ)


def all_of(items, typ):
    return [v for t, v in items if t == typ]


def util(name, content):
    return api_post(f"/util/{name}", {"conn_id": 0, "content": content})


def new_conn():
    return api_post("/model/9", {})["conn_id"]


def asym_encrypt(pubkey, plaintext):
    return first(parse(util("asym_encrypt", f"k:{pubkey}|t:{plaintext}")), "d")


def asym_decrypt(privkey, ciphertext):
    return util("asym_decrypt", f"k:{privkey}|d:{ciphertext}")


def generate_identity(name="mallory"):
    keys = parse(util("gen_asym_key_pair", ""))
    pub, priv = all_of(keys, "k")
    cert = first(parse(util("get_cert", f"k:{pub}|n:{name}")), "d")
    return name, pub, priv, cert


def send_to_alice(conn_id, pub_x, name_x, cert_x, outer_cipher):
    content = f"k:{pub_x}|n:{name_x}|d:{cert_x}|d:{outer_cipher}|n:alice"
    return api_post("/alice", {"conn_id": conn_id, "content": content})


def unwrap_alice_response(priv_x, response):
    outer = all_of(parse(response), "d")[1]
    outer_plain = parse(asym_decrypt(priv_x, outer))
    inner = first(outer_plain, "d")
    return asym_decrypt(priv_x, inner)


def attempt():
    name_m, pub_m, priv_m, cert_m = generate_identity("mallory")

    conn = new_conn()
    time.sleep(1)

    alice_hello = api_post("/alice", {"conn_id": conn, "content": ""})
    pa = parse(alice_hello)
    pub_a = first(pa, "k")
    print(f"[+] alice opener: {alice_hello}")

    bob_msg = api_post("/bob", {"conn_id": conn, "content": alice_hello})
    pb = parse(bob_msg)
    outer_from_bob = all_of(pb, "d")[1]
    print(f"[+] bob message:   {bob_msg}")

    # Round 1: ask Alice to decrypt Bob's outer blob and re-encrypt its plaintext to Mallory.
    oracle_1 = asym_encrypt(pub_a, f"d:{outer_from_bob}|n:{name_m}")
    alice_resp_1 = send_to_alice(conn, pub_m, name_m, cert_m, oracle_1)
    inner_plain_1 = unwrap_alice_response(priv_m, alice_resp_1)
    print(f"[+] oracle #1 plaintext: {inner_plain_1}")
    enc_flag_for_alice = first(parse(inner_plain_1), "d")

    # Round 2: now ask Alice to decrypt the extracted inner flag ciphertext and re-encrypt the flag to Mallory.
    oracle_2 = asym_encrypt(pub_a, f"d:{enc_flag_for_alice}|n:{name_m}")
    alice_resp_2 = send_to_alice(conn, pub_m, name_m, cert_m, oracle_2)
    flag_plain = unwrap_alice_response(priv_m, alice_resp_2)
    print(f"[+] oracle #2 plaintext: {flag_plain}")
    return first(parse(flag_plain), "t")


def main():
    for i in range(8):
        print(f"=== Attempt {i + 1} ===")
        try:
            flag = attempt()
            print(f"FLAG: {flag}")
            return
        except Exception as e:
            print(f"[!] retrying after error: {e}")
            time.sleep(3)
    raise SystemExit("failed")


if __name__ == "__main__":
    main()