Digital Safety Annex

9 minutos de lectura

Se nos proporciona el código fuente en Python del servidor, separado en varios archivos:

server.py
_annex.py
_account.py
_dsa.py

Análisis del código fuente

El archivo principal es server.py. Veamos las opciones que ofrece:

from secret import FLAG, HTB_PASSWD
from _annex import Annex
import re

def show_menu():
    return input("""
Welcome to the Digital Safety Annex!
We will keep your data safe so you don't have to worry!

[0] Create Account 
[1] Store Secret
[2] Verify Secret
[3] Download Secret
[4] Developer Note
[5] Exit

[+] Option > """)

def input_number(prompt):
    n = input(prompt)
    return int(n) if n.isdigit() else None

def main():
    annex = Annex()
    annex.create_account("Admin", "5up3r_53cur3_P45sw0r6")
    annex.create_account("ElGamalSux", HTB_PASSWD)
    annex.sign("ElGamalSux", "DSA is a way better algorithm", HTB_PASSWD)
    annex.sign("Admin", "Testing signing feature", "5up3r_53cur3_P45sw0r6")
    annex.sign("ElGamalSux", "I doubt anyone could beat it", HTB_PASSWD)
    annex.sign("Admin", "I should display the user log and make sure its working", "5up3r_53cur3_P45sw0r6")
    annex.sign("ElGamalSux", "To prove it, I'm going to upload my most precious item! No one but me will be able to get it!", HTB_PASSWD)
    annex.sign("ElGamalSux", FLAG, HTB_PASSWD)

    account_username = ""

    while True:
        user_inp = show_menu()
        # ...

Obsérvese que la flag se almacena como un mensaje firmado por ElGamalSux.

Entre todas las opciones, la única que importa es la opción 4:

        elif user_inp == '4':
            print("\nWe are here to prove that DSA is waaayyyy better than El Gamal!\nWe also modified our signature algorithm to use the super secure SHA-256. No way you can bypass our authentication. If you must try, be sure to bring tissues for your tears of failure.\nI'll throw you a bone, these are public record anyway:\n")

            p, q, g = annex.dsa.get_public_params()

            print(f'{p = }')
            print(f'{q = }')
            print(f'{g = }')

            inp = input("[+] Test user log (y/n): ").lower()
            if inp == 'y':
                if annex.users['Admin'].login():
                    print(f'\n{annex.user_log}')

El programa nos dará los parámetros para la implementación del DSA y nos pedirá iniciar sesión como Admin:

class Account:
    def __init__(self, username="default-user", passwd=""):
        self.username = username
        self.password = sha256(passwd.encode()).digest()
        self.k_max = int(len(self.username) ** 6)
        if self.k_max < 65536:
            self.k_max += 1000000
        self.stored_msgs = 0

    def login(self, pw=None):
        if not pw:
            pw = input("Enter your password : ")
        return sha256(pw.encode()).hexdigest() == self.password.hex()

Por suerte, la contraseña está hard-coded en server.py (5up3r_53cur3_P45sw0r6). Mirando más de cerca la clase Account de _account.py, vemos un atributo k_max que es igual a la longitud del nombre de usuario elevado a . Por lo tanto, el atributo k_max para Admin sería . Dado que este valor es inferior a , se suma para aumentarlo un poco. Entonces el valor final de k_max es .

Volviendo a la opción 4, una vez que ponemos la contraseña correcta, se nos muestra user_log, que mostrará todas las firmas almacenadas en annex. La clase annex ( _annex.py) guarda cada firma en user_log cada vez que se crea una nueva firma, por lo que es una especie de almacén global:

class Annex:
    def __init__(self):
        self.user_log = []
        self.users = {}
        self.vault = {}
        self.dsa = DSA()

    def create_account(self, username="", password=""):
        # ...

    def log_info(self, account, msg, h, sig):
        _id = account.stored_msgs
        if account.username not in self.vault:
            self.vault[account.username] = []
        
        self.vault[account.username].append((h, msg, (str(sig[0]), str(sig[1]))))
        self.user_log.append((sig, h))
        account.stored_msgs += 1

    def sign(self, username, message, password=""):
        account = self.users[username]
        
        if not account.login(password):
            print("[!] Invalid Password!\n")
            return (0, 0)

        msg = message.encode()
        h = sha256(msg).hexdigest()
        
        r, s = self.dsa.sign(h, account.k_max)
        
        self.log_info(account, msg, h, (r, s))
        
        return (r, s)

    # ...

