Flask에서 소유자 검사로 IDOR 막기

IDOR(Insecure Direct Object Reference)는 요청에 담긴 객체 번호만 믿고 그 객체가 요청한 사람의 것인지 확인하지 않을 때 생기는 문제입니다. 이 글은 주문 조회 API가 있는 작은 Flask 앱으로 소유자 검사가 빠진 코드와 고친 코드를 비교하고, 목록, 수정, 삭제까지 같은 검사를 적용한 뒤 pytest 표로 고정합니다.

취약한 코드는 직접 만든 연습용이며 실제 서비스에 쓰면 안 됩니다. 요청은 자신이 띄운 127.0.0.1 서비스에만 보냅니다. SQL 인젝션과 경로 조작은 작은 Flask 서비스를 직접 공격하고 한 줄씩 고치기에서, 업로드 방어는 Flask 파일 업로드 검증과 경로 조작 방어하기에서 다루었으므로 이 글은 접근 제어에만 집중합니다.

준비물

명령은 Windows Git Bash 기준이며 소요 시간은 25분 안팎입니다. PowerShell에서는 curl 대신 curl.exe를 쓰고 따옴표 처리가 달라지므로 Git Bash를 권합니다. Linux나 macOS에서는 python을 python3으로, 가상환경 활성화를 source .venv/bin/activate로 바꾸면 됩니다.

1. 작업 폴더와 공통 코드 만들기

mkdir idor-demo && cd idor-demo
python -m venv .venv
source .venv/Scripts/activate
pip install flask==3.1.3 pytest==9.1.1

base.py는 두 앱이 함께 쓰는 부분입니다. 사용자 3명(일반 사용자 alice와 carol, 관리자 admin)과 주문 4건을 만들고, before_request에서 세션의 사용자 번호로 g.user를 채웁니다. 주문 1, 2번은 alice, 3, 4번은 carol의 것입니다. 세션은 Flask가 서명한 쿠키이며, Flask 공식 문서는 세션을 쓰려면 비밀 키를 설정해야 한다고 안내합니다. 아래 키는 연습용 값이므로 실제로는 python -c "import secrets; print(secrets.token_hex())" 같은 방법으로 만든 값을 씁니다.

import os
import sqlite3
import uuid
from functools import wraps

from flask import Flask, abort, g, jsonify, request, session
from werkzeug.security import check_password_hash, generate_password_hash

USERS = [
    (1, "alice", "alice-pw", "user"),
    (2, "carol", "carol-pw", "user"),
    (3, "admin", "admin-pw", "admin"),
]
ORDERS = [
    (1, 1, "키보드", 35000),
    (2, 1, "마우스", 18000),
    (3, 2, "모니터", 210000),
    (4, 2, "웹캠", 45000),
]


def init_db(path):
    if os.path.exists(path):
        os.remove(path)
    conn = sqlite3.connect(path)
    conn.executescript("""
    CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT UNIQUE,
                        password_hash TEXT, role TEXT);
    CREATE TABLE orders (id INTEGER PRIMARY KEY, owner_id INTEGER, item TEXT,
                         price INTEGER, ref TEXT UNIQUE);
    """)
    conn.executemany("INSERT INTO users VALUES (?, ?, ?, ?)",
                     [(i, n, generate_password_hash(p), r) for i, n, p, r in USERS])
    conn.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)",
                     [(i, o, it, pr, uuid.uuid4().hex) for i, o, it, pr in ORDERS])
    conn.commit()
    conn.close()


def get_db():
    if "db" not in g:
        g.db = sqlite3.connect(g.app_db_path)
        g.db.row_factory = sqlite3.Row
    return g.db


def login_required(view):
    @wraps(view)
    def wrapper(*args, **kwargs):
        if g.user is None:
            abort(401)
        return view(*args, **kwargs)
    return wrapper


