[Dreamhack] Level 4: PATCH-1

1. 문제

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

PATCH-1

 

2. 해결 과정 

(1) 문제 페이지 접속

Usage 페이지에 문제에 대한 자세한 설명이 적혀있다.

주어진 코드 (문제 파일)를 수정하여 취약점을 패치한 뒤 제출하면 플래그를 얻을 수 있다. 

 

이때, 수정 가능한 코드는 app,py 파일 하나이다. 

 

(2) 코드 분석 

전체 코드 

#!/usr/bin/python3
from flask import Flask, request, render_template_string, g, session, jsonify
import sqlite3
import os, hashlib

app = Flask(__name__)
app.secret_key = "Th1s_1s_V3ry_secret_key"

def get_db():
  db = getattr(g, '_database', None)
  if db is None:
    db = g._database = sqlite3.connect(os.environ['DATABASE'])
  db.row_factory = sqlite3.Row
  return db

def query_db(query, args=(), one=False):
  cur = get_db().execute(query, args)
  rv = cur.fetchall()
  cur.close()
  return (rv[0] if rv else None) if one else rv

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

@app.route('/')
def index():
  return "api-server"

@app.route('/api/me')
def me():
  if session.get('uid'):
    return jsonify(userid=session['uid'])
  return jsonify(userid=None)

@app.route('/api/login', methods=['POST'])
def login():
  userid = request.form.get('userid', '')
  password = request.form.get('password', '')
  if userid and password:
    ret = query_db(f"SELECT * FROM users where userid='{userid}' and password='{hashlib.sha256(password.encode()).hexdigest()}'" , one=True)
    if ret:
      session['uid'] = ret[0]
      return jsonify(result="success", userid=ret[0])
  return jsonify(result="fail")

@app.route('/api/logout')
def logout():
  session.pop('uid', None)
  return jsonify(result="success")

@app.route('/api/join', methods=['POST'])
def join():
  userid = request.form.get('userid', '')
  password = request.form.get('password', '')
  if userid and password:
    conn = get_db()
    cur = conn.cursor()
    cur.execute("Insert into users values(?, ?);", (userid, hashlib.sha256(password.encode()).hexdigest()))
    conn.commit()
    return jsonify(result="success")
  return jsonify(result="error")

@app.route('/api/memo/add', methods=['PUT'])
def memoAdd():
  if not session.get('uid'):
    return jsonify(result="no login")

  userid = session.get('uid')
  title = request.form.get('title')
  contents = request.form.get('contents')

  if title and contents:
    conn = get_db()
    cur = conn.cursor()
    ret = cur.execute("Insert into memo(userid, title, contents) values(?, ?, ?);", (userid, title, contents))
    conn.commit()
    return jsonify(result="success", memoidx=ret.lastrowid)
  return jsonify(result="error")

@app.route('/api/memo/<idx>', methods=['GET'])
def memoView(idx):
  mode = request.args.get('mode', 'json')
  ret = query_db("SELECT * FROM memo where idx=" + idx)[0]
  if ret:
    userid = ret['userid']
    title = ret['title']
    contents = ret['contents']
    if mode == 'html':
      template = ''' Written by {userid}<h3>{title}</h3>
      <pre>{contents}</pre>
      '''.format(title=title, userid=userid, contents=contents)
      return render_template_string(template)
    else:
      return jsonify(result="success",
        userid=userid,
        title=title,
        contents=contents)
  return jsonify(result="error")

@app.route('/api/memo/<int:idx>', methods=['PUT'])
def memoUpdate(idx):
  if not session.get('uid'):
    return jsonify(result="no login")

  ret = query_db('SELECT * FROM memo where idx=?', [idx,])[0]
  userid = session.get('uid')
  title = request.form.get('title')
  contents = request.form.get('contents')

  if ret and title and contents:
    conn = get_db()
    cur = conn.cursor()
    updateRet = cur.execute("UPDATE memo SET title=?, contents=? WHERE idx=?",(title, contents, idx))
    conn.commit()
    if updateRet:
      return jsonify(result="success")
  return jsonify(result="error")

먼저, DB 관련 유틸리티들이 실행된다. 

이때 secret key값이 하드코딩 되어있는데, 이 부분도 패치 대상일 것이다. 

#!/usr/bin/python3
from flask import Flask, request, render_template_string, g, session, jsonify
import sqlite3
import os, hashlib

app = Flask(__name__)
app.secret_key = "Th1s_1s_V3ry_secret_key"

def get_db():
  db = getattr(g, '_database', None)
  if db is None:
    db = g._database = sqlite3.connect(os.environ['DATABASE'])
  db.row_factory = sqlite3.Row
  return db

