Защита от инъекций (Injection Attacks)

Инъекция — это атака, при которой злоумышленник внедряет вредоносный код через пользовательский ввод. Это одна из топ-10 уязвимостей (OWASP Top 10). Рассмотрю подходы защиты с примерами.

1. Параметризованные запросы (Parameterized Queries)

Самый надёжный способ защиты от SQL injection.

# УЯЗВИМО! Строковая конкатенация
user_input = "'; DROP TABLE users; --"
query = f"SELECT * FROM users WHERE email = '{user_input}'"
# Результат: SELECT * FROM users WHERE email = ''; DROP TABLE users; --'
# Это выполнит DROP TABLE!

# ЗАЩИЩЕНО! Параметризованный запрос
from sqlalchemy import text

query = text("SELECT * FROM users WHERE email = :email")
result = db.execute(query, {"email": user_input})
# Здесь user_input передаётся как параметр, не как часть SQL

Как работает:

1. SQL парсируется отдельно от параметров
2. Параметры передаются в шифрованном/безопасном виде
3. БД знает, где параметры, и не интерпретирует их как SQL код

Python примеры:

# SQLAlchemy ORM (всегда safe)
user = db.session.query(User).filter(User.email == user_input).first()

# SQLAlchemy raw SQL с параметрами (safe)
result = db.execute(
    text("SELECT * FROM users WHERE email = :email"),
    {"email": user_input}
)

# psycopg2 (PostgreSQL)
cursor.execute(
    "SELECT * FROM users WHERE email = %s",
    (user_input,)  # Параметры передаются отдельно
)

# sqlite3
cursor.execute(
    "SELECT * FROM users WHERE email = ?",
    (user_input,)
)

# НИКОГДА не делай так:
cursor.execute(f"SELECT * FROM users WHERE email = '{user_input}'")  # ОПАСНО!

2. ORM (Object-Relational Mapping)

ORM автоматически использует параметризованные запросы.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql://...")
Session = sessionmaker(bind=engine)
session = Session()

# SQLAlchemy автоматически параметризует
users = session.query(User).filter(
    User.email == user_email,  # Safe
    User.age > 18              # Safe
).all()

# Под капотом это становится параметризованным запросом
# SELECT * FROM users WHERE email = %s AND age > %s

Преимущества:

• Защита от SQL injection встроена
• Читаемый код
• Type-safe

3. Валидация входных данных

Ограничь допустимые значения на входе.

from pydantic import BaseModel, EmailStr, validator
from typing import Optional

class UserInput(BaseModel):
    email: EmailStr  # Только валидные emails
    age: int
    name: str
    
    @validator('age')
    def age_valid(cls, v):
        if v < 0 or v > 150:
            raise ValueError('Age must be 0-150')
        return v
    
    @validator('name')
    def name_valid(cls, v):
        if len(v) > 100:
            raise ValueError('Name too long')
        if not v.replace(' ', '').isalnum():  # Только буквы и цифры
            raise ValueError('Invalid characters')
        return v

# Использование
user_input = UserInput(
    email="test@example.com",
    age=25,
    name="John Doe"
)
# Если input невалиден, пример: age=999, будет ValidationError

Типы валидации:

• Type validation (int, str, email)
• Length validation (макс длина)
• Pattern validation (регулярные выражения)
• Whitelist validation (только разрешённые значения)
• Range validation (min/max)

4. Whitelisting (Белые списки)

Разреши только определённые значения.

# УЯЗВИМО: user может передать любое имя столбца
column_name = request.args.get('sort_by')  # Может быть: password, admin_flag и т.п.
query = f"SELECT * FROM users ORDER BY {column_name}"

# ЗАЩИЩЕНО: только разрешённые столбцы
ALLOWED_COLUMNS = {'name', 'email', 'created_at'}
column_name = request.args.get('sort_by', 'created_at')

if column_name not in ALLOWED_COLUMNS:
    raise ValueError(f"Invalid column: {column_name}")

query = f"SELECT * FROM users ORDER BY {column_name}"
# Теперь column_name гарантированно безопасен

Другой пример:

# API endpoints
ALLOWED_ACTIONS = {'create', 'read', 'update', 'delete'}

action = request.json.get('action')
if action not in ALLOWED_ACTIONS:
    raise ValueError(f"Invalid action: {action}")

if action == 'create':
    # ...
elif action == 'read':
    # ...

5. Экранирование (Escaping)

Хотя менее надёжно, чем параметризация, помогает в некоторых случаях.

import html
import json

# HTML escaping (для вывода в веб)
user_input = "<script>alert('XSS')</script>"
escaped = html.escape(user_input)  # &lt;script&gt;alert('XSS')&lt;/script&gt;

# JSON escaping
escaped_json = json.dumps({"user_input": user_input})
# {"user_input": "<script>alert('XSS')<\/script>"}

# SQL escaping (менее надёжно, используй параметризацию!)
from sqlalchemy import literal_column
# Некоторые БД имеют функции escaping, но это НЕ рекомендуется

Важно: экранирование МЕНЕЕ безопасно чем параметризация, используй его как дополнение, не как основу.

6. Command Injection защита

Защита от инъекций в системные команды.

import subprocess

# УЯЗВИМО! Shell injection
user_filename = input("Filename: ")  # Может быть: test.txt; rm -rf /
subprocess.run(f"cat {user_filename}", shell=True)  # Опасно!

