lg 시큐리티 해커톤 - simple-ECDSA

타원 곡선 문제였다.
곡선은 NIST P-521이고, nonce를 521비트로 생성해야 하는데 512비트로 생성해 상위 9비트가 빈다. 인터넷을 찾아보며 관련 CVE를 찾았고(CVE-2024-31497), https://github.com/HugoBond/CVE-2024-31497-POC POC 레포지터리도 발견해 키를 얻었다. 이후 메르센 트위스터 state를 k값으로 leak해 다음 k값을 추출했고, 이를 통해 메시지 sign을 수행할 수 있었다.

import socket
import re
import subprocess
import json
import os
from hashlib import sha512
from Crypto.Util.number import bytes_to_long
from ecdsa import NIST521p
import secrets

HOST = '16.184.46.213'
PORT = 3502
NUM_SIGS = 100
B = 1 << 512

PROMPT = b"Enter a message to sign : "


def recv_until(sock, buffer, token):
    while token not in buffer:
        chunk = sock.recv(4096)
        if not chunk:
            raise EOFError('Connection closed')
        buffer += chunk
    idx = buffer.index(token)
    data = buffer[:idx]
    buffer = buffer[idx + len(token):]
    return data, buffer


def collect_signatures(sock):
    buffer = b''
    # Wait for initial prompt
    while PROMPT not in buffer:
        chunk = sock.recv(4096)
        if not chunk:
            raise EOFError('No prompt received')
        buffer += chunk
    buffer = buffer[buffer.index(PROMPT) + len(PROMPT):]

    signatures = []
    for i in range(NUM_SIGS):
        msg = f"msg_{i}"
        sock.sendall(msg.encode() + b"\n")
        data, buffer = recv_until(sock, buffer, PROMPT)
        text = data.decode(errors='ignore')
        m = re.search(r"Signature : \((\d+), (\d+)\)", text)
        if not m:
            raise ValueError(f'Failed to parse signature for {msg!r}: {text!r}')
        r = int(m.group(1))
        s = int(m.group(2))
        h = bytes_to_long(sha512(msg.encode()).digest())
        signatures.append((msg, r, s, h))
    return signatures, buffer


def write_signature_files(signatures):
    baselen = 66  # bytes for P-521
    hex_len = baselen * 2

    with open('signatures_stage1.txt', 'w') as f_plain, open('signatures_poc.txt', 'w') as f_poc:
        for msg, r, s, _ in signatures:
            h_hex = sha512(msg.encode()).hexdigest()
            r_hex = format(r, 'x').zfill(hex_len)
            s_hex = format(s, 'x').zfill(hex_len)
            f_plain.write(f"{msg} {r} {s} {h_hex}\n")
            f_poc.write(f"{h_hex} {r_hex}{s_hex}\n")

    with open('dummy_pubkey.txt', 'w') as f:
        f.write('none')


def recover_private_key():
    result = subprocess.run(
        [
            'sage', '-python', 'CVE-2024-31497-POC/main.py',
            '--signatures', 'signatures_poc.txt',
            '--pubkey', 'dummy_pubkey.txt',
            '--output', 'stage1_priv.pem'
        ],
        capture_output=True,
        text=True,
        timeout=600,
        check=True
    )
    return result.stdout


def parse_private_scalar():
    from ecdsa import SigningKey
    with open('stage1_priv.pem') as f:
        sk = SigningKey.from_pem(f.read())
    return sk.privkey.secret_multiplier


def compute_k_values(signatures, priv):
    n = NIST521p.order
    ks = []
    for msg, r, s, _ in signatures:
        h = bytes_to_long(sha512(msg.encode()).digest())
        k = ((h + r * priv) * pow(s, -1, n)) % n
        ks.append(k)
    return ks


