Mind In The Clouds

17 minutos de lectura

Se nos proporciona el código fuente en Python del servidor que contiene la flag:

import json
import signal
import subprocess
import socketserver
from hashlib import sha1
from random import randint
from Crypto.Util.number import bytes_to_long, long_to_bytes, inverse
from ecdsa.ecdsa import curve_256, generator_256, Public_key, Private_key, Signature
import os


fnames = [b'subject_kolhen', b'subject_stommb', b'subject_danbeer']
nfnames = []


class ECDSA:
    def __init__(self):
        self.G = generator_256
        self.n = self.G.order()
        self.key = randint(1, self.n - 1)
        self.pubkey = Public_key(self.G, self.key * self.G)
        self.privkey = Private_key(self.pubkey, self.key)

    def sign(self, fname):
        h = sha1(fname).digest()
        nonce = randint(1, self.n - 1)
        sig = self.privkey.sign(bytes_to_long(h), nonce)
        return {"r": hex(sig.r)[2:], "s": hex(sig.s)[2:], "nonce": hex(nonce)[2:]}

    def verify(self, fname, r, s):
        h = bytes_to_long(sha1(fname).digest())
        r = int(r, 16)
        s = int(s, 16)
        sig = Signature(r, s)

        if self.pubkey.verifies(h, sig):
            return retrieve_file(fname)
        else:
            return 'Signature is not valid\n'


ecc = ECDSA()


def init_storage():
    i = 0
    for fname in fnames[:-1]:
        data = ecc.sign(fname)
        r, s = data['r'], data['s']
        nonce = data['nonce']
        nfname = fname.decode() + '_' + r + '_' + s + '_' + nonce[(14 + i):-14]
        nfnames.append(nfname)
        i += 2


def retrieve_file(fname):
    try:
        dt = open(fname, 'rb').read()
        return dt.hex()
    except:
        return 'The file does not exist!'


def challenge(req):
    req.sendall(b'This is a cloud storage service.\n' +
                b'You can list the files inside and also see their contents if your signatures are valid.\n')

    while True:
        req.sendall(b'\nOptions:\n1.List files\n2.Access a file\n')
        try:
            payload = json.loads(req.recv(4096))
            if payload['option'] == 'list':
                payload = json.dumps(
                    {'response': 'success', 'files': nfnames})
                req.sendall(payload.encode())
            elif payload['option'] == 'access':
                fname = payload['fname']
                r, s = payload['r'], payload['s']
                dt = ecc.verify(fname.encode(), r, s)
                if ('not exist' in dt) or ('not valid' in dt):
                    payload = json.dumps({'response': 'error', 'message': dt})
                else:
                    payload = json.dumps({'response': 'success', 'data': dt})
                req.sendall(payload.encode())
            else:
                payload = json.dumps(
                    {'response': 'error', 'message': 'Invalid option!'})
                req.sendall(payload.encode())
        except:
            payload = json.dumps(
                {'response': 'error', 'message': 'An error occured!'})
            req.sendall(payload.encode())


class incoming(socketserver.BaseRequestHandler):
    def handle(self):
        signal.alarm(30)
        req = self.request
        challenge(req)


class ReusableTCPServer(socketserver.ForkingMixIn, socketserver.TCPServer):
    pass


def main():
    init_storage()
    socketserver.TCPServer.allow_reuse_address = True
    server = ReusableTCPServer(("0.0.0.0", 1337), incoming)
    server.serve_forever()


if __name__ == "__main__":
    main()

Análisis del código fuente

El programa comienza llamando a init_storage. Después de eso, inicia un servidor de sockets por TCP que llamará a challenge:

class incoming(socketserver.BaseRequestHandler):
    def handle(self):
        signal.alarm(30)
        req = self.request
        challenge(req)


class ReusableTCPServer(socketserver.ForkingMixIn, socketserver.TCPServer):
    pass


def main():
    init_storage()
    socketserver.TCPServer.allow_reuse_address = True
    server = ReusableTCPServer(("0.0.0.0", 1337), incoming)
    server.serve_forever()

Hay una alarma de 30 segundos, lo que significa que el servidor cierra las conexiones después de este tiempo. Sin embargo, esto no será un problema.

Esto es init_storage:

fnames = [b'subject_kolhen', b'subject_stommb', b'subject_danbeer']
nfnames = []

