[Layer7] Web CTF Write-up

Rednose U

드림핵에 있는 Rednose Airline과 같은 문제이다. 코드를 확인해보자.

`from flask import *
import jwt
import re
import string
import random
from datetime import datetime, timedelta
import subprocess
import os

request_count = 0
timeout_until = None
app = Flask(__name__)

RATE_LIMIT_REQUESTS = 10
RATE_LIMIT_TIMEOUT = timedelta(minutes=5)

JWTKey = ''.join(random.choices(string.ascii_letters, k=50))
app.config['JWTKey'] = JWTKey

users = [
    {"id": "admin", "pw": "{{REDACTED}}"},
    {"id": "guest", "pw": "guest"},
]

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

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'GET':
        if request.cookies.get('auth'):
            return render_template_string('<script>alert("You are already logged in.");history.back()</script>')
        return render_template('login.html')
    elif request.method == 'POST':
        form_id = request.form['id']
        form_pw = request.form['pw']
        print(form_id, form_pw)
        if len(form_id) > 10:
            return "Too Long"
        elif len(form_id) < 3:
            return "Too Short"
        elif not re.search(r"^[A-Za-z{}]*$", form_id):
            return "Blacklist Word"

        user = next((user for user in users if user['id'] == form_id), None)
        if user and user['pw'] == form_pw:
            payload = {
                'id': form_id,
                'isAdmin': (form_id == 'admin'),
            }
            encode_jwt = jwt.encode(payload, JWTKey, algorithm='HS256')
            resp = make_response(render_template_string('Login Success!'))
            resp.set_cookie('auth', encode_jwt)
            return resp
        else:
            return render_template_string('{form_id} is not registered, or the password is incorrect.'.format(form_id=form_id))
        
@app.route('/logout')
def logout():
    resp = make_response(render_template_string('<script>alert("Bye.");history.back()</script>'))
    resp.set_cookie('auth', '', expires=0)
    return resp
        
@app.route('/dashboard', methods=['GET', 'POST'])
def admin():
    if request.cookies.get('auth'):
        try:
            decode_jwt = jwt.decode(request.cookies.get('auth'), JWTKey, algorithms=['HS256'])
            if decode_jwt['id'] == 'admin' and decode_jwt['isAdmin'] == True:
                return render_template('admin.html')
            else:
                return render_template_string('You are not admin.')
        except:
            return send_file("./cat/fail.jpg", mimetype='image/jpg') # BAD JWT
    else:
        return render_template_string('<script>alert("You are not logged in.");history.back()</script>')

@app.route('/api/metar')
def metar():
    global request_count
    global timeout_until
    current_time = datetime.now()
    if timeout_until and current_time < timeout_until:
        remaining_time = timeout_until - current_time
        return "Timeout! {}".format(remaining_time), 429
    if request.cookies.get('auth'):
        try:
            decode_jwt = jwt.decode(request.cookies.get('auth'), JWTKey, algorithms=['HS256'])
            if decode_jwt['id'] == 'admin' and decode_jwt['isAdmin'] == True:
                request_count += 1
                if request_count >= RATE_LIMIT_REQUESTS:
                    timeout_until = current_time + RATE_LIMIT_TIMEOUT
                    request_count = 0 
                airport = request.args.get('airport')
                result = subprocess.run(f'curl {airport}', shell = True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)                    
                return result.stdout
            else:
                return render_template_string('You are not admin.')
        except Exception as e:
            print(e)
            return send_file("./cat/fail.jpg", mimetype='image/jpg') # BAD JWT
    else:
        return render_template_string('<script>alert("You are not logged in.");history.back()</script>')
    
app.run(host='0.0.0.0', port=13000, threaded=False)

 

여기서 우린 admin 권한을 탈취해야한다. 그러나 SQL Injection이 불가하므로 이 방법 말고 다른 방법을 사용해야 한다. 바로 JWT 키를 탈취하는 것이다. 이 JWT 키는 app.config에 저장되어 있으므로 SSTI 취약점을 사용하여 이를 탈취할 수 있다.

 

login 엔드포인트 코드를 확인해보면 로그인에 실패했을 때 다음 코드를 실행한다.

return render_template_string('{form_id} is not registered, or the password is incorrect.'.format(form_id=form_id))

 

여기서 render_template_string 함수에 있는 SSTI 취약점을 사용하여 템플릿 구문을 삽입할 수 있다.

 

로그인 페이지에서 {{config}}를 넣었을 때를 확인해보면 여러 설정 값들이 보이는데 이 값들 중 JWTKey를 찾을 수 있다.

이렇게 얻은 JWT Key를 사용하여 새로운 JWT 토큰을 얻을 수 있다.

 

 

이 사이트를 사용하여 값을 조작한 후, 키를 이용해 새로운 토큰을 만들 수 있다.

 

 

이 과정을 거친 토큰 값을 쿠키에다 넣어주면 서버가 쿠키를 확인한 후, 관리자로 로그인을 시켜준다,

 

이제 중요한 부분은 아래 코드이다.

@app.route('/api/metar')
def metar():
    global request_count
    global timeout_until
    current_time = datetime.now()
    if timeout_until and current_time < timeout_until:
        remaining_time = timeout_until - current_time
        return "Timeout! {}".format(remaining_time), 429
    if request.cookies.get('auth'):
        try:
            decode_jwt = jwt.decode(request.cookies.get('auth'), JWTKey, algorithms=['HS256'])
            if decode_jwt['id'] == 'admin' and decode_jwt['isAdmin'] == True:
                request_count += 1
                if request_count >= RATE_LIMIT_REQUESTS:
                    timeout_until = current_time + RATE_LIMIT_TIMEOUT
                    request_count = 0 
                airport = request.args.get('airport')
                result = subprocess.run(f'curl {airport}', shell = True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)                    
                return result.stdout
            else:
                return render_template_string('You are not admin.')
        except Exception as e:
            print(e)
            return send_file("./cat/fail.jpg", mimetype='image/jpg') # BAD JWT
    else:
        return render_template_string('<script>alert("You are not logged in.");history.back()</script>')

 

위 코드는 어드민이 확인되었을 때, airport 쿼리 값을 받아와서 curl 명령어를 실행하는 코드이다. 이 코드에는 subprocess 함수를 이용한 command injection 취약점이 존재한다. 이 취약점을 이용하여 명령어를 실행해 플래그 값을 얻어오면 된다.

 

 

ls를 사용하여 파일 구조를 확인하면 flag_qaiu.txt 파일이 있는 데 이게 플래그 파일이다.

 

 

cat 명령어를 실행시켜서 플래그 값을 얻어올 수 있다.

 

 

W2

from flask import Flask, request, render_template, redirect, url_for, session
import sqlite3
import hashlib
import re

app = Flask(__name__)
app.secret_key = 'super_secret_key_for_ctf'

DATABASE = 'users.db'

def init_db():
    conn = sqlite3.connect(DATABASE)
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            password TEXT NOT NULL,
            email TEXT,
            role TEXT DEFAULT 'user',
            secret TEXT
        )
    ''')
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS flags (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            flag_name TEXT,
            flag_value TEXT
        )
    ''')
    
    admin_password = hashlib.md5('{REDACTED}'.encode()).hexdigest() #no for user
    user_password = hashlib.md5('password123'.encode()).hexdigest()
    print(user_password)
    
    cursor.execute('INSERT OR REPLACE INTO users (id, username, password, email, role, secret) VALUES (?, ?, ?, ?, ?, ?)',
                   (1, 'admin', admin_password, 'admin@ctf.local', 'admin', '{REDACTED}'))
    
    cursor.execute('INSERT OR REPLACE INTO users (id, username, password, email, role, secret) VALUES (?, ?, ?, ?, ?, ?)',
                   (2, 'guest', user_password, 'guest@ctf.local', 'user', 'no_secret_here'))
    
    cursor.execute('INSERT OR REPLACE INTO users (id, username, password, email, role, secret) VALUES (?, ?, ?, ?, ?, ?)',
                   (3, 'test_user', hashlib.md5('test123'.encode()).hexdigest(), 'test@ctf.local', 'user', 'just_a_test_account'))
    
    cursor.execute('INSERT OR REPLACE INTO flags (id, flag_name, flag_value) VALUES (?, ?, ?)',
                   (1, 'main_flag', 'Layer7{FAKEFLAG}'))
    
    conn.commit()
    conn.close()

def get_db_connection():
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn

def simple_filter(query):
    blacklist = ['select', 'drop', 'delete', 'insert', 'update', 'create', 'alter']
    for word in blacklist:
        if word in query:
            return True
    return False

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

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        
        if simple_filter(username) or simple_filter(password):
            return render_template('login.html', error='Invalid input detected!')
        
        password_hash = hashlib.md5(password.encode()).hexdigest()
        
        conn = get_db_connection()
        
        
        try:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM users WHERE username = ? AND password = ?", (username, password_hash))
            user = cursor.fetchone()
            
            if user:
                session['user_id'] = user['id']
                session['username'] = user['username']
                session['role'] = user['role']
                return redirect(url_for('profile'))
            else:
                return render_template('login.html', error='Invalid credentials!')
        except Exception as e:
            return render_template('login.html', error=f'Database error: {str(e)}')
        finally:
            conn.close()
    
    return render_template('login.html')

@app.route('/profile')
def profile():
    if 'user_id' not in session:
        return redirect(url_for('login'))
    
    user_id = request.args.get('id', session['user_id'])
    
    if simple_filter(str(user_id)):
        return render_template('profile.html', error='Invalid input detected!')
    
    conn = get_db_connection()
    
    query = f"SELECT username, email, role, secret FROM users WHERE id = {user_id}"
    
    try:
        cursor = conn.cursor()
        cursor.execute(query)
        user = cursor.fetchone()
        
        if user:
            return render_template('profile.html', user=user)
        else:
            return render_template('profile.html', error='User not found!')
    except Exception as e:
        return render_template('profile.html', error=f'Database error: {str(e)}')
    finally:
        conn.close()

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

if __name__ == '__main__':
    init_db()
    app.run(host='0.0.0.0', port=5000, debug=True)

 

이 문제에서 우린 데이터베이스 flags 테이블에 있는 플래그 값을 얻어와야 한다. 위 코드에서는 로그인 엔드포인트에 SQL Injection은 할 수 없다. 그러나 아무 계정으로 로그인하고 profile 엔드포인트로 들어가면 user_id 쿼리 값을 사용하여 SQL Injection을 할 수 있다는 것을 알 수 있다. 

 

profile 엔드포인트로 가기 위해 ID와 PW가 나와 있는 test_user로 로그인한다. 로그인이 된 후, profile로 이동하면 user_id 값 안에 아래와 같은 쿼리를 넣어주면 플래그 값을 얻어올 수 있다.

 

1 UNION SELECT id, flag_name, flag_value, null FROM flags --

 

여기서 SELECT와 같은 쿼리 명령어는 WAF로 막혀 있다. 그러나 이 WAF에서는 소문자만 막기 때문에 대문자로 작성하여 우회할 수 있다. 

다른 테이블에 있는 값을 가져오기 위해 UNION SELECT를 사용한다. 그러나 여기서 중요한 점은 UNION SELECT에서는 전 SELECT 문과 컬럼 수를 맞혀야 하기 때문에 마지막 컬럼을 null로 하여 개수를 맞춘다. 이 쿼리를 실행하면 플래그가 나올 것이다.

 

 

 

SRS

이 문제에서는 파일이 app.py와 flag_server.py 두 가지로 나뉘어져 있다. app.py 파일의 코드를 확인해보자.

from flask import Flask, request, render_template, redirect, url_for, session
import requests
import re
from urllib.parse import urlparse
import time

app = Flask(__name__)
app.secret_key = 'ssrf_challenge_secret_key'

def url_filter(url):
    blocked_domains = ['localhost', '127.0.0.1', '0.0.0.0']
    blocked_schemes = ['file', 'ftp']
    
    try:
        parsed = urlparse(url)
        
        if parsed.scheme in blocked_schemes:
            return False
            
        if any(domain in parsed.netloc.lower() for domain in blocked_domains):
            return False
            
        if parsed.netloc.startswith('192.168.') or parsed.netloc.startswith('10.'):
            return False
            
        return True
    except:
        return False

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

@app.route('/check_url', methods=['GET', 'POST'])
def check_url():
    if request.method == 'POST':
        url = request.form.get('url', '').strip()
        
        if not url:
            return render_template('check_url.html', error='URL is required')
        
        if not url_filter(url):
            return render_template('check_url.html', error='Blocked URL detected')
        
        try:
            response = requests.get(url, timeout=5, allow_redirects=False)
            result = {
                'status_code': response.status_code,
                'headers': dict(response.headers),
                'content': response.text[:1000],
                'url': url
            }
            return render_template('check_url.html', result=result)
        except requests.exceptions.RequestException as e:
            return render_template('check_url.html', error=f'Request failed: {str(e)}')
    
    return render_template('check_url.html')

@app.route('/admin')
def admin():
    if session.get('role') != 'admin':
        return 'Access denied', 403
    return render_template('admin.html')

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        
        if username == 'admin' and password == 'REDACTED':
            session['role'] = 'admin'
            return redirect(url_for('admin'))
        else:
            return render_template('login.html', error='Invalid credentials')
    
    return render_template('login.html')

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

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

 

여기서 우리가 중요하게 확인해야할 부분은 check_url 엔드포인트이다. 여기에서는 requests 라이브러리를 사용하여 요청을 보내는 기능이 있다. 이걸로 우린 SSRF 공격을 할 수 있다. 

 

한 번 flag_server.py 코드를 확인해보자.

from flask import Flask, jsonify
import os

app = Flask(__name__)

@app.route('/')
def index():
    return 'Internal Flag Server'

@app.route('/flag')
def get_flag():
    return jsonify({
        'flag': 'Layer7{ssrf_easy}',
    })

@app.route('/health')
def health():
    return 'OK'

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8081, debug=False)

 

포트는 8081로 열려 있고 /flag 엔드포인트로 flag를 출력한다는 것을 알 수 있다. 결론적으로 우리는 SSRF 취약점을 사용하여 내부적으로 8081 포트로 열려 있는 /flag 엔드포인트에 진입하여 플래그를 획득해야 한다.

 

그러나 SSRF 공격을 하기 이전에 url_filter라는 함수를 우회해야한다. url_filter는 localhost와 127.0.0.1 등 루프백 주소를 막고 있다. 여기서 루프백 주소를 우회하는 방법은 여러가지가 있는데 나는 127.0.0.1을 10진수로 나타낸 루프백 주소를 사용하여 우회했다.

 

한번 10진수로 나타낸 루프백 주소와 flag 엔드포인트를 합쳐 check_url 페이지에 입력해보자.

http://2130706433:8081/flag

 

 

 

T야?

from flask import Flask, request, render_template, session, redirect, url_for
import sqlite3
import hashlib
import time
import os

app = Flask(__name__)
app.secret_key = '***REDACTED***'

def init_db():
    conn = sqlite3.connect('wargame.db')
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            password TEXT NOT NULL,
            role TEXT DEFAULT 'user'
        )
    ''')
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS flags (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            flag_value TEXT NOT NULL
        )
    ''')
    
    admin_password = hashlib.md5('***REDACTED***'.encode()).hexdigest()
    user_password = hashlib.md5('***REDACTED***'.encode()).hexdigest()
    
    cursor.execute('INSERT OR REPLACE INTO users (id, username, password, role) VALUES (1, "admin", ?, "admin")', (admin_password,))
    cursor.execute('INSERT OR REPLACE INTO users (id, username, password, role) VALUES (2, "user", ?, "user")', (user_password,))
    cursor.execute('INSERT OR REPLACE INTO flags (id, flag_value) VALUES (1, "***REDACTED***")')
    
    conn.commit()
    conn.close()

@app.route('/')
def index():
    if 'username' in session:
        return render_template('dashboard.html', username=session['username'])
    return redirect(url_for('login'))

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        
        password_hash = hashlib.md5(password.encode()).hexdigest()
        
        conn = sqlite3.connect('wargame.db')
        cursor = conn.cursor()
        
        query = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password_hash}'"
        
        try:
            cursor.execute(query)
            result = cursor.fetchone()
            
            if result:
                session['username'] = result[1]
                session['role'] = result[3]
                conn.close()
                return redirect(url_for('index'))
            else:
                conn.close()
                time.sleep(0.5)
                return render_template('login.html', message='Invalid credentials')
                
        except Exception as e:
            conn.close()
            time.sleep(0.5)
            return render_template('login.html', message='Invalid credentials')
    
    return render_template('login.html')

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

if __name__ == '__main__':
    init_db()
    app.run(host='0.0.0.0', port=5000, debug=True)

 

이 문제도 SQL Injection을 사용하여 flags 테이블에 있는 플래그 값을 가져오면 된다. 이 문제는 W2에서 사용한 페이로드를 컬럼에 맞게 수정을 해줘서 username에 입력해주면 플래그를 간단히 얻을 수 있다.

 

' UNION SELECT id, flag_value, null, null FROM flags --