Writeup - НаПасеке

nc napaseke-m9ybjl6b.avitoctf.ru 30012
Invite code: uCQq3YsK7TmoZBlN4UYkng

Разведка

После подключения сервис предлагает следующие действия:

• просмотреть обычных пользователей;
• просмотреть VIP-пользователей;
• зарегистрироваться;
• войти по токену HoneyPass;
• просмотреть стену пользователя;
• получить открытый ключ.

При регистрации сервер возвращает токен следующего вида:

HoneyPass token: <hex-строка>

Токен представляет собой подписанное имя пользователя. В качестве алгоритма электронной подписи используется XMSSMT:

unsigned char msg_to_sign[MAX_USERNAME_LEN];
int msg_len = strlen(username);
memcpy(msg_to_sign, username, msg_len);

xmssmt_sign_message(
    msg_to_sign,
    msg_len,
    xmss_sig,
    sizeof(xmss_sig),
    &sig_len,
    &signing_index
);

При авторизации сервер проверяет подпись и извлекает из неё имя пользователя:

if (xmssmt_verify_signature(
        token.signature,
        token.sig_len,
        msg_verified,
        sizeof(msg_verified),
        &msg_verified_len
    ) != 0) {
    write_all(sockfd, "Invalid token signature\n");
    return;
}

После успешной проверки профиль загружается из базы данных:

if (!load_user_data(username, &user_data)) {
    write_all(sockfd, "Profile not found\n");
    return;
}

Таким образом, для входа под пользователем kopatych необходимо получить корректную XMSS-подпись строки:

kopatych

Проверка VIP-ограничения

В функции просмотра стены проверяется не статус владельца стены, а статус текущего авторизованного пользователя:

int can_view_full = session->logged_in && session->user.is_vip;

if (can_view_full) {
    strncpy(display_msg, message, sizeof(display_msg) - 1);
} else {
    build_wall_preview(message, display_msg, sizeof(display_msg));
}

Обычный пользователь видит только первые десять символов публикации:

if (strlen(message) > 10) {
    strncpy(display_msg, message, 10);
    display_msg[10] = '\0';
    strcat(display_msg, "...");
}

Следовательно, есть два возможных пути:

1. войти под любым VIP-пользователем;
2. подделать токен непосредственно для kopatych.

Второй вариант удобнее, поскольку имя kopatych уже существует в базе данных и является VIP-профилем.

Анализ реализации XMSS

XMSS — это stateful hash-based signature scheme. Внутри неё используются одноразовые подписи WOTS+.
Критически важное требование XMSS: каждый одноразовый ключ WOTS+ должен использоваться только один раз. После создания подписи индекс секретного ключа должен увеличиваться.
В исходниках для этого даже реализована отдельная функция:

int xmssmt_update_signing_key(unsigned long long *signing_index) {
    unsigned long long current_index;
    unsigned long long next_index;

    current_index = signing_key_index(
        &g_xmss_state->xmss_params,
        g_xmss_state->xmss_sk
    );

    if (current_index == *signing_index) {
        if (xmssmt_update_sk(
                &g_xmss_state->xmss_params,
                g_xmss_state->xmss_sk
            ) != 0) {
            return -1;
        }

        next_index = signing_key_index(
            &g_xmss_state->xmss_params,
            g_xmss_state->xmss_sk
        );
    }

    return 0;
}

Однако вызов xmssmt_update_signing_key() в программе отсутствует.
Поиск по исходникам показывает, что функция определена, но нигде не используется:

xmss_keys.c: xmssmt_update_signing_key(...)
xmss_keys.h: xmssmt_update_signing_key(...)

При каждой регистрации выполняется только:

xmssmt_core_sign(
    &g_xmss_state->xmss_params,
    g_xmss_state->xmss_sk,
    sm,
    &smlen,
    msg,
    mlen
);

В результате один и тот же XMSS-индекс и один и тот же WOTS+-ключ используются для подписи множества разных имён пользователей.
Это и является основной уязвимостью задания.

Почему повторное использование WOTS+ опасно

WOTS+ состоит из набора хеш-цепочек. Для каждого элемента сообщения выбирается определённая позиция в соответствующей цепочке.
Упрощённо подпись одного элемента выглядит так:

signature[i] = F^a(secret[i])

где:
secret[i] — секретное начало цепочки;
F — односторонняя хеш-функция;
a — значение, полученное из хеша сообщения.
Проверяющая сторона продолжает вычисления до конца цепочки:

public[i] = F^(w - 1 - a)(signature[i])

Если известна подпись для значения a, можно получить подпись для любого большего значения b:

F^(b - a)(signature[i]) = F^b(secret[i])

Вернуться назад по цепочке невозможно, но двигаться вперёд можно.
Следовательно, если один WOTS+-ключ используется для подписания большого количества сообщений, для каждой цепочки можно выбрать подпись с минимальной известной позицией. После этого из неё можно вычислить подпись почти для любого другого сообщения.

Формат подписи

Размер XMSS-подписи в задании:

#define XMSS_SIG_SIZE 1684

Открытый ключ имеет размер 32 байта:

#define XMSS_PK_SIZE 32

Он состоит из:

root     — 16 байт
pub_seed — 16 байт

Начало подписи имеет следующий формат:

offset  size   назначение
0       4      XMSS index
4       16     случайное значение R
20      288    WOTS+ подпись нижнего уровня

Параметры WOTS+:

n        = 16
w        = 256
len_1    = 16
len_2    = 2
WOTS_LEN = 18

Размер WOTS+-части:

18 × 16 = 288 байт

Таким образом:

N = 16
WOTS_LEN = 18
WOTS_OFFSET = 4 + N
WOTS_BYTES = WOTS_LEN * N

Проверка повторного использования ключа

Регистрируем несколько пользователей и разбираем полученные токены.
Каждый токен состоит из XMSS-подписи и подписанного сообщения:

token = signature || username

Для каждого токена извлекаем индекс и значение R:

sig = token[:1684]

idx = int.from_bytes(sig[:4], "big")
R = sig[4:20]

У всех полученных подписей значения совпадают:

idx одинаковый
R одинаковый

Одинаковый idx подтверждает повторное использование одного листа XMSS и, следовательно, одного нижнего WOTS+-ключа.
Одинаковое значение R дополнительно упрощает атаку: для вычисления хеша целевого сообщения можно использовать параметры любой зарегистрированной подписи.

Вычисление хеша сообщения

В данной реализации используется MD5 с доменным разделением.
Хеш подписываемого сообщения вычисляется как:

H_msg = MD5(
    toByte(2, 4) ||
    R ||
    root ||
    index ||
    message
)

Реализация на python

def message_digest(root: bytes, idx: int, R: bytes, msg: bytes) -> bytes:
    return hashlib.md5(
        struct.pack(">I", 2)
        + R
        + root
        + idx.to_bytes(16, "big")
        + msg
    ).digest()

Так как n = 16, хеш содержит 16 байт. При w = 256 каждый байт непосредственно задаёт длину одной WOTS-цепочки.
К ним добавляется двухбайтовая контрольная сумма:

def chain_lengths(digest: bytes):
    values = list(digest)

    checksum = sum(255 - value for value in values)
    values.extend(checksum.to_bytes(2, "big"))

    return values

В итоге для каждого сообщения получается 18 длин цепочек.
8. Сбор элементов WOTS+
Регистрируем большое количество случайных пользователей. Для каждого имени:
получаем токен;
вычисляем хеш подписанного имени;
определяем 18 длин WOTS-цепочек;
извлекаем 18 элементов из подписи;
для каждой позиции сохраняем элемент с минимальной длиной цепочки.
Логика сбора:

best = [None] * WOTS_LEN

for j, length in enumerate(lengths):
    element = wots[j * N:(j + 1) * N]

    if best[j] is None or length < best[j][0]:
        best[j] = (length, element)

Для целевого сообщения kopatych также вычисляем необходимые длины:

target_digest = message_digest(
    root,
    common_idx,
    common_R,
    b"kopatych"
)

target_lengths = chain_lengths(target_digest)

Для успешной подделки подписи должно выполняться условие для всех 18 цепочек.

known_length[i] <= target_length[i]