def create_base(db_path):
    app = Flask(__name__)
    app.config["SECRET_KEY"] = "dev-only-change-me"
    app.json.ensure_ascii = False  # 한글을 그대로 보이게 한다
    init_db(db_path)

    @app.before_request
    def load_user():
        g.app_db_path = db_path
        g.user = None
        if "user_id" in session:
            g.user = get_db().execute(
                "SELECT id, username, role FROM users WHERE id = ?",
                (session["user_id"],)).fetchone()

    @app.teardown_appcontext
    def close_db(exc):
        db = g.pop("db", None)
        if db is not None:
            db.close()

    @app.post("/login")
    def login():
        data = request.get_json(silent=True) or {}
        row = get_db().execute("SELECT * FROM users WHERE username = ?",
                               (data.get("username", ""),)).fetchone()
        if row is None or not check_password_hash(row["password_hash"],
                                                  data.get("password", "")):
            return jsonify(error="로그인 실패"), 401
        session.clear()
        session["user_id"] = row["id"]
        return jsonify(user=row["username"])

    @app.errorhandler(401)
    def unauthorized(e):
        return jsonify(error="로그인이 필요합니다"), 401

    @app.errorhandler(404)
    def not_found(e):
        # 이 예제의 404 는 모두 주문 조회에서 나오므로 같은 메시지를 쓴다
        return jsonify(error="없는 주문입니다"), 404

    return app

2. 로그인만 확인하는 코드와 이를 잡는 테스트

vuln_app.py의 주문 조회는 로그인 여부(login_required)만 보고, 쿼리는 주문 번호만으로 찾습니다. 로그인한 사용자라면 누구의 주문이든 번호만 바꾸면 응답이 옵니다.

from flask import jsonify

from base import create_base, get_db, login_required


def create_app(db_path="vuln.db"):
    app = create_base(db_path)

    @app.get("/orders/<int:order_id>")
    @login_required
    def get_order(order_id):
        row = get_db().execute("SELECT * FROM orders WHERE id = ?",
                               (order_id,)).fetchone()
        if row is None:
            return jsonify(error="없는 주문입니다"), 404
        return jsonify(dict(row))

    return app


if __name__ == "__main__":
    create_app().run(host="127.0.0.1", port=19901)

테스트는 한 건입니다. carol으로 로그인해 alice의 1번 주문을 요청하고, 404가 와야 한다고 적습니다. 이 앱은 아직 고치기 전이므로 실패하는 것이 정상입니다.

from vuln_app import create_app


def test_carol_cannot_read_alices_order(tmp_path):
    client = create_app(str(tmp_path / "t.db")).test_client()
    client.post("/login", json={"username": "carol", "password": "carol-pw"})
    res = client.get("/orders/1")  # 1번 주문은 alice 의 것
    assert res.status_code == 404, f"남의 주문이 보입니다: {res.get_json()}"

위 두 파일을 vuln_app.py, test_vuln.py로 저장하고 실행합니다.

python -m pytest -q --tb=short test_vuln.py

F                                                                        [100%]
...
E   AssertionError: 남의 주문이 보입니다: {'id': 1, 'item': '키보드', 'owner_id': 1, ...}
E   assert 200 == 404
1 failed in 0.53s

...는 줄여 쓴 부분이며 화면에는 전체 값이 나옵니다.

번호 1을 바꾸는 것만으로 다른 사용자의 데이터가 보입니다. 문제는 로그인 검사가 아니라, 주문과 사용자를 연결하는 조건이 없다는 점입니다. 이 현상은 OWASP API Security Top 10의 API1 Broken Object Level Authorization과 같은 종류입니다.

3. 쿼리에 소유자 조건을 넣고 번호를 받는 엔드포인트에 적용하기

고칠 때 가장 중요한 것은 검사를 한 곳에 모으는 일입니다. 엔드포인트마다 if order.owner_id != g.user.id를 따로 쓰면 하나를 빠뜨리기 쉽습니다. 그래서 find_order 함수 하나가 소유자 조건을 쿼리에 넣고, owned_order 데코레이터가 주문 번호를 받는 엔드포인트(조회, 수정, 삭제)에서 이 함수를 부릅니다. 관리자만 조건 없이 조회합니다. 쿼리는 허용한 두 열(id, ref)에 대해 문장 전체를 미리 적어 두고 고르는 방식입니다. 열 이름을 f-string으로 끼우면 SQL 문자열 이어붙이기 금지 원칙에 어긋나고 Bandit이 경고하는 모양이 되기 때문입니다.

