FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · 심각
보안 연구 · FlawOpen

Python의 SQL 인젝션 (psycopg2, SQLAlchemy, Django)

cursor.execute() 안에서 f-string이나 %를 쓰면 요청 파라미터가 SQL을 다시 쓰게 되고, 사용자 한 명을 조회하던 쿼리가 테이블 전체 덤프로 바뀝니다(CWE-89). Python의 각 데이터베이스 API별 안전한 호출 방법과, 플레이스홀더로 처리할 수 없는 두 가지 경우인 컬럼 이름과 IN 목록을 정리했습니다.[1], [2], [3], [4], [5], [6], [7], [9]

💡 알기 쉬운 설명 (ELI5)

우편물 분류대에 ‘라벨이 ___인 소포를 분류하라’는 고정 지시 카드가 있다고 상상해 보세요. 작업자가 읽기 전에 발신자의 라벨을 카드에 그대로 옮기면 추가 문구가 새 분류 지시가 될 수 있습니다. Python f-string이나 % 포매팅은 SQL에 똑같은 일을 합니다. 드라이버가 값을 받기도 전에 하나의 명령 문자열을 만듭니다. execute(sql, params)는 지시 카드를 고정하고 라벨을 데이터로 따로 전달합니다. 열 이름은 지시의 일부이므로 코드에서 정한 허용 목록에서 선택해야 합니다.[10], [1], [2], [9]

핵심 개념 및 용어

cursor.execute(sql, params)
DB-API 2.0(PEP 249)의 호출입니다. 드라이버는 params의 값을 SQL 텍스트와 따로 보내므로 값이 구문을 바꿀 수 없습니다. 값이 두 번째 인자에 있을 때만 안전하며, 첫 번째 인자에 f-string이나 % 식을 쓰는 것이 바로 버그입니다.[2], [3]
paramstyle: %s, ?, :name
각 드라이버는 paramstyle 속성에 자신의 플레이스홀더를 선언합니다. psycopg2와 PyMySQL은 %s 또는 %(name)s, sqlite3는 ? 또는 :name을 씁니다. %s는 Python의 % 연산자가 아니라 드라이버가 해석하므로 문자열을 직접 포맷하면 안 됩니다.[2], [3], [5]
SQLAlchemy text()
SQLAlchemy에서 원시 SQL 문자열을 감싸는 함수입니다. :email 같은 이름 있는 바인드 파라미터를 쓰고 값을 dict로 execute()에 넘기면 안전합니다. 넘기는 문자열이 이미 f-string이면 위험합니다. text()는 문자열 안의 내용을 전혀 이스케이프하지 않습니다.[6], [9]
Django raw() / RawSQL / extra()
Django ORM을 우회하는 방법들입니다. raw(sql, params)와 RawSQL(sql, params)는 모든 값을 params로 넘길 때만 안전하며, sql에 끼워 넣은 내용은 Django가 그대로 보냅니다. Django 문서는 extra()를 최후의 수단이라고 하므로 리뷰에서 금지하세요.[7], [9]
psycopg2.sql.Identifier
플레이스홀더는 값만 전달할 뿐 테이블이나 컬럼 이름은 전달할 수 없습니다. sql.Identifier는 이름을 큰따옴표로 감싸 식별자 밖으로 벗어나지 못하게 합니다. 그 전에 허용 목록과 대조해 공개할 생각이 없던 컬럼을 사용자가 고르지 못하게 하세요.[4], [9]
IN lists and expanding binds
sqlite3에서는 Python 리스트를 ? 하나에 바인딩할 수 없어서 결국 문자열 이어 붙이기로 돌아가곤 합니다. SQLAlchemy에서는 확장형 bindparam(요소마다 플레이스홀더 하나), psycopg2에서는 리스트와 id = ANY(%s)를 쓰고, sqlite3에서는 요소 수만큼 ?를 생성하세요.[6], [3], [5]

프레임워크 매핑

단계별 공격 실행 흐름

단계 1

요청 값이 데이터 계층에 도달

요청 매개변수는 텍스트로 get_user()에 전달됩니다. Flask 뷰는 request.args["id"]를 읽고, Django 뷰는 request.GET["id"]를 읽습니다. 뷰는 검증되지 않은 텍스트를 get_user()에 넘깁니다.[11], [12], [9]

단계 2

f-string이 값을 SQL에 끼워 넣음

f"... WHERE id = {user_id}"는 하나의 평평한 문자열을 만듭니다. 무해한 테스트 값 0 OR 1=1을 넣으면 텍스트가 WHERE id = 0 OR 1=1이 되고, 값이 어디서 끝나야 했는지 알려 주는 표시는 아무것도 없습니다.[10], [1], [9]