def query_db(query, args=(), one=False):
  cur = get_db().execute(query, args)
  rv = cur.fetchall()
  cur.close()
  return (rv[0] if rv else None) if one else rv

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

 

1. 기본 경로 ('/')
 "api-server" 문자열 반환 

@app.route('/')
def index():
  return "api-server"

2. 사용자 정보 확인 ('/api/me')

현재 세션에 'uid' 값이 있는지 확인하여 로그인 상태인 경우 사용자 ID를, 아닌 경우에는 None을 반환한다.

• {"userid": 세션에_저장된_uid_값}
• {"userid": null}

@app.route('/api/me')
def me():
  if session.get('uid'):
    return jsonify(userid=session['uid'])
  return jsonify(userid=None)

 

3. 로그인 ('/api/login', POST 방식)

사용자가 form 에 입력한 userid와 password를 받아와 users 테이블에서 일치하는 정보를 찾는다. 

일치하는 사용자가 있는 경우, 세션에 uid를 저장하고 succes 응답을 반환하고, 아닌 경우 fail을 반환한다.

@app.route('/api/login', methods=['POST'])
def login():
  userid = request.form.get('userid', '')
  password = request.form.get('password', '')
  if userid and password:
    ret = query_db(f"SELECT * FROM users where userid='{userid}' and password='{hashlib.sha256(password.encode()).hexdigest()}'" , one=True)
    if ret:
      session['uid'] = ret[0]
      return jsonify(result="success", userid=ret[0])
  return jsonify(result="fail")

4. 로그아웃 ('/api/logout')
세션에서 'uid'를 제거하고  succes 응답을 반환한다, 

@app.route('/api/logout')
def logout():
  session.pop('uid', None)
  return jsonify(result="success")

5. 회원가입 ('/api/join', POST 방식)
사용자가 제출한 userid와 password를 받고, 비밀번호를 SHA-256으로 해시하여 사용자 테이블에 저장한다.
성공하면 success, 실패하면 error 결과를 반환한다.

@app.route('/api/join', methods=['POST'])
def join():
  userid = request.form.get('userid', '')
  password = request.form.get('password', '')
  if userid and password:
    conn = get_db()
    cur = conn.cursor()
    cur.execute("Insert into users values(?, ?);", (userid, hashlib.sha256(password.encode()).hexdigest()))
    conn.commit()
    return jsonify(result="success")
  return jsonify(result="error")

6. 메모 추가 ('/api/memo/add', PUT 방식)

• 로그인 상태 확인: 세션의 UID를 확인하여 로그인되지 않았으면 "no login" 결과 반환
• 세션에서 현재 사용자 ID를 가져와, 사용자가 제출한 title과 contents를 받는다.
• title과 contents가 모두 존재하면, 데이터베이스에 Insert문을 이용해서 메모를 추가하고 요청이 성공하면 메모 ID(lastrowid)와 함께 success 응답을 반환한다.

app.route('/api/memo/add', methods=['PUT'])
def memoAdd():
  if not session.get('uid'):
    return jsonify(result="no login")

  userid = session.get('uid')
  title = request.form.get('title')
  contents = request.form.get('contents')

  if title and contents:
    conn = get_db()
    cur = conn.cursor()
    ret = cur.execute("Insert into memo(userid, title, contents) values(?, ?, ?);", (userid, title, contents))
    conn.commit()
    return jsonify(result="success", memoidx=ret.lastrowid)
  return jsonify(result="error")

 

7. 메모 조회 ('/api/memo/<idx>', GET 방식)

URL 파라미터로 받은 idx를 사용하여 해당 메모를 데이터베이스에서 검색한다.
이때 mode 파라미터에 따라 응답 형식이 달라진다.  // 예시: /api/memo/5?mode=html

• 'html' 모드: 템플릿 문자열을 렌더링하여 HTML 형식으로 메모 내용을 반환
• 그 외: 메모 데이터를 JSON 형식으로 반환

@app.route('/api/memo/<idx>', methods=['GET'])
def memoView(idx):
  mode = request.args.get('mode', 'json')
  ret = query_db("SELECT * FROM memo where idx=" + idx)[0]
  if ret:
    userid = ret['userid']
    title = ret['title']
    contents = ret['contents']
    if mode == 'html':
      template = ''' Written by {userid}<h3>{title}</h3>
      <pre>{contents}</pre>
      '''.format(title=title, userid=userid, contents=contents)
      return render_template_string(template)
    else:
      return jsonify(result="success",
        userid=userid,
        title=title,
        contents=contents)
  return jsonify(result="error")