Если для какой-либо позиции сохранённый элемент находится дальше по цепочке, чем требуется для kopatych, нужно продолжать регистрацию пользователей.
Поскольку значения байтов хеша распределены практически равномерно, через некоторое количество регистраций для каждой позиции находится достаточно ранний элемент цепочки.

Продолжение WOTS-цепочек

Для преобразования найденного элемента в элемент подписи kopatych необходимо продолжить соответствующую хеш-цепочку.
В WOTS+ хеширование зависит не только от текущего значения, но и от адреса элемента.
Адрес содержит:

• слой XMSSMT;
• индекс дерева;
• индекс листа;
• номер WOTS-цепочки;
• текущий шаг цепочки;
• тип хеширования.

Реализация функции F:

def prf(pub_seed: bytes, addr_words) -> bytes:
    addr = b"".join(struct.pack(">I", x) for x in addr_words)
    return hashlib.md5(
        struct.pack(">I", 3)
        + pub_seed
        + addr
    ).digest()
def thash_f(value: bytes, pub_seed: bytes, addr_words) -> bytes:
    addr = list(addr_words)

    addr[7] = 0
    key = prf(pub_seed, addr)

    addr[7] = 1
    mask = prf(pub_seed, addr)

    masked = bytes(
        x ^ y for x, y in zip(value, mask)
    )

    return hashlib.md5(
        struct.pack(">I", 0)
        + key
        + masked
    ).digest()

Для каждого элемента подписи выполняем необходимое количество шагов:

for j, target_length in enumerate(target_lengths):
    start_length, value = best[j]

    address = [
        0,
        (idx_tree >> 32) & 0xffffffff,
        idx_tree & 0xffffffff,
        0,
        idx_leaf,
        j,
        0,
        0
    ]

    for step in range(start_length, target_length):
        address[6] = step
        value = thash_f(value, pub_seed, address)

    offset = WOTS_OFFSET + j * N
    forged_signature[offset:offset + N] = value
    

Почему достаточно изменить только нижнюю WOTS-подпись

XMSSMT-подпись содержит несколько уровней дерева.
Можно было бы предположить, что для подделки сообщения потребуется заново построить:

• WOTS-подписи всех уровней;
• пути аутентификации;
• корни промежуточных деревьев.

Однако это не требуется.
Повторно используется тот же XMSS-индекс. После модификации нижней WOTS+-подписи она всё равно восстанавливает тот же открытый WOTS-ключ, что и легитимная подпись.
Следовательно:

• лист дерева остаётся прежним;
• путь аутентификации нижнего дерева остаётся валидным;
• корень нижнего дерева не изменяется;
• подписи верхних уровней остаются валидными.

Поэтому из любого легитимного токена можно сохранить:

• индекс;
• R;
• пути аутентификации;
• WOTS-подписи верхних уровней.

Заменить необходимо только первые 288 байт WOTS+-части, начиная со смещения 20.

Формирование поддельного токена

Берём любую легитимную подпись как основу:

forged_signature = bytearray(base_signature)

Заменяем в ней нижнюю WOTS+-часть, после чего добавляем целевое имя пользователя:

forged_token = bytes(forged_signature) + b"kopatych"

Токен отправляется в функцию авторизации в hex-представлении:

client.login(forged_token)

Сервер успешно проверяет подпись и извлекает имя:

Login successful

Затем запрашиваем стену пользователя:

P
kopatych

Поскольку сессия теперь принадлежит VIP-пользователю kopatych, сервер возвращает полное содержимое публикации, включая искомую локацию.

Эксплойт

#!/usr/bin/env python3
import argparse
import hashlib
import os
import random
import socket
import string
import struct
import sys
import time

INVITE_DEFAULT = 'uCQq3YsK7TmoZBlN4UYkng'
XMSS_SIG_SIZE = 1684
N = 16
WOTS_LEN = 18
WOTS_OFFSET = 4 + N
WOTS_BYTES = WOTS_LEN * N


def md5(data: bytes) -> bytes:
    return hashlib.md5(data).digest()


def u32(x: int) -> bytes:
    return struct.pack('>I', x & 0xffffffff)


def prf(pub_seed: bytes, addr_words) -> bytes:
    addr = b''.join(u32(x) for x in addr_words)
    return md5(u32(3) + pub_seed + addr)