# ЗАЩИЩЕНО! Без shell
subprocess.run(["cat", user_filename], shell=False)  # Safe
# Аргументы передаются массивом, не через shell

# ЗАЩИЩЕНО! Валидация
import re
if not re.match(r'^[a-zA-Z0-9._-]+$', user_filename):
    raise ValueError("Invalid filename")
subprocess.run(f"cat {user_filename}", shell=True)  # Теперь safe

Правило: если передаёшь пользовательский ввод команде, используй shell=False и массив аргументов.

7. XSS (Cross-Site Scripting) защита

Защита от инъекции JavaScript в веб-страницы.

from flask import Flask, render_template_string, escape

app = Flask(__name__)

@app.route('/user/<username>')
def user_profile(username):
    # УЯЗВИМО!
    html = f"<h1>Welcome {username}</h1>"
    return html
    # Если username = "<script>alert('XSS')</script>"
    # То скрипт будет выполнен!
    
    # ЗАЩИЩЕНО! Escape output
    html = f"<h1>Welcome {escape(username)}</h1>"
    return html
    # Теперь тег escape-ится: &lt;script&gt;...

Также нужен CSP (Content Security Policy) заголовок:

@app.route('/page')
def page():
    response.headers['Content-Security-Policy'] = "script-src 'self'"
    # Блокирует inline скрипты и скрипты с других источников
    return response

8. NoSQL Injection защита

Дажe в NoSQL нужна защита.

from pymongo import MongoClient

client = MongoClient()
db = client['myapp']
users = db['users']

# УЯЗВИМО!
email = request.args.get('email')  # Может быть {"$ne": null}
query = {"email": email}
user = users.find_one(query)  # Вернёт любого пользователя!

# ЗАЩИЩЕНО! Валидация
from pydantic import EmailStr
from pydantic import BaseModel

class EmailQuery(BaseModel):
    email: EmailStr

email_query = EmailQuery(email=email)  # Валидирует
query = {"email": email_query.email}  # Теперь safe
user = users.find_one(query)

9. LDAP Injection защита

Для LDAP запросов.

import ldap

# УЯЗВИМО!
username = input("Username: ")  # Может быть: *)(uid=*))(|(uid=*
filter_str = f"(uid={username})"
result = ldap.initialize("ldap://localhost").search_s(
    "dc=example,dc=com",
    ldap.SCOPE_SUBTREE,
    filter_str  # Опасно!
)

# ЗАЩИЩЕНО! Экранирование LDAP
from ldap.filter import escape_filter_chars

username = input("Username: ")
filter_str = f"(uid={escape_filter_chars(username)})"
result = ldap.initialize("ldap://localhost").search_s(
    "dc=example,dc=com",
    ldap.SCOPE_SUBTREE,
    filter_str  # Теперь safe
)

10. Логирование и мониторинг

Обнаружение попыток инъекции.

import logging

logger = logging.getLogger(__name__)

# Patterns для обнаружения инъекций
Injection_PATTERNS = [
    r"(?i)('|(\-\-)|(;)|(\|\|)|(\*)).*?(union|select|insert|update|delete|drop)",
    r"(?i)(<script|javascript:|onerror=|onload=)",
    r"(?i)(\.\.|\/|\\\\)",
]

def check_injection(user_input: str) -> bool:
    import re
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, user_input):
            logger.warning(f"Potential injection detected: {user_input}")
            return True
    return False

# В middleware
@app.before_request
def check_input():
    for param in request.args.values():
        if check_injection(param):
            abort(400, "Invalid input")

Сравнение методов

Чеклист безопасности

# ✅ DO

# 1. Используй ORM или параметризованные запросы
db.execute(text("SELECT * FROM users WHERE id = :id"), {"id": user_id})

# 2. Валидируй входные данные
from pydantic import BaseModel, validator
class UserInput(BaseModel):
    email: str
    
    @validator('email')
    def email_valid(cls, v):
        # ...

# 3. Используй whitelist для перечисляемых значений
if action not in ALLOWED_ACTIONS:
    raise ValueError()

# 4. Экранируй вывод в HTML
html.escape(user_input)

# 5. Используй shell=False для команд
subprocess.run(["cmd", arg], shell=False)

# 6. Логируй подозрительную активность
logger.warning(f"Suspicious input: {input}")

# ❌ DON'T

# 1. Строковая конкатенация в SQL
db.execute(f"SELECT * FROM users WHERE id = {user_id}")

# 2. exec() или eval() с пользовательским вводом
eval(user_input)  # НИКОГДА!

# 3. shell=True с пользовательским вводом
subprocess.run(f"cat {filename}", shell=True)

# 4. Вывод пользовательского ввода без экранирования
return f"<h1>{user_input}</h1>"

# 5. Доверие фильтрам (нужна валидация на сервере)
if "script" not in user_input:  # Легко обойти
    # ...

Заключение

Защита от инъекций это не опция, это необходимость. Следуй правилу 80/20:

80% проблем решают:

1. Параметризованные запросы (SQL, NoSQL)
2. ORM (SQLAlchemy, Tortoise)
3. Валидация входных данных (Pydantic)
4. Экранирование вывода (html.escape)
5. shell=False (subprocess)

20% проблем решают дополнительные меры (CSP, LDAP escaping, мониторинг).

В моих проектах я всегда использую ORM + Pydantic валидацию + экранирование вывода. Это базовая гигиена безопасности.