# ...

ecc = ECDSA()


def init_storage():
    i = 0
    for fname in fnames[:-1]:
        data = ecc.sign(fname)
        r, s = data['r'], data['s']
        nonce = data['nonce']
        nfname = fname.decode() + '_' + r + '_' + s + '_' + nonce[(14 + i):-14]
        nfnames.append(nfname)
        i += 2

Vemos que usa una firma ECDSA en cada uno de los nombres de archivo de fnames, excepto que en el último (subject_danbeer). Cada firma se añade a nfnames, con información adicional (más sobre esto más adelante).

La función challenge muestra las opciones que podemos usar para interactuar con el servidor:

def challenge(req):
    req.sendall(b'This is a cloud storage service.\n' +
                b'You can list the files inside and also see their contents if your signatures are valid.\n')

    while True:
        req.sendall(b'\nOptions:\n1.List files\n2.Access a file\n')
        try:
            payload = json.loads(req.recv(4096))
            if payload['option'] == 'list':
                payload = json.dumps(
                    {'response': 'success', 'files': nfnames})
                req.sendall(payload.encode())
            elif payload['option'] == 'access':
                fname = payload['fname']
                r, s = payload['r'], payload['s']
                dt = ecc.verify(fname.encode(), r, s)
                if ('not exist' in dt) or ('not valid' in dt):
                    payload = json.dumps({'response': 'error', 'message': dt})
                else:
                    payload = json.dumps({'response': 'success', 'data': dt})
                req.sendall(payload.encode())
            else:
                payload = json.dumps(
                    {'response': 'error', 'message': 'Invalid option!'})
                req.sendall(payload.encode())
        except:
            payload = json.dumps(
                {'response': 'error', 'message': 'An error occured!'})
            req.sendall(payload.encode())

Bueno, no hay muchas cosas que hacer aquí. Usaremos list para tomar dichas firmas (nfnames). Luego, necesitamos usar access para leer cualquiera de los archivos, proporcionando su correspondiente firma.

Sabemos que lee un archivo porque el método verify de ECDSA utiliza retrieve_file:

class ECDSA:
    # ...

    def verify(self, fname, r, s):
        h = bytes_to_long(sha1(fname).digest())
        r = int(r, 16)
        s = int(s, 16)
        sig = Signature(r, s)

        if self.pubkey.verifies(h, sig):
            return retrieve_file(fname)
        else:
            return 'Signature is not valid\n'

# ...

def retrieve_file(fname):
    try:
        dt = open(fname, 'rb').read()
        return dt.hex()
    except:
        return 'The file does not exist!'

Por lo tanto, es posible que necesitemos leer subject_danbeer para obtener la flag. ¡Pero no tenemos su firma! Por lo tanto, debemos encontrar una manera de firmar este nombre de archivo para poder leerlo.

El servidor usa ECDSA con la librería ecdsa en una clase llamada ECDSA:

class ECDSA:
    def __init__(self):
        self.G = generator_256
        self.n = self.G.order()
        self.key = randint(1, self.n - 1)
        self.pubkey = Public_key(self.G, self.key * self.G)
        self.privkey = Private_key(self.pubkey, self.key)

    def sign(self, fname):
        h = sha1(fname).digest()
        nonce = randint(1, self.n - 1)
        sig = self.privkey.sign(bytes_to_long(h), nonce)
        return {"r": hex(sig.r)[2:], "s": hex(sig.s)[2:], "nonce": hex(nonce)[2:]}

    def verify(self, fname, r, s):
        h = bytes_to_long(sha1(fname).digest())
        r = int(r, 16)
        s = int(s, 16)
        sig = Signature(r, s)

        if self.pubkey.verifies(h, sig):
            return retrieve_file(fname)
        else:
            return 'Signature is not valid\n'


ecc = ECDSA()

Obsérvese que la variable ecc se inicializa cuando comienza el programa, no cuando llega una nueva conexión. Por lo tanto, no importa que el servidor cierre la conexión después de 30 segundos, porque simplemente podemos volver a conectarnos y se utilizará la misma variable ecc.

ECDSA

Básicamente, las firmas de ECDSA son un par de valores tal que:

Donde:

es el mensaje a firmar (fname)
es una función hash (SHA256)
es un nonce elegido al azar
es la clave privada
es un punto generador de la curva elíptiva P-256
es el orden de la curva elíptiva P-256

Solución

Esta vez, se nos proporciona información adicional sobre los nonces en init_storage:

def init_storage():
    i = 0
    for fname in fnames[:-1]:
        data = ecc.sign(fname)
        r, s = data['r'], data['s']
        nonce = data['nonce']
        nfname = fname.decode() + '_' + r + '_' + s + '_' + nonce[(14 + i):-14]
        nfnames.append(nfname)
        i += 2

Obsérvese que solo tenemos dos mensajes y firmas conocidos, así que escribamos las ecuaciones que tenemos (sea ):

Normalmente, en esta situación en la que tenemos información parcial sobre los nonces, debemos usar una técnica basada en retículos para encontrar una solución al sistema de ecuaciones anterior. Vemos que tenemos 2 ecuaciones con 3 incógnitas, por lo que no podemos resolver esto directamente.

Si leemos el código cuidadosamente, nos dan nonce[(14 + i):-14], con i += 2; y el nonce es un número de 256 bits en formato hexadecimal. Por lo tanto, podemos decir:

Donde tenemos el valor de y , y sabemos que y que .

Ideas fallidas

Al tratar con esta situación (conocida como nonces sesgados), necesitamos transformar las ecuaciones en un Hidden Number Problem (HNP).

El HNP generalmente se define como:

Donde ese el número oculto, está acotado, y los valores y son conocidos, dadas ecuaciones.

Entonces, podemos obtener una expresión de HNP de la siguiente manera:

Muy bien, ahora podemos añadir la información que conocemos sobre :

No resolveremos una versión exacta del HNP porque tenemos un a variable adicional. Sin embargo, todavía podemos usar una técnica basada en retículos para resolver esto.

Primero, debemos tratar el , para esto, podemos agregar múltiplos de :

He destacado las incógnitas en color rojo y he sustituido lo siguiente:

En este punto, podemos escribir las ecuaciones anteriores como producto de dos matrices:

Dado que el resultado es un vector nulo, podemos intentar buscar un vector corto que contenga la información que queremos. Por ejemplo:

La expresión anterior dice que el retículo formado por las columnas de la matriz contiene un vector , que es corto porque hay muchas coordenadas pequeñas (por ejemplo, , , ). Por lo tanto, podemos usar un algoritmo llamado LLL para reducir la base del retículo y ver si encuentra este vector corto.

Para mejorar la efectividad de LLL, podemos agregar un peso arbitrariamente grande a la columna con , porque queremos obligar a LLL a usarlo solo una vez. Por ejemplo, . Entonces esta será la matriz que será reducida por LLL:

Y el vector objetivo será .

Intenté implementar este enfoque y no fue exitoso. Modifiqué un poco la matriz, usando solo una ecuación, combinando ambas ecuaciones, aumentando los pesos… y nada. Así que di un paso atrás y probé algo más inteligente.

Idea correcta

Revisemos el sistema de ecuaciones que tenemos (con ):

¿Qué pasa si eliminamos la variable ? Este es un entero de 256 bits, por lo que es muy grande en comparación con y . Obtenemos esto:

Esta ecuación es mucho mejor, porque solo tenemos y . También necesitamos agregar una variable adicional debido al , pero no debería ser un problema:

Entonces, podemos usar la siguiente matriz de la base del retículo:

Y el vector objetivo será , ¡que es mucho más corto!

La solución prevista a este reto fue encontrar recursos que expliquen cómo lidiar con información parcialmente conocida sobre nonces en ECDSA. Uno de estos recursos es este paper, donde también emplean el truco de eliminar la variable para acortar el vector objetivo.

Implementación

En primer lugar, nos conectamos a la instancia remota y tomamos los parámetros necesarios para el ataque:

io = remote(host, port)

io.sendlineafter(
    b'Options:\n1.List files\n2.Access a file\n',
    json.dumps({'option': 'list'}).encode()
)

files = json.loads(io.recvline().decode())['files']

values = files[0].split('_')
fname = '_'.join(values[:2])
r_1 = int(values[2], 16)
s_1 = int(values[3], 16)
b_1 = int(values[4], 16)
h_1 = int(sha1(fname.encode()).hexdigest(), 16)