def thash_f(value: bytes, pub_seed: bytes, addr_words) -> bytes:
    a = list(addr_words)
    a[7] = 0
    key = prf(pub_seed, a)
    a[7] = 1
    mask = prf(pub_seed, a)
    return md5(u32(0) + key + bytes(x ^ y for x, y in zip(value, mask)))


def message_digest(root: bytes, idx: int, R: bytes, msg: bytes) -> bytes:
    return md5(u32(2) + R + root + idx.to_bytes(N, 'big') + msg)


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


def recv_until(sock: socket.socket, marker: bytes, timeout=20) -> bytes:
    sock.settimeout(timeout)
    data = bytearray()
    while marker not in data:
        chunk = sock.recv(65536)
        if not chunk:
            raise EOFError(f'connection closed while waiting for {marker!r}')
        data += chunk
    return bytes(data)


class Client:
    def __init__(self, host, port, invite):
        self.s = socket.create_connection((host, port), timeout=10)
        recv_until(self.s, b'Invite code: ')
        self.s.sendall(invite.encode() + b'\n')
        recv_until(self.s, b'Enter your choice: ')

    def cmd(self, letter: str, prompt: bytes | None = None, answer: str | None = None) -> bytes:
        self.s.sendall(letter.encode() + b'\n')
        data = b''
        if prompt is not None:
            data = recv_until(self.s, prompt)
            self.s.sendall(answer.encode() + b'\n')
        data += recv_until(self.s, b'Enter your choice: ')
        return data

    def get_public_key(self) -> bytes:
        out = self.cmd('K').decode(errors='replace')
        line = next(x for x in out.splitlines() if 'Public Key' in x)
        return bytes.fromhex(line.rsplit(': ', 1)[1].strip())

    def list_vips(self) -> str:
        return self.cmd('V').decode(errors='replace')

    def register(self, username: str) -> bytes:
        out = self.cmd('R', b'Nickname (max 16 chars): ', username).decode(errors='replace')
        for line in out.splitlines():
            if 'HoneyPass token: ' in line:
                return bytes.fromhex(line.split('HoneyPass token: ', 1)[1].strip())
        raise RuntimeError(f'failed to register {username!r}: {out}')

    def login(self, token: bytes) -> str:
        return self.cmd('L', b'HoneyPass token: ', token.hex()).decode(errors='replace')

    def peek(self, username: str) -> str:
        return self.cmd('P', b'or another nickname): ', username).decode(errors='replace')


def random_username(counter: int) -> str:
    # Printable, unique, at most 16 characters.
    suffix = format(counter, 'x')
    rnd = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(15-len(suffix)))
    return ('x' + rnd + suffix)[:16]