Como resultado, si usamos la opción 4, obtendremos los parámetros del DSA y todas las firmas, incluida la última, que es la flag firmada.

Digital Signature Algorithm

El programa está utilizando DSA como algoritmo de firma. Este es un algoritmo asimétrico basado en el cifrado ElGamal donde los parámetros públicos son enteros , La clave privada es un entero y la clave pública es .

La seguridad de este algoritmo recae en la dificultad de resolver el problema de logaritmo discreto. Es decir, dado , hallar . Para hacer esto difícil, el número primo se elige para que . Como resultado, cualquier tal que será un generador de un subgrupo de orden en $\mathbb{Z}_q^ $ó$ q $ó$ \mathrm{ord}(\mathbb{Z}_q^) = \phi(q) = q - 1 = 2pp$ es un número primo, el subgrupo no contiene más subgrupos más pequeños (que no sean los triviales), por lo que es difícil resolver el DLP.

De todas formas, una firma que usa DSA se compone de dos valores :

Donde es el mensaje para firmar, es una función hash como SHA256, y es un nonce aleatorio que debe ser diferente en cada firma y elegido al azar entre y .

Echemos un vistazo a la implementación (_dsa.py):

class DSA:
    def __init__(self, key_size=2048):
        key = PrimeGenerator.generate(key_size)
        self.p, self.q = key.p, key.q
        self.x, self.y, self.g = self.generate_keys()
        self.k_min = 65500

    def get_public_params(self):
        return (self.p, self.q, self.g)

    def generate_keys(self):
        h = random.randint(2, self.p-2)
        g = pow(h, (self.p-1)//self.q, self.p)
        x = random.randint(1, self.q-1)
        y = pow(g, x, self.p)
        return x, y, g

    def sign(self, h, k_max):
        k = random.randint(self.k_min, k_max)
        r = pow(self.g, k, self.p) % self.q
        s = (pow(k, -1, self.q) * (int(h, 16) + self.x * r)) % self.q
        return (r, s)

    def verify(self, h, signature):
        r, s = signature
        r = int(r)
        s = int(s)
        w = pow(s, -1, self.q)
        u1 = (int(h, 16) * w) % self.q
        u2 = (r * w) % self.q
        v = ((pow(self.g, u1, self.p) * pow(self.y, u2, self.p)) % self.p) % self.q
        return r == v

Como se puede ver, el valor del nonce es aleatorio, pero el rango es muy pequeño. Por ejemplo, para Admin, , lo que significa que podemos simplemente tomar una firma e intentar todos los posibles valores de para calcular hasta que encontremos el mismo valor de .

Solución

Entonces, la solución es simplemente fuerza bruta. Estamos interesados en el último mensaje guardado, que pertenece a ElGamalSux. Este nombre tiene caracteres, por lo que k_max es (es más grande que , por lo que ya no se actualiza). Como resultado, necesitaremos probar valores de .

Nuevamente, la idea es tomar el valor de de la firma de destino y calcular para cada en el intervalo hasta que encontremos una coincidencia con el valor objetivo. En ese momento, podemos tomar el valor de y encontrar :

Una vez que encontramos la clave privada , podemos usar la opción 3 para recibir el mensaje firmado en texto claro:

        elif user_inp == '3':
            uname = input("\nPlease enter the username that stored the message: ")
            if not uname in annex.vault:
                print("\n[!] Sorry, need valid existing username to download secret!")
                continue

            req_id = input("\nPlease enter the message's request id: ")
            if not req_id.isdigit() or not (0 <= int(req_id) < len(annex.vault[uname])):
                print("\n[!] Sorry, need valid request id to download secret!")
                continue
            
            req_id = int(req_id)
            if uname == account_username:
                # ...
            else:
                k = input_number("\nPlease enter the message's nonce value: ")
                if not k:
                    print("\n[!] Sorry, need a valid nonce to download secret!")
                    continue
            
                x = input_number("\n[+] Please enter the private key: ")
                if not x:
                    print("\n[!] Sorry, need a valid private key to download secret!")
                    continue

                annex.download(x, k, req_id, uname)

Al final, el programa llamará a annex.download, que verificará que la clave privada proporcionada y el nonce dan como resultado la misma firma que la almacenada:

    def download(self, priv, nonce, req_id, username):
        h, m, sig = self.vault[username][req_id]
        
        p, q, g = self.dsa.get_public_params()
 
        rp = pow(g, nonce, p) % q
        sp = (pow(nonce, -1, q) * (int(h, 16) + priv * rp)) % q

        new_sig = (str(rp), str(sp))

        if new_sig == sig:
            print(f"[+] Here is your super secret message: {m}")
        else:
            print(f"[!] Invalid private key or nonce value! This attempt has been recorded!")

En este punto, obtendremos la flag.

Implementación

Elegí Go para programar la solución, usando mi módulo gopwntools.

En primer lugar, nos conectamos al servidor y usamos la opción 4 para obtener los parámetros DSA, iniciar sesión como Admin usando la contraseña hard-coded y tomar la última firma y el hash :

type bi = big.Int

func main() {
        io := getProcess()
        defer io.Close()

        io.SendLineAfter([]byte("[+] Option > "), []byte{'4'})
        io.RecvUntil([]byte("p = "))
        p, _ := new(bi).SetString(strings.TrimSpace(io.RecvLineS()), 10)
        io.RecvUntil([]byte("q = "))
        q, _ := new(bi).SetString(strings.TrimSpace(io.RecvLineS()), 10)
        io.RecvUntil([]byte("g = "))
        g, _ := new(bi).SetString(strings.TrimSpace(io.RecvLineS()), 10)
        io.SendLineAfter([]byte("[+] Test user log (y/n): "), []byte{'y'})
        io.SendLineAfter([]byte("Enter your password : "), []byte("5up3r_53cur3_P45sw0r6"))
        io.RecvUntil([]byte{'['})

        for range 6 {
                io.RecvUntil([]byte("(("))
        }

        r, _ := new(bi).SetString(io.RecvUntilS([]byte(", "), true), 10)
        s, _ := new(bi).SetString(io.RecvUntilS([]byte("), '"), true), 10)
        h, _ := new(bi).SetString(io.RecvUntilS([]byte{'\''}, true), 16)

Obsérvese que necesitamos usar big.int en Go porque estamos trabajando con valores enteros grandes, y también nótense las codificaciones numéricas, ya que algunos de los números están puestos en decimal, mientras que otros se muestran en hexadecimal.

Después de eso, comenzamos el ataque de fuerza bruta:

        k := int64(65500)

        for gkp := new(bi).Exp(g, new(bi).SetInt64(k), p); r.Cmp(new(bi).Mod(gkp, q)) != 0; k++ {
                gkp.Mod(gkp.Mul(gkp, g), p)
        }

Como se puede ver, no estamos calculando explícitamente . En cambio, calculamos y en cada iteración del bucle for, lo multiplicamos por . Esto es más eficiente que calcular la expresión explícita en cada iteración.

El bucle for parará cuando r.Cmp(new(bi).Mod(gkp, q)) devuelva true. Después de eso, calculamos x:

        x := new(bi).Mod(new(bi).Mul(new(bi).Sub(new(bi).Mul(s, new(bi).SetInt64(k)), h), new(bi).ModInverse(r, q)), q)

Lo sé, la expresión no es fácil de leer, pero así es como funciona big.Int en Go, no me culpéis.

Finalmente, enviamos el nonce k y la clave privada x en la opción 3 para obtener la flag:

        io.SendLineAfter([]byte("[+] Option > "), []byte{'3'})
        io.SendLineAfter([]byte("Please enter the username that stored the message: "), []byte("ElGamalSux"))
        io.SendLineAfter([]byte("Please enter the message's request id: "), []byte{'3'})
        io.SendLineAfter([]byte("Please enter the message's nonce value: "), fmt.Appendf(nil, "%d", k))
        io.SendLineAfter([]byte("[+] Please enter the private key: "), []byte(x.String()))

        io.RecvUntil([]byte("[+] Here is your super secret message: "))
        io.RecvUntil([]byte("b'"))
        pwn.Success(io.RecvUntilS([]byte{'\''}, true))
}

Flag

Con todo esto, podemos ejecutar el programa contra la instancia remota y obtener la flag en menos de 5 segundos:

$ go run solve.go 94.237.58.95:42004
[+] Opening connection to 94.237.58.95 on port 42004: Done
[+] HTB{1_Gu3ss_d54_15_N07_4s_s3CuR3_A5_1_7h0u647}
[*] Closed connection to 94.237.58.95 port 42004

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