values = files[1].split('_')
fname = '_'.join(values[:2])
r_2 = int(values[2], 16)
s_2 = int(values[3], 16)
b_2 = int(values[4], 16)
h_2 = int(sha1(fname.encode()).hexdigest(), 16)

Luego, definimos la matriz de la base del retículo:

A_1 = 2 ** 200 * s_1
A_2 = 2 ** 192 * s_2
B_1 = 2 ** 56 * s_1 * b_1 - h_1
B_2 = 2 ** 56 * s_2 * b_2 - h_2

n = generator_256.order()

W = 2 ** 1024

M = Matrix(ZZ, [
    [
        r_2 * A_1 % n,
        r_1 * A_2 % n,
        r_2 * s_1 % n,
        r_1 * s_2 % n,
        n,
        (r_2 * B_1 - r_1 * B_2) % n
    ],
    [1, 0, 0, 0, 0, 0],
    [0, 1, 0, 0, 0, 0],
    [0, 0, 1, 0, 0, 0],
    [0, 0, 0, 1, 0, 0],
    [0, 0, 0, 0, 1, 0],
    [0, 0, 0, 0, 0, W],
]).transpose()

Ahora podemos aplicar LLL. Es importante reducir las entradas con % n, de lo contrario, serán demasiado grandes para LLL. Además, obsérvese que necesitamos escalar la matriz por para obligar a LLL a encontrar la solución deseada (y luego volver a escalarla):

M[:, 0] *= W
L = M.LLL()
L[:, 0] /= W

Si LLL tiene éxito, la última fila será nuestro vector objetivo, por lo que podemos afirmar que el primer elemento es y el último elemento es . Entonces, podemos tomar los valor intermedios , , y (usamos valor absoluto para evitar números negativos):

row = L[-1]
assert row[0] == 0 and row[-1] == W

a_1 = int(abs(row[1]))
a_2 = int(abs(row[2]))
c_1 = int(abs(row[3]))
c_2 = int(abs(row[4]))

Con estos, podemos reconstruir los nonces y , y encontrar fácilmente , simplemente resolviendo ecuaciones lineales:

k_1 = 2 ** 200 * a_1 + 2 ** 56 * b_1 + c_1
k_2 = 2 ** 192 * a_2 + 2 ** 56 * b_2 + c_2

x = (s_1 * k_1 - h_1) * pow(r_1, -1, n) % n
assert x == (s_2 * k_2 - h_2) * pow(r_2, -1, n) % n

Una vez que tenemos la clave privada, podemos firmar cualquier mensaje. Entonces, podemos firmar subject_danbeer y leer este archivo desde el servidor:

k = 1337
fname = 'subject_danbeer'
h = int(sha1(fname.encode()).hexdigest(), 16)
r = int((k * generator_256).x())
s = pow(k, -1, n) * (h + x * r) % n

io.sendlineafter(
    b'Options:\n1.List files\n2.Access a file\n',
    json.dumps({'option': 'access', 'fname': fname, 'r': hex(r), 's': hex(s)}).encode()
)

data = bytes.fromhex(json.loads(io.recvline().decode())['data']).decode()
io.success(f'{fname}:\n{data}')

Flag

Si ejecutamos el script contra la instancia remota, capturaremos la flag:

$ python3 solve.py 94.237.54.190:39685
[+] Opening connection to 94.237.54.190 on port 39685: Done
[+] subject_danbeer:
    Test subject - Danbeer
    
    DEBUG_MSG - Starting Mind...
    
        What a life this is...
        I lost my only child.
        My home got destroyed by the army.
        They took everything from me and now I'm trapped inside a cloud server.
        I don't even know where my real body is.
        I remember the day that they captured me.
        It was 2 weeks after I lost my precious Klaus.
        I was in the Inn of the city, trying to find more information about the
        android graveyard planet.
        3 men jumped on me!
        I won't give up, I shouted!
        On the day that the suns of the Nim cluster are aligned, I will be there.
        Draeger won't wi...
    
    DEBUG_MSG - Shutting Down...
    
    Notes:
        The subject seems to be rebellious and dangerous for android usage
        HTB{m@st3r1ng_LLL_1s_n0t_3@sy_TODO}
[*] Closed connection to 94.237.54.190 port 39685

El script completo se puede encontrar aquí: solve.py.