НаПасеке — avitoctf

Description

A social network hides honey expert kopatych's exact location from non-VIP users. A TCP service and its C source archive are provided, together with an invite code.

The goal is to authenticate as the existing VIP user kopatych and read the full wall post. The service issues a hexadecimal XMSS-MT signed-message token whenever a new nickname is registered.

Analysis

Token and authorization flow

register_user() stores a non-VIP account and signs the raw username. A token is simply the XMSS-MT signed-message byte string encoded as hex. During login, the server verifies the signature, recovers the appended username, and loads that user's database record. Consequently, a valid signature over kopatych grants a session with the existing account's VIP bit; no separate role field needs to be modified.

The decisive source-level bug is in commands.c: registration calls xmssmt_sign_message() but never calls xmssmt_update_signing_key(). Although the latter function exists and safely increments the shared secret-key index, the registration path leaves the state unchanged. Every registration therefore signs with global XMSS-MT index zero.

The supplied implementation uses these parameters:

Parameter	Value
Hash size n	16 bytes
Winternitz base w	256
wots_len1	16
wots_len2	2
Total WOTS+ coordinates	18
XMSS-MT layers d	4
Full height	32
Per-layer tree height	8
Index size	4 bytes
Signature size	1684 bytes

The signature layout is:

index (4) || R (16) || 4 * (WOTS+ signature (18*16) || auth path (8*16))

The username follows these 1684 signature bytes. Reusing index zero also reuses the same bottom WOTS+ private key and address, while R, the authentication paths, and the upper XMSS-MT layers remain identical.

Why WOTS+ reuse permits a forgery

For w = 256, the 16-byte message digest directly supplies the first 16 chain lengths. The final two lengths are the big-endian base-256 representation of the WOTS+ checksum:

checksum = sum(255 - digest_byte)

A WOTS+ signature coordinate for chain length a exposes the chain value at position a. If a target requires position b >= a, that value can be moved forward from a to b. It cannot be moved backward.

One source signature need not satisfy all 18 inequalities. We can generate many candidate usernames offline because the public key gives the root and public seed, and index zero's R is learned from the first token. For each coordinate, choose any candidate with:

source_length[i] <= target_length[i]

Only the union of selected candidates must be registered. In the successful run, seven source registrations covered all coordinates (other runs may need eight). For each coordinate, extract its 16-byte bottom WOTS+ value and advance it to the target position.

Importantly, advancement is not ordinary repeated MD5. WOTS+ uses the address-dependent thash_f: each step derives a key and bitmask from the public seed and an address containing the layer, tree, OTS, chain, hash-step, and key/mask selector. The solver reproduces this exact construction.

After all 18 coordinates are forged, they reconstruct the same bottom WOTS+ public key as a legitimate target signature. The bottom leaf and root therefore remain valid, so the bottom authentication path and every upper XMSS-MT layer can be copied unchanged from one index-zero token.

Why other approaches are unnecessary

• MD5 collisions: MD5 is the internal primitive, but no collision or preimage attack is needed. Chosen usernames provide favorable chain positions, and the one-time-key reuse supplies the actual forgery primitive.
• Alternate ports: The listed ports expose equivalent challenge instances. Registration, collection, forgery, login, and wall retrieval all succeeded against port 30012; cross-port behavior is not required.
• Full XMSS-MT reconstruction: Only the bottom WOTS+ signature changes. The index, R, authentication paths, and upper-layer signatures are copied.

The prior CTFBase writeup 20260622_cryptohack_wots_up was consulted for the general WOTS one-time-key reuse pattern. The exploit here additionally had to reproduce this implementation's address-dependent WOTS+ thash_f and splice the result into an XMSS-MT signature.

Solution

