OWASP Top 10:2025, инъекции, десериализация, валидация, секреты, HTTPS, CVE
- Маршрут: Работа в эксплуатации · тема 4 из 6
- Навыки:
M7,S8- До урока:
python_basics,exceptions,python_testing,stdlib. Тест воспроизведения уязвимости опирается наtest_doubles.- Результат: защита границ ввода, секретов и опасных API на основе угроз.
- Основа: валидация, безопасные значения по умолчанию и управление секретами. Работа в эксплуатации: модель угроз, сценарии злоупотребления, проверка зависимостей и тест воспроизведения уязвимости.
- Углубление: системный разбор безопасности после модуля эксплуатации.
- Подтверждение: модель угроз, воспроизводимая атака и тест исправления.
Безопасность начинается с границ доверия. Внешний ввод считается недоверенным, пока не прошёл проверку по явному контракту, а ошибка проверки должна приводить к безопасному состоянию.
Перед review назовите активы, участников и границы доверия: какие данные нужно защитить, кто обращается к системе и где данные переходят из внешней среды во внутреннюю. Затем выпишите abuse cases — способы злоупотребить разрешённым интерфейсом или обойти ограничение.
Defense in depth означает несколько независимых барьеров. Для загружаемого файла это проверка прав, размера и формата, безопасное имя и каталог, изолированная обработка, лимиты ресурсов, журналирование и наблюдение за результатом. Одна проверка расширения файла не заменяет остальные.
Проверка приложения сочетает разные источники доказательств:
Ни один инструмент не закрывает все классы ошибок, поэтому набор проверок выбирают из модели угроз.
Самая частая уязвимость веб-приложений.
# Проблема: ОПАСНО: имя пользователя вставляется напрямую
name = "'; DROP TABLE users; --"
query = f"SELECT * FROM users WHERE name = '{name}'"
# Результат: DELETE всей таблицы!
# Вариант: БЕЗОПАСНО: параметризованные запросы
cursor.execute("SELECT * FROM users WHERE name = %s", (name,))
# SQLAlchemy ORM — автоматически защищает
user = session.query(User).filter(User.name == name).first()
# SQLAlchemy Core с bindparams
from sqlalchemy import text
result = conn.execute(
text("SELECT * FROM users WHERE name = :name"),
{"name": name}
)Правило: никогда не формируйте SQL через f-string или % форматирование с пользовательскими данными.
# Проблема: ОПАСНО: os.system и shell=True с пользовательским вводом
filename = "file.txt; rm -rf /"
os.system(f"cat {filename}") # выполнит rm -rf /!
# Проблема: ОПАСНО: shell=True с конкатенацией
subprocess.run(f"cat {filename}", shell=True)
# Вариант: БЕЗОПАСНО: список аргументов без shell=True
subprocess.run(["cat", filename], check=True)
# Вариант: БЕЗОПАСНО: явная проверка пути
import pathlib
safe_path = pathlib.Path(filename).resolve()
if not safe_path.is_relative_to(BASE_DIR):
raise ValueError("Path traversal attempt")
subprocess.run(["cat", str(safe_path)], check=True)# Проблема: ОПАСНО: пользователь контролирует путь
def serve_file(filename):
path = f"/var/www/files/{filename}" # filename = "../../../etc/passwd"
return open(path).read()
# Вариант: БЕЗОПАСНО: canonicalize и проверяйте базовый путь
import os
BASE_DIR = "/var/www/files"
def serve_file_safe(filename):
# realpath разрешает symlinks и ..
safe_path = os.path.realpath(os.path.join(BASE_DIR, filename))
if not safe_path.startswith(BASE_DIR + os.sep):
raise PermissionError(f"Access denied: {filename}")
return open(safe_path).read()# Проблема: ОПАСНО: запрос к URL, контролируемому пользователем
def fetch_avatar(url):
return requests.get(url).content # может быть http://169.254.169.254/
# Неполная защита: такая проверка полезна, но сама по себе не делает запрос
# безопасным.
import ipaddress
import socket
from urllib.parse import urlparse
ALLOWED_HOSTS = {"images.example.com"}
def checked_fetch(url: str) -> bytes:
parsed = urlparse(url)
if parsed.scheme != "https" or parsed.hostname not in ALLOWED_HOSTS:
raise ValueError("URL is outside the allowlist")
# Проверяем все IPv4/IPv6-адреса, а не только первый IPv4.
addresses = {
item[4][0]
for item in socket.getaddrinfo(parsed.hostname, 443, type=socket.SOCK_STREAM)
}
if not addresses or any(
not ipaddress.ip_address(address).is_global
for address in addresses
):
raise ValueError("Host resolves to a non-public address")
limit = 2_000_000
with requests.get(
url,
timeout=5,
allow_redirects=False,
stream=True,
) as response:
response.raise_for_status()
declared_size = int(response.headers.get("Content-Length", "0"))
if declared_size > limit:
raise ValueError("Response is too large")
body = bytearray()
for chunk in response.iter_content(chunk_size=64 * 1024):
body.extend(chunk)
if len(body) > limit:
raise ValueError("Response is too large")
return bytes(body)Между DNS-проверкой и соединением остаётся TOCTOU-окно: библиотека разрешает имя повторно, а адрес может измениться. Промышленная защита строится слоями: короткий allowlist назначений, запрет редиректов, лимиты размера и времени, исходящий proxy или firewall, который блокирует внутренние сети и metadata service. Не называйте одну проверку URL полной защитой от SSRF.
pickle — произвольное выполнение кода# Проблема: КРИТИЧЕСКИ ОПАСНО: pickle может выполнить любой код
import pickle
# Безвредная демонстрация: код выполняется уже при loads().
class Exploit:
def __reduce__(self):
return (print, ("Нежелательный код из pickle выполнен",))
payload = pickle.dumps(Exploit())
pickle.loads(payload) # напечатает сообщение без явного вызова print
# Вариант: БЕЗОПАСНО: никогда не загружайте pickle из недоверенных источников
# Используйте: json, msgpack, cbor2, protobuf, marshmallow
data = json.loads(untrusted_json) # только встроенные типыПравило:
pickle,marshal,shelve— только для доверенных данных между известными системами.
yaml.load() — тот же вектор# Проблема: ОПАСНО: yaml.load() с небезопасным загрузчиком
import yaml
payload = "!!python/object/apply:os.system ['ls']"
yaml.load(payload, Loader=yaml.UnsafeLoader) # может выполнить os.system!
# Вариант: БЕЗОПАСНО: safe_load разрешает только базовые типы
yaml.safe_load(payload) # yaml.constructor.ConstructorErrorFullLoader также отклоняет !!python/object/apply; выполнение произвольных
Python-тегов относится к UnsafeLoader. Для недоверенных конфигураций выбирайте
safe_load, потому что его контракт уже базовых Python-типов и проще для
аудита.
# ElementTree не раскрывает внешние сущности и при ссылке на неизвестную
# внешнюю сущность поднимет ParseError. Но недоверенный XML всё ещё может быть
# источником атак на ресурсы, например exponential entity expansion.
from xml.etree import ElementTree
tree = ElementTree.parse(untrusted_xml_file)
# Для недоверенного XML используйте библиотеку с явными защитными лимитами.
import defusedxml.ElementTree as ET
tree = ET.parse(untrusted_xml_file)
# Вариант: ИЛИ: lxml с resolve_entities=False
from lxml import etree
parser = etree.XMLParser(resolve_entities=False, no_network=True)
tree = etree.parse(untrusted_xml_file, parser)eval() и exec()# Проблема: eval исполняет выражение с полномочиями процесса.
user_input = "__import__('os').getcwd()"
eval(user_input) # получает доступ к окружению процесса
# Проблема: Ограниченный eval НЕ работает как sandbox
eval(user_input, {"__builtins__": {}})
# Атакующий всё равно может добраться до os через объекты
# Ограниченный учебный вычислитель. Для сложной грамматики лучше отдельный
# parser; даже разрешённые операции должны иметь лимиты ресурсов.
import ast
import math
import operator
OPERATORS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
}
def safe_eval(expr: str) -> float:
if len(expr) > 100:
raise ValueError("Expression is too long")
tree = ast.parse(expr, mode="eval")
def evaluate(node: ast.AST) -> int | float:
if isinstance(node, ast.Expression):
return evaluate(node.body)
if isinstance(node, ast.Constant):
if (
isinstance(node.value, bool)
or not isinstance(node.value, (int, float))
or abs(node.value) > 1_000_000_000_000
):
raise ValueError("Invalid number")
return node.value
if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
return -evaluate(node.operand)
if isinstance(node, ast.BinOp) and type(node.op) in OPERATORS:
result = OPERATORS[type(node.op)](
evaluate(node.left),
evaluate(node.right),
)
if not math.isfinite(result) or abs(result) > 1e18:
raise ValueError("Result is out of bounds")
return result
raise ValueError(f"Disallowed expression: {type(node).__name__}")
return float(evaluate(tree))
assert safe_eval("2 + 3 * 4") == 14.0
# safe_eval("__import__('os')") # ValueError: Disallowed expression: Callsecrets — криптографически безопасные случайные числаimport secrets
# Проблема: random — НЕ подходит для криптографии!
import random
token = f"{random.getrandbits(256):064x}" # предсказуем для токена!
# Вариант: secrets — CSPRNG (Cryptographically Secure Pseudo-Random Number Generator)
token = secrets.token_hex(32) # 64-символьный hex
token = secrets.token_urlsafe(32) # URL-safe base64
token = secrets.token_bytes(32) # 32 случайных байта
otp = secrets.randbelow(1_000_000) # случайное 6-значное число
# Генерация паролей
import string
alphabet = string.ascii_letters + string.digits + string.punctuation
password = ''.join(secrets.choice(alphabet) for _ in range(16))# Проблема: Хардкод в коде
DB_PASSWORD = "super_secret_123"
# Проблема: В логах
logger.info(f"Connecting with password={password}")
# Проблема: В git-истории (даже если удалили — история сохраняется)
# Вариант: Переменные окружения
import os
DB_PASSWORD = os.environ.get("DB_PASSWORD")
# Вариант: .env файлы (только локально, в .gitignore)
from dotenv import load_dotenv
load_dotenv()
DB_PASSWORD = os.environ["DB_PASSWORD"]
# Вариант: Для рабочей среды: Vault, AWS Secrets Manager, K8s Secrets
import boto3
def get_secret(name):
client = boto3.client('secretsmanager')
return client.get_secret_value(SecretId=name)['SecretString']# Проблема: НИКОГДА не храните пароли в открытом виде или в MD5/SHA1
import hashlib
hashed = hashlib.md5(password.encode()).hexdigest() # взламывается за секунды!
hashed = hashlib.sha256(password.encode()).hexdigest() # тоже плохо — rainbow tables
# Проблема: SHA-256 без соли — уязвим к атакам по словарю
# Если два пользователя имеют одинаковый пароль — одинаковый хеш!
# Вариант: bcrypt — медленный адаптивный хэш
import bcrypt
def hash_password(password: str) -> bytes:
salt = bcrypt.gensalt(rounds=12) # rounds: стоимость работы
return bcrypt.hashpw(password.encode(), salt)
def verify_password(password: str, hashed: bytes) -> bool:
return bcrypt.checkpw(password.encode(), hashed)
# Вариант: argon2 — победитель Password Hashing Competition 2015
from argon2 import PasswordHasher
ph = PasswordHasher(time_cost=3, memory_cost=65536, parallelism=4)
hashed = ph.hash(password)
ph.verify(hashed, password) # True или argon2.exceptions.VerifyMismatchError
# Вариант: Встроенный hashlib с PBKDF2 (если нет bcrypt/argon2)
import hashlib, os
salt = os.urandom(32)
key = hashlib.pbkdf2_hmac('sha256', password.encode(), salt, 100_000)# Проблема: УЯЗВИМО к timing attack — Python прерывает сравнение при первом несовпадении
if user_token == expected_token: # атакующий может угадать байт за байтом!
authenticate()
# Вариант: Constant-time сравнение — hmac.compare_digest
import hmac
if hmac.compare_digest(user_token.encode(), expected_token.encode()):
authenticate()
# Принцип: функция всегда занимает одинаковое время независимо от совпадения
# Разница в 1 нс между "нет" и "почти да" позволяет угадать секрет за O(n) попытокimport ssl
import requests
# Проблема: ОПАСНО: отключение проверки сертификата
requests.get(url, verify=False) # уязвимо к MITM-атакам
# Вариант: БЕЗОПАСНО: проверка включена по умолчанию
requests.get(url)
# Вариант: Для самоподписанных сертификатов
requests.get(url, verify="/path/to/ca-cert.pem")
# Настройка SSL-контекста вручную
ctx = ssl.create_default_context()
ctx.minimum_version = ssl.TLSVersion.TLSv1_2 # запрет устаревших версий
ctx.verify_mode = ssl.CERT_REQUIRED
ctx.check_hostname = True
conn = ctx.wrap_socket(socket.socket(), server_hostname=hostname)import re
# Проблема: ОПАСНО: уязвимый паттерн — exponential backtracking
pattern = re.compile(r"(a+)+$")
pattern.match("a" * 30 + "!") # зависает! (экспоненциальное время)
# Проблема: (a+)+ — вложенные квантификаторы создают катастрофический backtracking
# Вариант: Лучше убрать неоднозначность из самого шаблона
safe_pattern = re.compile(r"a+$")
# Вариант: Ограничивайте длину ввода перед проверкой
def validate_email(email: str) -> bool:
if len(email) > 254: # RFC 5321 limit
return False
return bool(re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email))
# У stdlib re нет параметра timeout. Если шаблон задаёт пользователь,
# выполняйте проверку в изолированном процессе с лимитами ресурсов
# либо выбирайте движок, который явно гарантирует ограничение времени.bandit — статический анализ# Установка
# pip install bandit
# Запуск
# bandit -r my_project/ -ll
# Найдёт: hardcoded passwords, использование assert для security,
# subprocess shell=True, eval/exec, md5/sha1 хэширование, и др.pip-audit и safety — уязвимости в зависимостях# pip-audit — проверяет PyPI Advisory Database
pip install pip-audit
pip-audit
# safety — проверяет safety-db
pip install safety
safety checksemgrep — гибкий SAST# Поиск опасных паттернов с готовыми правилами
semgrep --config=p/python-security .from pydantic import BaseModel, EmailStr, Field, ValidationError, field_validator
class UserCreate(BaseModel):
email: EmailStr # встроенная валидация email
name: str = Field(min_length=1, max_length=100)
age: int = Field(ge=0, le=150)
bio: str = Field(default="", max_length=1000)
@field_validator("name")
@classmethod
def name_no_special(cls, v: str) -> str:
if not v.replace(' ', '').isalnum():
raise ValueError('Имя должно содержать только буквы и цифры')
return v.strip()
# Ошибки полей снаружи модели объединяются в ValidationError
try:
user = UserCreate(email="invalid", name="", age=-1)
except ValidationError as e:
print(e.json())import html
import bleach
# Экранирование HTML — для вставки пользовательских данных в HTML
safe_text = html.escape("<script>alert('xss')</script>")
# "<script>alert('xss')</script>"
# bleach — разрешает только безопасные теги
clean = bleach.clean(
user_content,
tags=['p', 'b', 'i', 'a'],
attributes={'a': ['href']},
strip=True
)CSRF использует тот факт, что браузер автоматически прикладывает cookie к
запросу на знакомый origin. Для операций, меняющих состояние, сервер должен
проверять CSRF-токен и допустимый Origin; SameSite у cookie добавляет ещё
один слой защиты. Referer может быть запасным сигналом, но не единственным
механизмом. API с токеном в явном Authorization-заголовке имеет другую модель
угроз, потому что сторонняя страница не может автоматически подставить такой
секрет.
Защита от CSRF не заменяет проверку прав: даже корректный запрос от пользователя должен быть авторизован для конкретного объекта.
Актуальная редакция списка — OWASP Top 10:2025. Это категории рисков, а не десять независимых рецептов для конкретного фреймворка.
| Категория OWASP 2025 | Пример в Python-проекте | Базовая защита |
|---|---|---|
| A01 Broken Access Control | объект загружается только по id из URL | проверять право на каждый объект на сервере |
| A02 Security Misconfiguration | режим отладки и лишние заголовки в рабочей среде | безопасные настройки по умолчанию и проверка конфигурации |
| A03 Software Supply Chain Failures | непроверенная или подменённая зависимость | lock-файл, аудит и доверенная сборочная цепочка |
| A04 Cryptographic Failures | собственный алгоритм или слабое хранение паролей | стандартные протоколы и Argon2id/bcrypt для паролей |
| A05 Injection | SQL или shell собирается из пользовательской строки | параметры SQL и аргументы subprocess списком |
| A06 Insecure Design | нет лимитов, threat model и защиты от abuse | моделирование угроз и защитные ограничения до реализации |
| A07 Authentication Failures | слабое восстановление пароля или сессии | MFA, ротация сессий и защищённый reset-flow |
| A08 Software or Data Integrity Failures | доверие к неподписанному артефакту или pickle | проверка происхождения; не читать недоверенный pickle |
| A09 Security Logging and Alerting Failures | атака не оставляет полезного события | структурированные логи, алерты и защита журналов |
| A10 Mishandling of Exceptional Conditions | fail-open после таймаута или ошибки проверки | явные безопасные состояния и тесты отказов |
import jwt
from datetime import datetime, timedelta, timezone
SECRET_KEY = os.environ["JWT_SECRET"] # минимум 256 бит
ALGORITHM = "HS256"
# Создание токена
def create_access_token(user_id: int) -> str:
payload = {
"sub": str(user_id),
"iat": datetime.now(timezone.utc),
"exp": datetime.now(timezone.utc) + timedelta(hours=1),
"type": "access",
}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
# Валидация токена
def decode_token(token: str) -> dict:
try:
payload = jwt.decode(
token,
SECRET_KEY,
algorithms=[ALGORITHM],
options={"verify_exp": True},
)
return payload
except jwt.ExpiredSignatureError:
raise AuthenticationError("Token expired")
except jwt.InvalidTokenError:
raise AuthenticationError("Invalid token")
# Проблема: НЕ ДЕЛАЙТЕ: decode без верификации
jwt.decode(token, options={"verify_signature": False}) # уязвимо!
# Проблема: Не храните чувствительные данные в payload (payload не зашифрован, только подписан!)import time
from collections import defaultdict
from threading import Lock
class InMemoryRateLimiter:
"""Sliding window rate limiter (только для одного процесса)."""
def __init__(self, max_requests: int, window_seconds: int):
self._max = max_requests
self._window = window_seconds
self._requests: dict[str, list[float]] = defaultdict(list)
self._lock = Lock()
def is_allowed(self, key: str) -> bool:
now = time.time()
cutoff = now - self._window
with self._lock:
# Удаляем старые запросы
self._requests[key] = [t for t in self._requests[key] if t > cutoff]
if len(self._requests[key]) >= self._max:
return False
self._requests[key].append(now)
return True
# Для рабочей среды используйте Redis:
def redis_rate_limit(redis_client, key: str, limit: int, window: int) -> bool:
"""Token bucket через Redis INCR + EXPIRE."""
current = redis_client.incr(key)
if current == 1:
redis_client.expire(key, window)
return current <= limitfrom pydantic_settings import BaseSettings
from pydantic import SecretStr, validator
class Settings(BaseSettings):
"""Настройки приложения с валидацией безопасности."""
database_url: SecretStr # SecretStr скрывает значение в логах
jwt_secret: SecretStr
allowed_hosts: list[str] = ["localhost"]
debug: bool = False
@validator("jwt_secret")
def validate_jwt_secret(cls, v):
secret = v.get_secret_value()
if len(secret) < 32:
raise ValueError("JWT secret must be at least 32 characters")
return v
class Config:
env_file = ".env"
settings = Settings()
# SecretStr предотвращает случайный вывод:
print(settings.jwt_secret) # '**********'
print(settings.jwt_secret.get_secret_value()) # только явноБезопасность начинается с границы доверия и безопасного значения по умолчанию. Результат основы — проверка входа до опасной операции и тест, воспроизводящий первоначальную уязвимость.
import tempfile
from pathlib import Path
def confined_path(root: Path, user_value: str) -> Path:
resolved_root = root.resolve()
candidate = (resolved_root / user_value).resolve()
if not candidate.is_relative_to(resolved_root):
raise ValueError("путь выходит за разрешённый каталог")
return candidate
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
assert confined_path(root, "reports/day.json").is_relative_to(root.resolve())
try:
confined_path(root, "../secret.txt")
except ValueError:
pass
else:
raise AssertionError("обход каталога должен быть запрещён")Практика с подсказками: составьте модель угроз для загрузки пользовательского файла. Ограничьте размер и формат до разбора, исключите небезопасную десериализацию, уберите содержимое из журналов и добавьте проверку обхода пути.
Проведите разбор цепочки поставки и поведения при компрометации зависимости. Опишите активы, нарушителя, сценарии злоупотребления, остаточный риск и сигнал для отката; один список категорий OWASP без связи со сценарием не считается.
safe_eval(expr), разрешающий только числовые выражения через AST-проверку.secrets.bandit -r . и исправьте найденные проблемы.hmac.compare_digest() и == при сравнении токенов.safe_open(filename, base_dir) защищённую от path traversal.collections.deque. Затем объясните, почему для нескольких процессов
требуется общее хранилище и атомарная операция, например Lua-скрипт в Redis.Settings, которая проверяет длину SECRET_KEY и запрещает DEBUG=True в рабочей среде.secrets, сохраняет ведущие нули и проверяется по формату,
а не тестом на «случайность» одного значения.# nosec содержит конкретную
модель угроз, а не только подавляет предупреждение.compare_digest применяется к значениям одинакового типа и длины; он снижает
утечку по времени сравнения, но не заменяет хэширование или аутентификацию.safe_open проверяется на .., абсолютном пути, символической ссылке и
соседнем каталоге с общим строковым префиксом.Закрепите тему в лаборатории «Безопасное и атомарное извлечение ZIP». Вы построите модель угроз для архива, воспроизведёте обход каталога и архивную бомбу, а затем докажете отсутствие частичного результата после отказа.
git clone --branch v0.6.2 --depth 1 https://gitlab.potapov.me/courses/python-labs.git
cd python-labs
uv sync --group test
uv run --group test pytest python_security/testsДалее: Профилирование и оптимизация