app.py로 저장합니다.

from functools import wraps

from flask import abort, g, jsonify, request

from base import create_base, get_db, login_required


def is_admin():
    return g.user["role"] == "admin"


# 열 이름을 문자열에 끼우지 않도록 쿼리 전체를 미리 적어 둔다 (관리자용, 일반 사용자용)
QUERIES = {
    "id": ("SELECT * FROM orders WHERE id = ?",
           "SELECT * FROM orders WHERE id = ? AND owner_id = ?"),
    "ref": ("SELECT * FROM orders WHERE ref = ?",
            "SELECT * FROM orders WHERE ref = ? AND owner_id = ?"),
}


def find_order(column, value):
    """소유자 조건을 쿼리에 넣는 유일한 곳. 관리자만 조건 없이 조회한다."""
    admin_sql, user_sql = QUERIES[column]
    if not is_admin():
        return get_db().execute(user_sql, (value, g.user["id"])).fetchone()
    return get_db().execute(admin_sql, (value,)).fetchone()


def owned_order(view):
    """주문 번호를 받는 모든 엔드포인트에 붙이는 데코레이터."""
    @wraps(view)
    @login_required
    def wrapper(order_id, *args, **kwargs):
        order = find_order("id", order_id)
        if order is None:
            abort(404)  # 없는 주문과 남의 주문을 같은 응답으로 돌려준다
        return view(order, *args, **kwargs)
    return wrapper


def create_app(db_path="fixed.db"):
    app = create_base(db_path)

    @app.get("/orders")
    @login_required
    def list_orders():
        if is_admin():
            rows = get_db().execute("SELECT * FROM orders").fetchall()
        else:
            rows = get_db().execute("SELECT * FROM orders WHERE owner_id = ?",
                                    (g.user["id"],)).fetchall()
        return jsonify([dict(r) for r in rows])

    @app.get("/orders/<int:order_id>")
    @owned_order
    def get_order(order):
        return jsonify(dict(order))

    @app.patch("/orders/<int:order_id>")
    @owned_order
    def update_order(order):
        item = (request.get_json(silent=True) or {}).get("item", "")
        db = get_db()
        db.execute("UPDATE orders SET item = ? WHERE id = ?", (item, order["id"]))
        db.commit()
        return jsonify(id=order["id"], item=item)

    @app.delete("/orders/<int:order_id>")
    @owned_order
    def delete_order(order):
        db = get_db()
        db.execute("DELETE FROM orders WHERE id = ?", (order["id"],))
        db.commit()
        return "", 204

    @app.get("/o/<ref>")
    @login_required
    def get_by_ref(ref):
        order = find_order("ref", ref)
        if order is None:
            abort(404)
        return jsonify(dict(order))

    return app


if __name__ == "__main__":
    create_app().run(host="127.0.0.1", port=19902)

예외가 두 곳 있습니다. 목록(/orders)은 번호를 받지 않으므로 데코레이터 대신 owner_id 조건을 직접 씁니다. 목록이 소유자 조건 없이 SELECT *만 하면, 번호를 맞힐 필요도 없이 전체가 나옵니다. /o/<ref>는 번호 대신 UUID 문자열을 받는 경로여서 owned_order를 쓰지 않고 find_order("ref", ref)를 직접 부릅니다. 같은 함수를 거치므로 소유자 조건은 똑같이 적용됩니다. 수정과 삭제는 조회한 order의 id로만 쿼리하므로, 데코레이터를 통과하지 못하면 쿼리 자체가 실행되지 않습니다.