def reconstruct_mt_state(ks):
    outputs = []
    for k in ks:
        for j in range(16):
            outputs.append((k >> (32 * j)) & 0xffffffff)

    WIDTH = 32

    def unshift_right_xor(y, shift):
        result = 0
        for i in range(WIDTH - 1, -1, -1):
            bit = (y >> i) & 1
            if i + shift < WIDTH:
                bit ^= (result >> (i + shift)) & 1
            result |= bit << i
        return result

    def unshift_left_xor_and(y, shift, mask):
        result = 0
        for i in range(WIDTH):
            bit = (y >> i) & 1
            if i - shift >= 0 and ((mask >> i) & 1):
                bit ^= (result >> (i - shift)) & 1
            result |= bit << i
        return result

    u, s, t, l = 11, 7, 15, 18
    b = 0x9d2c5680
    c = 0xefc60000

    def untemper(y):
        y = unshift_right_xor(y, l)
        y = unshift_left_xor_and(y, t, c)
        y = unshift_left_xor_and(y, s, b)
        y = unshift_right_xor(y, u)
        return y & 0xffffffff

    state_values = [int(untemper(y)) for y in outputs[-624:]]
    return state_values


def advance_to_stage2(sock, buffer):
    sock.sendall(b"I'm ready\n")
    data = buffer
    while b"Format : " not in data:
        chunk = sock.recv(4096)
        if not chunk:
            raise EOFError('Connection closed before stage 2 info')
        data += chunk
    text = data.decode()
    msg_match = re.search(r"message : ([0-9a-f]+)\.", text)
    fmt_match = re.search(r"Format : \((\d+), (\d+)\)", text)
    if not msg_match or not fmt_match:
        raise ValueError('Failed to parse stage 2 data:\n' + text)
    msg_hex = msg_match.group(1)
    r_format = int(fmt_match.group(1))
    s_format = int(fmt_match.group(2))
    return msg_hex, r_format, s_format, data[data.index(b"Format : ")+len("Format : (".encode()):]


def compute_stage2_priv(format_bytes, r_format, s_format, k_format):
    n = NIST521p.order
    h_format = bytes_to_long(sha512(format_bytes).digest())
    k_mod = k_format % n
    numerator = (s_format * k_mod - h_format) % n
    priv2 = (numerator * pow(r_format, -1, n)) % n
    return priv2


def sign_message(priv, message_bytes):
    from ecdsa.ellipticcurve import Point
    from ecdsa.curves import NIST521p as Curve
    from ecdsa.util import number_to_string

    curve = Curve
    G = curve.generator
    n = curve.order
    h = bytes_to_long(sha512(message_bytes).digest())

    while True:
        k = secrets.randbelow(n - 1) + 1
        R = k * G
        r = R.x() % n
        if r == 0:
            continue
        s = (pow(k, -1, n) * (h + r * priv)) % n
        if s == 0:
            continue
        return r, s


def main():
    sock = socket.create_connection((HOST, PORT))
    try:
        signatures, buffer = collect_signatures(sock)
        write_signature_files(signatures)
        recover_private_key()
        priv1 = parse_private_scalar()
        ks = compute_k_values(signatures, priv1)
        state_values = reconstruct_mt_state(ks)

        import random
        clone = random.Random()
        clone.setstate((3, tuple(state_values + [624]), None))
        challenge_bytes = clone.randbytes(32)
        format_msg_bytes = clone.randbytes(32)
        k_format = clone.getrandbits(512)

        msg_hex, r_format, s_format, _ = advance_to_stage2(sock, buffer)
        if msg_hex != challenge_bytes.hex():
            raise ValueError('Predicted challenge message mismatch')

        priv2 = compute_stage2_priv(format_msg_bytes, r_format, s_format, k_format)
        r, s = sign_message(priv2, bytes.fromhex(msg_hex))
        sock.sendall(f"({r}, {s})\n".encode())

        response = b''
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk
        print(response.decode(errors='ignore'))
    finally:
        sock.close()


if __name__ == '__main__':
    main()

ai 코드