8. 메모 업데이트 ('/api/memo/int:idx', PUT 방식)

• 로그인 상태 확인: 로그인되지 않았으면 "no login" 결과 반환
• URL 파라미터로 받은 idx로 해당 메모를 데이터베이스에서 검색한다
• 사용자가 제출한 새 title과 contents로 메모를 업데이트하고, 성공하면 "success" 결과를, 실패하면 "error" 결과를 반환한다.

@app.route('/api/memo/<int:idx>', methods=['PUT'])
def memoUpdate(idx):
  if not session.get('uid'):
    return jsonify(result="no login")

  ret = query_db('SELECT * FROM memo where idx=?', [idx,])[0]
  userid = session.get('uid')
  title = request.form.get('title')
  contents = request.form.get('contents')

  if ret and title and contents:
    conn = get_db()
    cur = conn.cursor()
    updateRet = cur.execute("UPDATE memo SET title=?, contents=? WHERE idx=?",(title, contents, idx))
    conn.commit()
    if updateRet:
      return jsonify(result="success")
  return jsonify(result="error")

 

(3) 패치 

1. secret key 하드 코딩 

flask 는 세션관리 및 암호화를 위해 app.secret_key가 사용되는데 이 값이 하드코딩되어있으면 세션을 위조하는 등의 공격이 가능하다. 따라서 랜덤값을 생성하여 secert key로 사용하도록 수정해주었다. 

import secrets

app.secret_key = secrets.token_hex(16)

 

2. login : fstring 사용 -> prepared statement (SQL Injection)

    ret = query_db(f"SELECT * FROM users where userid='{userid}' and password='{hashlib.sha256(password.encode()).hexdigest()}'" , one=True)

로그인할 때 사용하는 쿼리문에서 Prepared Statements을 사용하지 않고 fstring을 이용하여 로그인을 하는 것을 알 수 있다. 이런 경우 SQL Injection이 발생할 수 있으므로, 다른 쿼리들과 마찬가지로 수정해주어야 한다. 

sqlite를 사용하고 있으므로 플레이스홀더는 ?를 사용해주어야 한다. 

- SQLite, MySQL: ? 사용
- PostgreSQL: %s 또는 $1, $2 등 사용
- Oracle: :name 형식 사용

@app.route('/api/login', methods=['POST'])
def login():
  userid = request.form.get('userid', '')
  password = request.form.get('password', '')
  if userid and password:
    ret = query_db("SELECT * FROM users WHERE userid=? AND password=?", 
              (userid, hashlib.sha256(password.encode()).hexdigest()), one=True)
    if ret:
      session['uid'] = ret[0]
      return jsonify(result="success", userid=ret[0])
  return jsonify(result="fail")

 

3. memoView : + 연산자 사용 -> prepared statement  (SQL Injection)

  ret = query_db("SELECT * FROM memo where idx=" + idx)[0]

memoView함수에서 또한 Prepared Statement를 사용하지 않고 문자열 연결(+ 연산자)을 통해 쿼리를 요청한다. 

따라서 SQL Injection의 방어를 위해 수정해주어야 한다. 

@app.route('/api/memo/<idx>', methods=['GET'])
def memoView(idx):
  mode = request.args.get('mode', 'json')
  ret = query_db("SELECT * FROM memo WHERE idx=?", (idx,), one=True)

 

4. memoView : render_template_string 함수 사용 (SSTI)

render_template_string 함수를 사용하면 template로 사용되는 데이터가 사용자의 입력 데이터에 의해 변조될 수 있으며, SSTI가 발생할 수 있다. ( 사용자가 입력하는 데이터(userid, title, contents)가 템플릿에 직접 삽입된 후에 이 템플릿이 Jinja2 엔진으로 해석되기 때문)

이는 escape 함수를 사용하여 수정해주었다. 

from flask import escape
....
if mode == 'html':
      html_content = f'''<p>Written by {escape(userid)}</p>
      <h3>{escape(title)}</h3>
      <pre>{escape(contents)}</pre>'''
      return html_content

 

5. memoUpdate : 사용자 검증 부족 (권한 상승 취약점)

memoUpdate 함수에서는 메모를 업데이트할 때 (수정할 때), 로그인 여부만 확인하고 로그인한 사용자가 그 메모의 작성자인지 확인하지 않는다. 따라서 세션의 uid값을 확인하는 검증 로직을 추가해주어야 한다. 