1. Connect to port 30012, submit the invite code, and request the public key.
2. Register one random printable nickname and parse the index-zero token. Split the public key into the root and public seed, and recover R from the token.
3. Compute the index-zero message digest and 18 WOTS+ chain lengths for kopatych.
4. Search random candidate nicknames offline. Assign each target coordinate a candidate whose source chain length is no greater than the target length.
5. Register only the distinct selected candidates and collect their reused index-zero signatures.
6. For each bottom WOTS+ coordinate, apply the exact address-dependent thash_f from its source length to its target length.
7. Replace the bottom WOTS+ block in a source token, preserve the rest of the 1684-byte signature, and append b"kopatych".
8. Log in with the forged token and request the target wall. The service accepts the token as the VIP account and reveals the full post.

The following pure-Python solver requires no local C compilation or OpenSSL headers:

#!/usr/bin/env python3
"""Forge an index-0 XMSS-MT token by exploiting WOTS+ key reuse."""

import argparse
import hashlib
import re
import secrets
import socket
import struct

N = 16
WOTS_LEN = 18
WOTS_BYTES = WOTS_LEN * N
SIG_BYTES = 1684
INDEX_BYTES = 4
BOTTOM_WOTS_OFF = INDEX_BYTES + N
BOTTOM_AUTH_OFF = BOTTOM_WOTS_OFF + WOTS_BYTES
MENU = b"Enter your choice: "


def md5(data):
    return hashlib.md5(data).digest()


def u32(x):
    return struct.pack(">I", x)


def address(chain, hash_index, key_and_mask):
    # layer=0, tree=0, type=OTS(0), ots=0 for reused global index 0.
    words = [0, 0, 0, 0, 0, chain, hash_index, key_and_mask]
    return b"".join(u32(x) for x in words)


def prf(pub_seed, addr_bytes):
    return md5(u32(3) + pub_seed + addr_bytes)


def thash_f(value, pub_seed, chain, hash_index):
    key = prf(pub_seed, address(chain, hash_index, 0))
    mask = prf(pub_seed, address(chain, hash_index, 1))
    return md5(u32(0) + key + bytes(a ^ b for a, b in zip(value, mask)))


def advance(value, pub_seed, chain, start, stop):
    for step in range(start, stop):
        value = thash_f(value, pub_seed, chain, step)
    return value


def hash_message(message, r_value, root, index=0):
    return md5(u32(2) + r_value + root + index.to_bytes(N, "big") + message)


def chain_lengths(digest):
    vals = list(digest)
    checksum = sum(255 - x for x in vals)
    return vals + list(checksum.to_bytes(2, "big"))


class Client:
    def __init__(self, host, port, invite):
        self.sock = socket.create_connection((host, port), timeout=10)
        self.sock.settimeout(20)
        self.recv_until(b"Invite code: ")
        self.sendline(invite)
        self.recv_until(MENU)

    def recv_until(self, marker):
        data = bytearray()
        while marker not in data:
            chunk = self.sock.recv(65536)
            if not chunk:
                raise EOFError(f"connection closed before {marker!r}: {data[-500:]!r}")
            data.extend(chunk)
        return bytes(data)

    def sendline(self, value):
        if isinstance(value, str):
            value = value.encode()
        self.sock.sendall(value + b"\n")

    def command(self, command, answer=None):
        self.sendline(command)
        if answer is not None:
            self.recv_until(b": ")
            self.sendline(answer)
        return self.recv_until(MENU)

    def public_key(self):
        out = self.command("K")
        match = re.search(rb"Public Key \(XMSS-MD5\): ([0-9a-f]{64})", out)
        if not match:
            raise RuntimeError(f"public key not found: {out!r}")
        return bytes.fromhex(match.group(1).decode())

    def register(self, username):
        out = self.command("R", username)
        match = re.search(rb"HoneyPass token: ([0-9a-f]+)", out)
        if not match:
            raise RuntimeError(f"registration failed for {username!r}: {out!r}")
        token = bytes.fromhex(match.group(1).decode())
        if len(token) != SIG_BYTES + len(username):
            raise RuntimeError("unexpected token size")
        return token

    def login_and_read(self, token, wall):
        out = self.command("L", token.hex())
        if b"Login successful" not in out:
            raise RuntimeError(f"forged login failed: {out!r}")
        return self.command("P", wall)