단계 3

드라이버는 파라미터를 하나도 받지 못함

execute(query)가 SQL만으로 호출되므로 psycopg2는 텍스트를 그대로 보내고, PostgreSQL은 OR 1=1을 데이터가 아니라 WHERE 절의 일부로 파싱합니다.[3], [1], [9]

단계 4

엔드포인트가 모든 행을 반환

조건이 모든 행에서 참이 되어 사용자 한 명을 조회하는 요청이 테이블 전체를 돌려줍니다. psycopg2는 세미콜론으로 구분한 여러 문장을 한 번의 호출로 받아들이기도 합니다. sqlite3는 이를 거부하므로 피해가 줄어들지만 데이터 탈취까지 막지는 못합니다.[13], [5], [1], [9]

소스 코드 비교: 취약한 구현 vs 보안 패치

취약한 구현
# repository.py: 아래 모든 쿼리가 요청 데이터를 SQL 텍스트에 끼워 넣는다
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2: 드라이버가 보기 전에 f-string이 문장을 만들어 버린다
    pg_cursor.execute(f"SELECT id, email FROM users WHERE id = {user_id}")
    return pg_cursor.fetchone()


def find_orders(conn: sqlite3.Connection, status):
    # sqlite3: 이 %는 Python 문자열 포맷이지 드라이버의 플레이스홀더가 아니다
    return conn.execute("SELECT id FROM orders WHERE status = '%s'" % status).fetchall()


def find_by_email(session, email):
    # SQLAlchemy: text()는 이미 문자열 안에 있는 내용을 이스케이프하지 않는다
    return session.execute(text(f"SELECT id FROM users WHERE email = '{email}'")).all()


def find_users(User, username):
    # Django: raw()는 끼워 넣은 값을 데이터베이스에 그대로 보낸다
    return User.objects.raw(f"SELECT * FROM app_user WHERE username = '{username}'")


def get_orders_by_ids(session, ids):
    # IN 목록: 이어 붙인 문자열이면 한 요소가 추가 SQL을 실어 나를 수 있다
    id_list = ", ".join(ids)
    return session.execute(text(f"SELECT id FROM orders WHERE id IN ({id_list})")).all()


def list_users(pg_cursor, sort_column):
    # 식별자: 정렬 컬럼을 쿼리 문자열에서 그대로 받는다
    pg_cursor.execute(f"SELECT id, email FROM users ORDER BY {sort_column}")
    return pg_cursor.fetchall()
보안 강화 패치
# repository.py: 값은 바인드 파라미터로 보내고, 이름은 허용 목록에서 가져온다
import sqlite3

from psycopg2 import sql
from sqlalchemy import bindparam, text

SORTABLE_COLUMNS = {"id", "email", "created_at"}


def get_user(pg_cursor, user_id: int):
    # psycopg2: %s는 드라이버의 플레이스홀더이며 값은 두 번째 인자로 넘긴다
    pg_cursor.execute("SELECT id, email FROM users WHERE id = %s", (user_id,))
    return pg_cursor.fetchone()


def find_orders(conn: sqlite3.Connection, status: str):
    # sqlite3는 qmark 방식: ? 플레이스홀더와 튜플로 넘기는 값
    return conn.execute("SELECT id FROM orders WHERE status = ?", (status,)).fetchall()


def find_by_email(session, email: str):
    # SQLAlchemy: 이름 있는 바인드 파라미터, 값은 execute() 호출 시 전달
    stmt = text("SELECT id FROM users WHERE email = :email")
    return session.execute(stmt, {"email": email}).all()


def find_users(User, username: str):
    # Django: raw()는 params를 따로 보낸다. ORM 필터가 더 간단하다
    return User.objects.raw("SELECT * FROM app_user WHERE username = %s", [username])


def get_orders_by_ids(session, ids: list[int]):
    # IN 목록: 확장형 바인드가 요소마다 플레이스홀더를 만든다
    stmt = text("SELECT id FROM orders WHERE id IN :ids").bindparams(
        bindparam("ids", expanding=True)
    )
    return session.execute(stmt, {"ids": ids}).all()


def list_users(pg_cursor, sort_column: str):
    # 식별자: 먼저 허용 목록을 확인한 뒤 sql.Identifier로 인용한다
    if sort_column not in SORTABLE_COLUMNS:
        raise ValueError(f"unsupported sort column: {sort_column!r}")
    query = sql.SQL("SELECT id, email FROM users ORDER BY {}").format(sql.Identifier(sort_column))
    pg_cursor.execute(query)
    return pg_cursor.fetchall()

엔지니어링 및 시스템 보안 강화 체크리스트

출처