SQL Injection

# ✅ SQL Injection 방어 - Prepared Statement

# ===== Python (sqlite3) =====
import sqlite3

def login_secure(username, password):
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()

    # 안전: 파라미터 바인딩 (? 플레이스홀더)
    query = "SELECT * FROM users WHERE username = ? AND password = ?"
    cursor.execute(query, (username, password))
    return cursor.fetchone()

# ' OR '1'='1 입력해도 문자열로 처리됨
# 쿼리 구조와 데이터가 분리되어 SQL로 해석되지 않음


# ===== Python (psycopg2 - PostgreSQL) =====
import psycopg2

def get_user_secure(user_id):
    conn = psycopg2.connect(database="mydb")
    cursor = conn.cursor()

    # %s 플레이스홀더 사용
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    return cursor.fetchone()


# ===== Python (mysql-connector) =====
import mysql.connector

def search_products_secure(keyword):
    conn = mysql.connector.connect(host='localhost', database='shop')
    cursor = conn.cursor(prepared=True)

    # 이름 있는 파라미터
    query = "SELECT * FROM products WHERE name LIKE %(keyword)s"
    cursor.execute(query, {'keyword': f'%{keyword}%'})
    return cursor.fetchall()


# ===== Node.js (mysql2) =====
# const mysql = require('mysql2/promise');
#
# async function getUser(userId) {
#     const connection = await mysql.createConnection({...});
#
#     // 파라미터화된 쿼리
#     const [rows] = await connection.execute(
#         'SELECT * FROM users WHERE id = ?',
#         [userId]
#     );
#     return rows[0];
# }


# ===== Java (JDBC) =====
# String query = "SELECT * FROM users WHERE username = ? AND password = ?";
# PreparedStatement pstmt = connection.prepareStatement(query);
# pstmt.setString(1, username);
# pstmt.setString(2, password);
# ResultSet rs = pstmt.executeQuery();


# ===== 추가 방어 계층 =====

# 1. 입력 검증 (화이트리스트)
def validate_order_column(column):
    allowed = ['id', 'name', 'created_at', 'price']
    if column not in allowed:
        raise ValueError(f"Invalid column: {column}")
    return column

# 2. 최소 권한 원칙
# - 애플리케이션 DB 계정에 SELECT, INSERT, UPDATE만 부여
# - DROP, ALTER, CREATE 권한 제거
# - 민감한 테이블에 대한 접근 제한

# 3. 에러 메시지 숨기기
# - 프로덕션에서 상세 SQL 에러 메시지 숨김
# - 로그에만 기록, 사용자에게는 일반 에러 표시

# ✅ ORM 사용 - SQL Injection 자동 방어

# ===== Python (SQLAlchemy) =====
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    email = Column(String(100))
    password = Column(String(255))

# DB 연결
engine = create_engine('postgresql://user:pass@localhost/mydb')
Session = sessionmaker(bind=engine)
session = Session()

# 안전한 쿼리 - ORM이 자동으로 파라미터화
def login_with_orm(username, password):
    # filter()는 내부적으로 Prepared Statement 사용
    user = session.query(User).filter(
        User.username == username,
        User.password == password
    ).first()
    return user

# LIKE 검색도 안전
def search_users(keyword):
    return session.query(User).filter(
        User.username.like(f'%{keyword}%')
    ).all()


# ===== Python (Django ORM) =====
# from myapp.models import User
#
# # 자동으로 안전한 쿼리 생성
# user = User.objects.filter(username=username, password=password).first()
#
# # Raw SQL 사용 시에도 파라미터 바인딩
# User.objects.raw('SELECT * FROM users WHERE id = %s', [user_id])


# ===== Node.js (Prisma) =====
# const user = await prisma.user.findFirst({
#     where: {
#         username: username,
#         password: password
#     }
# });


# ===== Node.js (Sequelize) =====
# const user = await User.findOne({
#     where: {
#         username: username,
#         password: password
#     }
# });


# ===== 주의: Raw Query 사용 시 =====

# SQLAlchemy raw query (안전)
from sqlalchemy import text

result = session.execute(
    text("SELECT * FROM users WHERE id = :user_id"),
    {"user_id": user_id}
)

# Django raw query (안전)
# User.objects.raw('SELECT * FROM users WHERE id = %s', [user_id])

# ❌ 위험한 사용 (문자열 포맷팅)
# session.execute(f"SELECT * FROM users WHERE id = {user_id}")  # 절대 금지!


# ===== 정렬/컬럼명은 파라미터화 불가 =====

# 동적 ORDER BY는 화이트리스트로 검증
ALLOWED_SORT_COLUMNS = ['id', 'name', 'created_at']

def get_sorted_users(sort_column, sort_order):
    if sort_column not in ALLOWED_SORT_COLUMNS:
        sort_column = 'id'
    if sort_order not in ['asc', 'desc']:
        sort_order = 'asc'

    # 검증된 값만 사용
    from sqlalchemy import asc, desc
    order_func = asc if sort_order == 'asc' else desc
    column = getattr(User, sort_column)

    return session.query(User).order_by(order_func(column)).all()