Protocol Analysis 7: Mediation — metactf

Description

Alice send: pubA, A, certA, nA Bob recv: pubA, A, certA, nA Bob send: pubB, B, certB, nB, {B, nB, nA}privB Alice recv: pubX, X, certX, nX, {X, nX, nA}privX Alice send: {A, nX, nA}privA Bob recv: {A, nB, nA}privA Bob send: [FLAG]

Source: https://github.com/UMBCCyberDawgs/dawgctf-sp26/blob/main/Protocol%20Analysis%20(1-9)/Protocol_Analysis_chals.pdf

This challenge presents a simple authentication protocol between Alice and Bob. The goal is to drive the live service, satisfy Bob's authentication checks, and make him release the flag without possessing Alice's private key.

Recon

The service lives at https://protocols.live.

• POST /model/7 creates a fresh protocol instance and returns a conn_id.
• The same conn_id must be reused for both /alice and /bob inside that instance.
• Empty content to /alice triggers Alice's first send.
• Messages are pipe-delimited typed fields such as k:<hex>, n:<name>, d:<hex>, t:<text>.

Observed message shapes:

Alice first message:
k:<pubA>|n:alice|d:<certA>|d:<nA>

Bob reply:
k:<pubB>|n:bob|d:<certB>|d:<nB>|d:<sigB>

One subtle but important quirk is that signatures are verified over the literal typed-text framing, not over a reconstructed abstract tuple. Bob's signature verifies against:

t:n:bob|d:<nB>|d:<nA>

So when forging the attacker-controlled message for Alice, we must also sign the exact literal form:

t:n:mallory|d:<nB>|d:<nA>

Vulnerability

This protocol is vulnerable to an identity misbinding / mediation attack.

Bob signs (B, nB, nA), but Alice does not include Bob's identity, Bob's public key, or Bob's certificate in what she later signs. Alice only signs (A, nX, nA). That means an attacker can:

1. Relay Alice's initial cleartext message to Bob.
2. Learn Bob's nonce nB from Bob's plaintext response.
3. Present Bob's nonce to Alice inside an attacker-controlled, properly certified identity X = mallory.
4. Receive Alice's valid signature on (alice, nB, nA).
5. Forward that signature to Bob, who accepts it as if it were intended for him.

The protocol authenticates that Alice signed some nonce pair, but it fails to bind which responder that signature was meant for.

Exploitation

1. Start a fresh instance and obtain Mallory credentials

Generate an attacker keypair with /util/gen_asym_key_pair, then request a valid certificate for mallory using /util/get_cert.

2. Trigger Alice and capture her opener

Send empty content to /alice to get:

k:<pubA>|n:alice|d:<certA>|d:<nA>

This gives us Alice's public key, certificate, and nonce nA.

3. Relay Alice's first message to Bob

Forward Alice's exact first message to /bob. Bob responds with:

k:<pubB>|n:bob|d:<certB>|d:<nB>|d:<sigB>

Now we know Bob's nonce nB. Bob's signature verifies over literal text t:n:bob|d:<nB>|d:<nA>.

4. Ask Alice to authenticate Mallory using Bob's nonce

Use /util/asym_sign with Mallory's private key to sign:

t:n:mallory|d:<nB>|d:<nA>

Then send Alice:

k:<pubM>|n:mallory|d:<certM>|d:<nB>|d:<sigM>

Alice accepts the certified attacker identity, treats nB as Mallory's nonce, and returns:

d:<sigA>

This verifies as Alice's signature over:

t:n:alice|d:<nB>|d:<nA>

5. Forward Alice's signature to Bob

Send d:<sigA> to /bob. Bob accepts it as the expected {A, nB, nA}privA and returns:

t:DawgCTF{REDACTED}

The flag is stored in the service as a typed text item, so the raw response includes the t: prefix.

Full solver

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

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


def post(path, body, retries=12):
    for _ in range(retries):
        try:
            r = requests.post(f"{BASE}{path}", json=body, timeout=TIMEOUT)
        except requests.exceptions.RequestException:
            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 SystemExit(f"{path} failed: HTTP {r.status_code} {r.text}")
    raise SystemExit(f"{path} retries exhausted")


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


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


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


def attempt():
    kp = parse(post("/util/gen_asym_key_pair", {"conn_id": 0, "content": ""}))
    pub_m, priv_m = kp[1][1], kp[3][1]
    cert_m = first(
        parse(
            post("/util/get_cert", {"conn_id": 0, "content": f"k:{pub_m}|n:mallory"})
        ),
        "d",
    )

    conn = post("/model/7", {})["conn_id"]
    time.sleep(1)

    a1 = post("/alice", {"conn_id": conn, "content": ""})
    pa = parse(a1)
    pub_a = first(pa, "k")
    name_a = first(pa, "n")
    cert_a, n_a = all_of(pa, "d")

    b1 = post("/bob", {"conn_id": conn, "content": a1})
    pb = parse(b1)
    pub_b = first(pb, "k")
    name_b = first(pb, "n")
    cert_b, n_b, sig_b = all_of(pb, "d")

    bob_ok = post(
        "/util/asym_verify",
        {"conn_id": 0, "content": f"k:{pub_b}|d:{sig_b}|t:n:{name_b}|d:{n_b}|d:{n_a}"},
    )
    if bob_ok != "d:True":
        raise RuntimeError("Bob signature framing guess failed")

    sig_m = first(
        parse(
            post(
                "/util/asym_sign",
                {"conn_id": 0, "content": f"k:{priv_m}|t:n:mallory|d:{n_b}|d:{n_a}"},
            )
        ),
        "d",
    )

    a2 = post(
        "/alice",
        {
            "conn_id": conn,
            "content": f"k:{pub_m}|n:mallory|d:{cert_m}|d:{n_b}|d:{sig_m}",
        },
    )
    sig_a = first(parse(a2), "d")

    alice_ok = post(
        "/util/asym_verify",
        {"conn_id": 0, "content": f"k:{pub_a}|d:{sig_a}|t:n:{name_a}|d:{n_b}|d:{n_a}"},
    )
    if alice_ok != "d:True":
        raise RuntimeError("Alice signature verification failed")

    out = post("/bob", {"conn_id": conn, "content": f"d:{sig_a}"})
    return first(parse(out), "t")


def main():
    for i in range(5):
        try:
            print(f"attempt {i + 1}")
            print(attempt())
            return
        except Exception as e:
            print(f"retrying after error: {e}")
            time.sleep(3)
    raise SystemExit("failed")


if __name__ == "__main__":
    main()