앱을 띄우고, 새 터미널을 열어 같은 폴더에서 가상환경을 켠 뒤 alice로 로그인해 요청합니다. 로그인 결과 쿠키는 alice.txt에 저장하고 이후 요청에서 읽습니다. 앱을 띄운 터미널과 이 새 터미널은 5절까지 그대로 둡니다. 5절의 명령은 이 서버와 alice.txt를 이어서 쓰며, 6절과 7절은 서버 없이 pytest만 실행하므로 서버를 꺼도 됩니다.

python app.py

curl -s -c alice.txt -H 'Content-Type: application/json' -d '{"username":"alice","password":"alice-pw"}' http://127.0.0.1:19902/login
curl -s -b alice.txt -w ' (%{http_code})\n' http://127.0.0.1:19902/orders/1
curl -s -b alice.txt -w ' (%{http_code})\n' http://127.0.0.1:19902/orders/3
curl -s -b alice.txt -w ' (%{http_code})\n' http://127.0.0.1:19902/orders/999
curl -s -b alice.txt -w ' (%{http_code})\n' -X DELETE http://127.0.0.1:19902/orders/3
curl -s -b alice.txt http://127.0.0.1:19902/orders

{"user":"alice"}
{"id":1,"item":"키보드","owner_id":1,"price":35000,"ref":"21fc91b0..."}
 (200)
{"error":"없는 주문입니다"}
 (404)
{"error":"없는 주문입니다"}
 (404)
{"error":"없는 주문입니다"}
 (404)
[{"id":1,...},{"id":2,...}]

...는 줄여 쓴 부분이며 화면에는 전체 값이 나옵니다. ref 값은 실행할 때마다 달라지므로 이 글의 값과 다릅니다.

carol의 3번 주문을 지우려던 요청도 404로 끝났고, 3번 주문은 그대로 남아 있습니다.

4. 403과 404 중 무엇을 돌려줄까

남의 주문에 403을 돌려주면 "그 번호의 주문은 있지만 당신 것이 아니다"라는 사실이 함께 전달되어 어떤 번호가 쓰이는지 알려 줍니다. 404를 돌려주면 없는 주문과 남의 주문이 구분되지 않습니다. 이 글은 404로 통일했고, 테스트로 두 응답이 같은지 확인합니다(5단계의 test_other_user_and_missing_look_the_same).

다만 404 통일은 이 글의 선택이며 OWASP의 권고가 아닙니다. API1 문서와 A01 문서 모두 상태 코드를 어느 쪽으로 정하라는 내용은 없습니다. 주문처럼 번호가 있다는 사실 자체가 민감한 자원에는 404가 어울리고, 팀 문서처럼 존재를 알려도 무방하고 사용자에게 "권한이 없다"고 알려 주는 편이 나은 화면에는 403이 어울립니다. 어느 쪽이든 API 전체에서 같은 규칙을 쓰고, 그 규칙을 테스트에 적어 두는 것이 좋습니다.

404로 통일해도 한계가 있습니다. 두 경우의 응답 시간이나 크기가 다르면 그 차이로 존재 여부가 드러날 수 있고, 이 글의 테스트는 상태 코드와 본문만 비교합니다. 번호를 하나씩 바꿔 가며 시도하는 열거는 소유자 검사로는 막을 수 없으므로 요청 횟수 제한(rate limit)을 따로 두어야 하며, 이 글에서는 다루지 않습니다.

5. UUID는 보조 수단이고 소유자 검사를 대신하지 않는다

순차 번호는 1, 2, 3으로 추측하기 쉽습니다. 그래서 주문마다 ref라는 UUID를 두고 /o/<ref>로도 조회할 수 있게 했습니다. OWASP API1 문서도 레코드 ID로 예측하기 어려운 값을 쓰기를 권하지만, 그것은 추가 방어입니다. 링크가 로그, 메일, 브라우저 기록으로 새면 UUID를 아는 사람이 생기고, 그때 막아 주는 것은 소유자 검사뿐입니다. /o/<ref>도 같은 find_order를 거치므로, 관리자가 본 carol의 ref를 alice가 넣어도 404입니다.

먼저 admin과 carol으로도 로그인해 쿠키 파일을 만듭니다.