def forge(host, port, invite, target, max_samples):
    c = Client(host, port, invite)
    print('[+] Connected')
    print(c.list_vips().strip())

    pk = c.get_public_key()
    if len(pk) != 32:
        raise RuntimeError(f'unexpected public key length: {len(pk)}')
    root, pub_seed = pk[:16], pk[16:]
    print(f'[+] Public root: {root.hex()}')
    print(f'[+] Public seed: {pub_seed.hex()}')

    best = [None] * WOTS_LEN  # (chain_length, 16-byte WOTS element)
    base_sig = None
    common_idx = None
    common_R = None

    for i in range(max_samples):
        username = random_username(i)
        signed = c.register(username)
        if len(signed) != XMSS_SIG_SIZE + len(username):
            raise RuntimeError(f'unexpected signed token length {len(signed)}')
        sig, msg = signed[:XMSS_SIG_SIZE], signed[XMSS_SIG_SIZE:]
        idx = int.from_bytes(sig[:4], 'big')
        R = sig[4:20]

        if base_sig is None:
            base_sig = bytearray(sig)
            common_idx, common_R = idx, R
        elif idx != common_idx or R != common_R:
            raise RuntimeError('signing index/R changed; reconnect to a fresh instance or use concurrent collection')

        digest = message_digest(root, idx, R, msg)
        lengths = chain_lengths(digest)
        wots = sig[WOTS_OFFSET:WOTS_OFFSET + WOTS_BYTES]
        for j, length in enumerate(lengths):
            elem = wots[j*N:(j+1)*N]
            if best[j] is None or length < best[j][0]:
                best[j] = (length, elem)

        target_digest = message_digest(root, common_idx, common_R, target.encode())
        target_lengths = chain_lengths(target_digest)
        missing = [j for j in range(WOTS_LEN) if best[j] is None or best[j][0] > target_lengths[j]]
        if i % 25 == 0 or not missing:
            mins = ','.join('--' if x is None else f'{x[0]:02x}' for x in best)
            print(f'[*] samples={i+1:4d}, missing={missing}, minima={mins}')
        if not missing:
            break
    else:
        raise RuntimeError(f'not enough samples after {max_samples}; increase --max-samples')

    target_digest = message_digest(root, common_idx, common_R, target.encode())
    target_lengths = chain_lengths(target_digest)
    print(f'[+] Target @{target} digest: {target_digest.hex()}')
    print('[+] Target chain lengths:', target_lengths)

    idx_tree = common_idx >> 8
    idx_leaf = common_idx & 0xff
    for j, target_len in enumerate(target_lengths):
        start_len, value = best[j]
        addr = [0, (idx_tree >> 32) & 0xffffffff, idx_tree & 0xffffffff, 0,
                idx_leaf, j, 0, 0]
        for step in range(start_len, target_len):
            addr[6] = step
            value = thash_f(value, pub_seed, addr)
        off = WOTS_OFFSET + j*N
        base_sig[off:off+N] = value

    forged = bytes(base_sig) + target.encode()
    print(f'[+] Forged token: {forged.hex()}')
    login_out = c.login(forged)
    print(login_out.strip())
    if 'Login successful' not in login_out:
        raise RuntimeError('forged login failed')
    wall = c.peek(target)
    print('\n[+] Full wall output:\n' + wall)


if __name__ == '__main__':
    ap = argparse.ArgumentParser(description='NaPaseke XMSS/WOTS one-time-key reuse exploit')
    ap.add_argument('host', nargs='?', default='napaseke-m9ybjl6b.avitoctf.ru')
    ap.add_argument('port', nargs='?', type=int, default=30012)
    ap.add_argument('--invite', default=INVITE_DEFAULT)
    ap.add_argument('--target', default='kopatych')
    ap.add_argument('--max-samples', type=int, default=2000)
    args = ap.parse_args()
    forge(args.host, args.port, args.invite, args.target, args.max_samples)
    

Запуск

python3 solve_napaseke.py \
    napaseke-m9ybjl6b.avitoctf.ru \
    30012

Если основной инстанс недоступен, можно использовать дополнительные порты:

python3 solve_napaseke.py napaseke-m9ybjl6b.avitoctf.ru 30112
python3 solve_napaseke.py napaseke-m9ybjl6b.avitoctf.ru 30212
python3 solve_napaseke.py napaseke-m9ybjl6b.avitoctf.ru 30312
python3 solve_napaseke.py napaseke-m9ybjl6b.avitoctf.ru 30412

После накопления достаточного количества подписей эксплойт выводит:

Login successful

А затем полное содержимое стены пользователя kopatych, в котором указана искомая локация.

Итог

Причиной уязвимости стало нарушение основного требования stateful hash-based signatures: одноразовый WOTS+-ключ был использован для подписи множества различных сообщений.
Функция обновления состояния секретного ключа была реализована, но не вызывалась после создания подписи. Поэтому все зарегистрированные пользователи получали токены, сформированные с одним XMSS-индексом.
Собрав несколько таких подписей, можно:
определить позиции элементов в WOTS+-цепочках;
выбрать наиболее ранний известный элемент каждой цепочки;
продолжить цепочки до значений, необходимых для строки kopatych;
заменить нижнюю WOTS+-часть легитимной XMSS-подписи;
сохранить остальные уровни XMSSMT без изменений;
сформировать валидный токен пользователя kopatych;
войти под VIP-профилем и прочитать закрытую публикацию.
Уязвимость классифицируется как повторное использование одноразового ключа в WOTS+/XMSS, приводящее к подделке цифровой подписи и обходу механизма аутентификации.