def candidate_name(namespace, counter):
    return ("f%s%011x" % (namespace, counter)).encode()


def solve(host, port, invite, target):
    client = Client(host, port, invite)
    pk = client.public_key()
    root, pub_seed = pk[:N], pk[N:]

    namespace = secrets.token_hex(2)
    bootstrap_name = candidate_name(namespace, 0)
    bootstrap = client.register(bootstrap_name)
    if bootstrap[:INDEX_BYTES] != b"\0" * INDEX_BYTES:
        raise RuntimeError("signing index is not reused at zero")
    r_value = bootstrap[INDEX_BYTES:BOTTOM_WOTS_OFF]

    target_digest = hash_message(target, r_value, root)
    target_lengths = chain_lengths(target_digest)

    chosen = [None] * WOTS_LEN
    chosen_lengths = [None] * WOTS_LEN
    bootstrap_lengths = chain_lengths(hash_message(bootstrap_name, r_value, root))
    for i, length in enumerate(bootstrap_lengths):
        if length <= target_lengths[i]:
            chosen[i] = bootstrap_name
            chosen_lengths[i] = length

    counter = 1
    while any(x is None for x in chosen):
        name = candidate_name(namespace, counter)
        lengths = chain_lengths(hash_message(name, r_value, root))
        for i in range(WOTS_LEN):
            if chosen[i] is None and lengths[i] <= target_lengths[i]:
                chosen[i] = name
                chosen_lengths[i] = lengths[i]
        counter += 1
        if counter > 10_000_000:
            raise RuntimeError("offline candidate search unexpectedly exhausted")

    needed_names = list(dict.fromkeys(chosen))
    tokens = {bootstrap_name: bootstrap}
    for name in needed_names:
        if name not in tokens:
            tokens[name] = client.register(name)

    forged_wots = bytearray()
    for i in range(WOTS_LEN):
        name = chosen[i]
        token = tokens[name]
        if token[:BOTTOM_WOTS_OFF] != bootstrap[:BOTTOM_WOTS_OFF]:
            raise RuntimeError("tokens do not share index/R")
        source = token[BOTTOM_WOTS_OFF + i*N:BOTTOM_WOTS_OFF + (i+1)*N]
        source_len = chosen_lengths[i]
        forged = advance(source, pub_seed, i, source_len, target_lengths[i])

        source_end = advance(source, pub_seed, i, source_len, 255)
        forged_end = advance(forged, pub_seed, i, target_lengths[i], 255)
        assert source_end == forged_end
        forged_wots.extend(forged)

    forged_token = (
        bootstrap[:BOTTOM_WOTS_OFF]
        + bytes(forged_wots)
        + bootstrap[BOTTOM_AUTH_OFF:SIG_BYTES]
        + target
    )
    assert len(forged_token) == SIG_BYTES + len(target)

    wall = client.login_and_read(forged_token, target)
    text = wall.decode(errors="replace")
    print(text)
    flag = re.search(r"avito\{[^}]*\}", text)
    if not flag:
        raise RuntimeError("VIP login worked, but no flag appeared on target wall")
    return flag.group(0)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="napaseke-m9ybjl6b.avitoctf.ru")
    parser.add_argument("--port", type=int, default=30012)
    parser.add_argument("--invite", default="E3MPQmoiWnaRkA35E8yqwg")
    parser.add_argument("--target", default="kopatych")
    args = parser.parse_args()
    solve(args.host, args.port, args.invite.encode(), args.target.encode())


if __name__ == "__main__":
    main()

Run it as:

python3 solve.py

Local compilation of the provided C source was not needed and failed on the analysis host because openssl/evp.h was unavailable. The pure-Python implementation was independently rerun successfully against the service.