curl -s -c admin.txt -H 'Content-Type: application/json' -d '{"username":"admin","password":"admin-pw"}' http://127.0.0.1:19902/login
curl -s -c carol.txt -H 'Content-Type: application/json' -d '{"username":"carol","password":"carol-pw"}' http://127.0.0.1:19902/login

이어서 admin이 본 carol의 ref를 alice와 carol이 각각 넣어 봅니다.

REF=$(curl -s -b admin.txt http://127.0.0.1:19902/orders/3 | grep -o '"ref":"[0-9a-f]*"' | cut -d'"' -f4); echo ref=$REF; echo -n 'alice: '; curl -s -b alice.txt -w ' (%{http_code})\n' http://127.0.0.1:19902/o/$REF; echo -n 'carol: '; curl -s -b carol.txt -w ' (%{http_code})\n' http://127.0.0.1:19902/o/$REF

ref=76b299534ae740f8a5dc2ceddc776d9b
alice: {"error":"없는 주문입니다"}
 (404)
carol: {"id":3,"item":"모니터","owner_id":2,"price":210000,"ref":"76b299534ae740f8a5dc2ceddc776d9b"}
 (200)

6. 사용자와 역할 조합 표로 회귀 테스트 만들기

수동 확인은 한 번뿐이므로, 사용자와 역할, 메서드, 주문 번호 조합을 표로 적어 pytest에 넣습니다. OWASP A01 문서도 접근 제어 단위 테스트와 통합 테스트를 포함하라고 안내합니다. test_orders.py로 저장합니다.

import pytest

from app import create_app

PASSWORDS = {"alice": "alice-pw", "carol": "carol-pw", "admin": "admin-pw"}
# 주문 1, 2 는 alice, 3, 4 는 carol 의 것이다.
CASES = [
    # (사용자, 메서드, 주문 번호, 기대 상태)
    ("alice", "GET", 1, 200),
    ("alice", "GET", 3, 404),
    ("alice", "PATCH", 3, 404),
    ("alice", "DELETE", 3, 404),
    ("alice", "GET", 999, 404),
    ("carol", "GET", 1, 404),
    ("carol", "PATCH", 1, 404),
    ("carol", "DELETE", 1, 404),
    ("carol", "PATCH", 3, 200),
    ("carol", "DELETE", 4, 204),
    ("admin", "GET", 3, 200),
    ("admin", "PATCH", 1, 200),
    (None, "GET", 1, 401),
    (None, "DELETE", 1, 401),
]


@pytest.fixture
def make_client(tmp_path):
    app = create_app(str(tmp_path / "t.db"))  # 테스트 한 건에 앱과 DB 한 벌

    def make(user):
        client = app.test_client()
        if user:
            client.post("/login", json={"username": user, "password": PASSWORDS[user]})
        return client
    return make


@pytest.mark.parametrize("user,method,order_id,expected", CASES)
def test_access_matrix(make_client, user, method, order_id, expected):
    client = make_client(user)
    res = client.open(f"/orders/{order_id}", method=method, json={"item": "x"})
    assert res.status_code == expected


def test_list_contains_only_own_orders(make_client):
    ids = [o["id"] for o in make_client("alice").get("/orders").get_json()]
    assert ids == [1, 2]


def test_other_user_and_missing_look_the_same(make_client):
    client = make_client("alice")
    other = client.get("/orders/3")
    missing = client.get("/orders/999")
    assert (other.status_code, other.get_json()) == (missing.status_code, missing.get_json())


def test_knowing_the_uuid_is_not_enough(make_client):
    ref = make_client("admin").get("/orders/3").get_json()["ref"]
    assert make_client("alice").get(f"/o/{ref}").status_code == 404
    assert make_client("carol").get(f"/o/{ref}").status_code == 200

create_app이 테스트마다 DB를 새로 만들므로 삭제 테스트가 다른 테스트에 영향을 주지 않습니다. 이 명령은 서버를 켜 두지 않아도 됩니다. 표는 아래와 같습니다. 열은 사용자, 행은 대상 주문입니다.

python -m pytest -v --no-header test_orders.py

collected 17 items
...
test_orders.py::test_knowing_the_uuid_is_not_enough PASSED               [100%]
============================= 17 passed in 6.21s ==============================

7. 검사를 빼면 테스트가 실패하는지 확인하기

테스트가 실제로 소유자 검사를 지키는지 보려고, 복사본에서 소유자 조건만 일부러 끕니다. 원본은 건드리지 않습니다.

mkdir mut && cp base.py app.py test_orders.py mut/ && cd mut
sed -i 's/    if not is_admin():/    if False:/' app.py && python -m pytest -q --tb=no test_orders.py

.FFF.FFF.......FF                                                        [100%]
FAILED test_orders.py::test_access_matrix[alice-GET-3-404] - assert 200 == 404
...
8 failed, 9 passed in 6.34s

확인한 뒤 cd .. && rm -rf mut으로 복사본을 지웁니다. macOS의 sed는 -i 다음에 ''가 필요합니다.

자주 막히는 곳

이어서 해 볼 것

• 조직 단위 소유권: owner_id 한 칸 대신 팀 소속 테이블을 두고, 같은 팀 구성원이 서로의 주문을 읽도록 find_order의 조건을 바꿔 봅니다. 표에 팀원 행을 추가해 테스트가 먼저 실패하는지 봅니다.
• 접근 실패 기록: 404를 돌려주기 전에 로그에 사용자와 주문 번호를 남깁니다. 한 사용자가 짧은 시간에 여러 번호를 시도하면 눈에 띄도록, 반복 횟수를 세어 봅니다.
• 응답 필드 제한: jsonify(dict(order))는 모든 컬럼을 내보냅니다. 응답에 필요한 필드만 골라 보내도록 바꾸고, 테스트에서 owner_id가 응답에 없는지 확인합니다.

확인 범위

직접 실행한 부분은 1단계부터 7단계까지입니다. 환경은 Windows 11, Git Bash, Python 3.12.10, Flask 3.1.3, Werkzeug 3.1.9, pytest 9.1.1, curl 8.19.0이고, 모든 요청은 127.0.0.1에 보냈습니다.

근거로 삼은 문서는 다음과 같습니다. OWASP API Security Top 10 2023의 API1 Broken Object Level Authorization은 객체 번호를 받는 곳에서 사용자별 권한을 확인하고, 레코드 ID에는 예측하기 어려운 값을 쓰며, 권한 테스트를 작성하라고 안내합니다. OWASP Top 10:2021 A01 Broken Access Control은 공개 자원을 빼고는 기본을 거부로 두고, 접근 제어를 한 번 구현해 재사용하며, 레코드 소유권을 강제하고, 접근 제어 단위 테스트와 통합 테스트를 포함하고, API와 컨트롤러 접근에 요청 횟수 제한을 두라고 안내합니다. 두 OWASP 문서 모두 403과 404 중 무엇을 쓸지는 다루지 않습니다. Flask 공식 문서의 Quickstart 세션 절은 세션이 쿠키 위에서 서명되며 세션을 쓰려면 비밀 키를 설정해야 한다고 설명하고, Flask.before_request 문서는 요청마다 실행되며 g.user에 사용자를 담는 예를 보여 줍니다.

브라우저 로그인 화면, 운영용 WSGI 서버, 사용자 수가 많을 때의 성능, 응답 시간 차이로 존재 여부가 드러나는지, Linux와 macOS의 명령(source .venv/bin/activate, macOS용 sed -i '')은 실행하지 못했습니다. 고친 app.py에는 Bandit 1.9.4로 bandit app.py base.py vuln_app.py를 실행해도 SQL 문자열 조립을 뜻하는 B608이 나오지 않았고, 연습용 SECRET_KEY 때문에 B105(하드코딩된 비밀번호 의심, Low) 한 건만 나왔습니다. 이 예제의 비밀번호와 SECRET_KEY는 연습용 값이며 실제 서비스에서는 쓰지 않습니다. 역할이 둘뿐인 단순한 모델이므로 더 복잡한 권한 구조는 별도 설계가 필요합니다.