Digital Safety Annex

9 minutes to read

We are given the Python source code of the server, separated into several files:

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

Source code analysis

The main file is server.py. Let’s show the options that it offers:

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()
        # ...

Notice that the flag is stored as a message signed by ElGamalSux.

Among all the options, the only one that matters is option 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}')

The program will give us parameters for the DSA implementation and will require us to login as 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()

Luckily, the password is hard-coded in server.py (5up3r_53cur3_P45sw0r6). Taking a closer look at the Account class from _account.py, we see a k_max attribute that equals the length of the username raised to the power of . Therefore, the attribute k_max for Admin would be . Since this value is less than , they add to increase it a bit. So the final k_max attribute is .

Going back to option 4, once we enter the correct password, we are shown user_log, which will show all signatures stored in annex. The class Annex (_annex.py) saves each signature in user_log every time a new signature is created, so it is kind of a global vault:

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)

    # ...

As a result, if we use option 4 we will get the DSA parameters and all signatures, including the last one, which is the signed flag.

Digital Signature Algorithm

The program is using DSA as a signature algorithm. This is an asymmetric algorithm based on ElGamal encryption where the public parameters are integers , the private key is an integer and the public key is .

The security of this algorithm relies on the difficulty of solving the discrete logarithm problem. That is, given , find . To make this hard, the prime number is chosen so that . As a result, any such that will be a generator of an -order subgroup of $\mathbb{Z}_q^q\mathrm{ord}(\mathbb{Z}_q^) = \phi(q) = q - 1 = 2pp$ is a prime number, the subgroup doesn’t contain any smaller subgroups (other than the trivial one), so solving the DLP is hard.

Anyways, a signature using DSA is composed of two values :

Where is the message to sign, is hash function such as SHA256, and is a random nonce that must be different in each signature and chosen at random between and .

Let’s have a look at the implementation (_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

As can be seen, the value of the nonce is random, but the range is so short! For instance, for Admin, , which means we can simply take a signature and try all possible values of to compute until we find the same value.

Solution

So, the solution is simply brute force. We are interested in the last message saved, which belongs to ElGamalSux. This name has characters, so k_max is (it is bigger than , so it is not updated anymore). As a result, we will need to try values of .

Again, the idea is to take the value of the target signature and compute for each in the interval until we find a match with the target . At that point, we can take the value and find :

Once we find the private key , we can use option 3 to get the plaintext message:

        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)

In the end, the program will call annex.download, which will verify that the provided private key and nonce result in the same signature as the one stored:

    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!")

At this point, we will get the flag.

Implementation

I chose Go to program the solution, using gopwntools module.

First of all, we connect to the server and use option 4 to get the DSA parameters, login as Admin using the hard-coded password and take the last signature and the message 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)

Notice that we need to use big.Int in Go because we are working with big integer values, and also mind the number encodings, since some of the numbers are printed out in decimal whereas others are shown in hexadecimal.

After that, we begin the brute force attack:

        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)
        }

As can be seen, we are not explicitly computing . Instead, we compute and on each iteration of the for loop, we multiply it by . This is more efficient than computing the explicit expression on each iteration.

The for loop will stop when r.Cmp(new(bi).Mod(gkp, q)) returns true. After that, we compute 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)

I know, the expression is not easy to read, but that’s how Go’s big.Int works, don’t blame me.

Finally, we send the nonce k and the private key x in option 3 in order to get the 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

With all this, we can run the program against the remote instance and get the flag in less than 5 seconds:

$ 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

The full script can be found in here: solve.go.