@app.route('/api/memo/<int:idx>', methods=['PUT'])
def memoUpdate(idx):
  if not session.get('uid'):
    return jsonify(result="no login")

  # 메모 정보 조회
  ret = query_db('SELECT * FROM memo WHERE idx=?', (idx,), one=True)
  if not ret:
    return jsonify(result="error", message="Memo not found")
  
  # 현재 로그인한 사용자
  current_user = session.get('uid')
  
  # 메모 소유자 확인
  if ret['userid'] != current_user:
    return jsonify(result="error", message="Permission denied")
  
  title = request.form.get('title')
  contents = request.form.get('contents')

  if title and contents:
    conn = get_db()
    cur = conn.cursor()
    updateRet = cur.execute("UPDATE memo SET title=?, contents=? WHERE idx=?",
                           (title, contents, idx))
    conn.commit()
    if updateRet:
      return jsonify(result="success")
  return jsonify(result="error")

 

[최종 코드]

#!/usr/bin/python3
from flask import Flask, request, render_template_string, g, session, jsonify, escape
import sqlite3
import os, hashlib, secrets

app = Flask(__name__)
app.secret_key = secrets.token_hex(16)

def get_db():
  db = getattr(g, '_database', None)
  if db is None:
    db = g._database = sqlite3.connect(os.environ['DATABASE'])
  db.row_factory = sqlite3.Row
  return db

def query_db(query, args=(), one=False):
  cur = get_db().execute(query, args)
  rv = cur.fetchall()
  cur.close()
  return (rv[0] if rv else None) if one else rv

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

@app.route('/')
def index():
  return "api-server"

@app.route('/api/me')
def me():
  if session.get('uid'):
    return jsonify(userid=session['uid'])
  return jsonify(userid=None)

@app.route('/api/login', methods=['POST'])
def login():
  userid = request.form.get('userid', '')
  password = request.form.get('password', '')
  if userid and password:
    ret = query_db("SELECT * FROM users WHERE userid=? AND password=?", 
                  (userid, hashlib.sha256(password.encode()).hexdigest()), one=True)
    if ret:
      session['uid'] = ret[0]
      return jsonify(result="success", userid=ret[0])
  return jsonify(result="fail")

@app.route('/api/logout')
def logout():
  session.pop('uid', None)
  return jsonify(result="success")

@app.route('/api/join', methods=['POST'])
def join():
  userid = request.form.get('userid', '')
  password = request.form.get('password', '')
  if userid and password:
    conn = get_db()
    cur = conn.cursor()
    cur.execute("Insert into users values(?, ?);", (userid, hashlib.sha256(password.encode()).hexdigest()))
    conn.commit()
    return jsonify(result="success")
  return jsonify(result="error")

@app.route('/api/memo/add', methods=['PUT'])
def memoAdd():
  if not session.get('uid'):
    return jsonify(result="no login")

  userid = session.get('uid')
  title = request.form.get('title')
  contents = request.form.get('contents')

  if title and contents:
    conn = get_db()
    cur = conn.cursor()
    ret = cur.execute("Insert into memo(userid, title, contents) values(?, ?, ?);", (userid, title, contents))
    conn.commit()
    return jsonify(result="success", memoidx=ret.lastrowid)
  return jsonify(result="error")

@app.route('/api/memo/<idx>', methods=['GET'])
def memoView(idx):
  mode = request.args.get('mode', 'json')
  ret = query_db("SELECT * FROM memo WHERE idx=?", (idx,), one=True)
  if ret:
    userid = ret['userid']
    title = ret['title']
    contents = ret['contents']
    if mode == 'html':
      html_content = f'''<p>Written by {escape(userid)}</p>
      <h3>{escape(title)}</h3>
      <pre>{escape(contents)}</pre>'''
      return html_content
    else:
      return jsonify(result="success",
        userid=userid,
        title=title,
        contents=contents)
  return jsonify(result="error")

@app.route('/api/memo/<int:idx>', methods=['PUT'])
def memoUpdate(idx):
  if not session.get('uid'):
    return jsonify(result="no login")

  # 메모 정보 조회
  ret = query_db('SELECT * FROM memo WHERE idx=?', (idx,), one=True)
  if not ret:
    return jsonify(result="error", message="Memo not found")
  
  # 현재 로그인한 사용자
  current_user = session.get('uid')
  
  # 메모 소유자 확인
  if ret['userid'] != current_user:
    return jsonify(result="error", message="Permission denied")
  
  title = request.form.get('title')
  contents = request.form.get('contents')

  if title and contents:
    conn = get_db()
    cur = conn.cursor()
    updateRet = cur.execute("UPDATE memo SET title=?, contents=? WHERE idx=?",
                          (title, contents, idx))
    conn.commit()
    if updateRet:
      return jsonify(result="success")
  return jsonify(result="error")

 

패치가 완료된 위 코드를 제출하면 플래그를 얻을 수 있다.