• Содержание
  • Зачем прямая загрузка и чем она лучше проксирования через бэкенд
  • Архитектура: подпись на загрузку, события после загрузки, жизненный цикл файла
  • Безопасность: политика бакета, CORS, антивирус, контроль типов и размеров
  • Пример кода: FastAPI + S3‑совместимое хранилище (presigned POST)
  • Клиентский код: форма, прогресс, обработка ошибок
  • Выдача файлов: подписанные ссылки, заголовки, CDN
  • Экономия: классы хранения, срок жизни, дедупликация
  • Учёт в продукте: состояния файла и интеграция в модель данных
  • Тестирование: проверка сценариев и инструменты
  • Частые ошибки и как их избежать
  • Чеклист внедрения

Зачем прямая загрузка и чем она лучше проксирования через бэкенд

Если файлы идут через ваш бэкенд, серверу приходится держать открытые соединения, считать контрольные суммы, перекладывать потоки и платить за исходящий трафик дважды: из браузера в бэкенд и затем из бэкенда в хранилище. При росте аудитории это превращается в узкое место и в счёт за инфраструктуру.

Прямая загрузка решает проблему: браузер отправляет файл сразу в объектное хранилище (S3‑совместимое: AWS S3, Yandex Object Storage, MinIO и т. п.) по «подписанным» данным. Бэкенд только:

• выдаёт короткоживущую подпись с ограничениями (тип, размер, ключ);
• принимает событие после загрузки (или периодически проверяет) и привязывает файл к сущности в продукте;
• запускает антивирус и обработку (например, конвертацию изображений).

Итог для бизнеса: быстрее отклик, меньше расходы на сервер и сеть, простой масштабируемый конвейер обработки.

Архитектура: подпись на загрузку, события после загрузки, жизненный цикл файла

Типовой поток:

1. Клиент запрашивает у вашего API «разрешение на загрузку» для будущего файла (контент‑тип, предполагаемый размер, имя для скачивания).
2. API генерирует короткоживущую подпись на загрузку (presigned POST или PUT), фиксирует ожидаемый файл в таблице uploads со статусом pending и возвращает клиенту url + поля формы.
3. Клиент загружает файл напрямую в бакет «uploads».
4. Хранилище отправляет событие (S3 Event) в очередь (SQS/версии для других провайдеров). Работник:
   • проверяет ограничения (размер, тип, расширение, контрольную сумму);
   • сканирует на вирусы; при угрозе — переносит в «quarantine» и помечает запись как blocked;
   • при успехе — переносит в «clean» (или ставит тег clean=true), обновляет статус ready.
5. Для скачивания/просмотра приложение выдаёт короткоживущую подписанную ссылку на «clean» или отдаёт через CDN.

Жизненный цикл файла в базе может быть таким: pending → uploading → scanning → processing → ready (или blocked при угрозе/нарушении правил).

Безопасность: политика бакета, CORS, антивирус, контроль типов и размеров

Основные опоры безопасности:

• Закрытый бакет. Включите «Block Public Access» и запретите анонимный доступ в политике.
• Подпись живёт минуты, а не часы. Не выдавайте длительных прав на загрузку.
• Жёсткие ограничения в подписи: фиксированный ключ, допустимые типы, предел размера.
• Антивирус. Сканируйте всё, что вы допускаете к скачиванию пользователями.
• Разделение бакетов/префиксов: uploads (сырьё), clean (прошедшие проверку), quarantine (заражённые).
• CORS: разрешайте только свои домены и необходимые методы HEAD/POST/PUT/GET.
• Заголовки защиты при скачивании: Content-Disposition, Content-Type, X-Content-Type-Options: nosniff.
• Верификация после загрузки: сверка метаданных и фактического размера по HeadObject.

Пример политики бакета (запрет публичного доступа и требования TLS):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DenyUnEncryptedInTransit",
      "Effect": "Deny",
      "Principal": "*",
      "Action": "s3:*",
      "Resource": [
        "arn:aws:s3:::my-company-uploads",
        "arn:aws:s3:::my-company-uploads/*"
      ],
      "Condition": { "Bool": { "aws:SecureTransport": "false" } }
    }
  ]
}

CORS для браузера (минимально необходимое):

[
  {
    "AllowedOrigins": ["https://app.example.com"],
    "AllowedMethods": ["POST", "PUT", "GET", "HEAD"],
    "AllowedHeaders": ["*"],
    "ExposeHeaders": ["ETag"],
    "MaxAgeSeconds": 3000
  }
]

Пример кода: FastAPI + S3‑совместимое хранилище (presigned POST)

Ниже — минимальный сервер на Python/FastAPI, который выдаёт подпись на загрузку в бакет uploads с ограничениями и переводит запись в БД из pending в ready после проверки фактического размера и типа. Для простоты храним «БД» в памяти и не реализуем очереди; в проде проверьте события из S3 через SQS, добавьте антивирус и перенос между бакетами.

# requirements:
#   fastapi==0.110.0
#   uvicorn[standard]==0.27.1
#   boto3==1.34.50
# Запуск: UVICORN_CMD='uvicorn app:app --reload'

import os
import uuid
import mimetypes
from datetime import datetime, timedelta
from typing import Literal, Optional

import boto3
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, constr

AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
UPLOAD_BUCKET = os.getenv("UPLOAD_BUCKET", "my-company-uploads")
CLEAN_BUCKET = os.getenv("CLEAN_BUCKET", "my-company-clean")
MAX_SIZE_BYTES = int(os.getenv("MAX_SIZE_BYTES", str(20 * 1024 * 1024)))  # 20 МБ

s3 = boto3.client("s3", region_name=AWS_REGION)

app = FastAPI(title="Direct Upload Demo")

# Простейшее хранилище метаданных (в проде — БД)
UPLOADS = {}

class PresignRequest(BaseModel):
    content_type: constr(strip_whitespace=True, min_length=3) = Field(..., examples=["image/png", "application/pdf"])
    size_bytes: int = Field(..., gt=0, le=MAX_SIZE_BYTES)
    filename: constr(strip_whitespace=True, min_length=1, max_length=200)
    tenant_id: constr(strip_whitespace=True, min_length=1)

class PresignResponse(BaseModel):
    url: str
    fields: dict
    key: str
    expires_at: datetime
    upload_id: str

class FinalizeRequest(BaseModel):
    upload_id: str

ALLOWED_TYPES = {
    "image/png",
    "image/jpeg",
    "application/pdf",
}


def safe_key(tenant_id: str, filename: str) -> str:
    # Генерируем безопасный ключ: <tenant>/<uuid>/<basename>
    base = os.path.basename(filename)
    # Подрезаем опасные символы
    base = base.replace("\\", "_").replace("/", "_")
    uid = str(uuid.uuid4())
    return f"{tenant_id}/{uid}/{base}"


@app.post("/uploads/presign", response_model=PresignResponse)
def create_presigned_post(req: PresignRequest):
    if req.content_type not in ALLOWED_TYPES:
        raise HTTPException(400, detail="Недопустимый тип файла")
    if req.size_bytes > MAX_SIZE_BYTES:
        raise HTTPException(400, detail="Слишком большой файл")

    key = safe_key(req.tenant_id, req.filename)
    # Дополнительные поля: шифрование на стороне сервера, сохранение метаданных
    fields = {
        "Content-Type": req.content_type,
        "x-amz-server-side-encryption": "AES256",
        "x-amz-meta-tenant": req.tenant_id,
        "x-amz-meta-filename": req.filename,
    }
    conditions = [
        {"bucket": UPLOAD_BUCKET},
        {"key": key},
        {"Content-Type": req.content_type},
        {"x-amz-server-side-encryption": "AES256"},
        ["content-length-range", 1, MAX_SIZE_BYTES],
    ]

    presigned = s3.generate_presigned_post(
        Bucket=UPLOAD_BUCKET,
        Key=key,
        Fields=fields,
        Conditions=conditions,
        ExpiresIn=300,  # 5 минут
    )

    upload_id = str(uuid.uuid4())
    UPLOADS[upload_id] = {
        "key": key,
        "bucket": UPLOAD_BUCKET,
        "tenant_id": req.tenant_id,
        "filename": req.filename,
        "content_type": req.content_type,
        "expected_size": req.size_bytes,
        "status": "pending",
        "created_at": datetime.utcnow().isoformat(),
    }

    return PresignResponse(
        url=presigned["url"],
        fields=presigned["fields"],
        key=key,
        expires_at=datetime.utcnow() + timedelta(seconds=300),
        upload_id=upload_id,
    )


@app.post("/uploads/finalize")
def finalize(req: FinalizeRequest):
    rec = UPLOADS.get(req.upload_id)
    if not rec:
        raise HTTPException(404, detail="Загрузка не найдена")

    # Проверяем, что объект действительно загружен и совпадает по типу/размеру
    try:
        head = s3.head_object(Bucket=rec["bucket"], Key=rec["key"])  # type: ignore
    except s3.exceptions.NoSuchKey:  # type: ignore
        raise HTTPException(400, detail="Файл ещё не загружен")

    actual_size = head["ContentLength"]
    actual_type = head.get("ContentType")

    if actual_size <= 0 or actual_size > rec["expected_size"] or actual_type != rec["content_type"]:
        rec["status"] = "blocked"
        raise HTTPException(400, detail="Несовпадение параметров файла")

    # Здесь можно отправить событие в очередь на антивирус и обработку
    # Для демо — считаем файл готовым и копируем в чистый бакет
    copy_source = {"Bucket": rec["bucket"], "Key": rec["key"]}
    s3.copy_object(
        Bucket=CLEAN_BUCKET,
        Key=rec["key"],
        CopySource=copy_source,
        MetadataDirective="COPY",
        ServerSideEncryption="AES256",
        ContentType=rec["content_type"],
    )

    rec["status"] = "ready"
    rec["bucket"] = CLEAN_BUCKET

    # Подписанная ссылка на скачивание (60 секунд)
    url = s3.generate_presigned_url(
        "get_object",
        Params={
            "Bucket": CLEAN_BUCKET,
            "Key": rec["key"],
            "ResponseContentType": rec["content_type"],
            "ResponseContentDisposition": f"attachment; filename=\"{rec['filename']}\"",
        },
        ExpiresIn=60,
    )
    return {"status": rec["status"], "download_url": url}

Пояснения:

• generate_presigned_post добавляет в политику условия, которые проверит само хранилище.
• Мы дополнительно дублируем валидацию на стороне приложения при finalize, чтобы не полагаться на клиентские поля формы.
• В проде копирование из uploads в clean делает фоновый обработчик после антивируса; здесь — упрощённо.

Клиентский код: форма, прогресс, обработка ошибок

Пример на чистом JavaScript — получаем подпись, отправляем файл как FormData с полями, которые вернул сервер. Прогресс можно отслеживать через XMLHttpRequest.

async function uploadFile(file) {
  // 1) Запрашиваем подпись
  const presignResp = await fetch('/api/uploads/presign', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      content_type: file.type,
      size_bytes: file.size,
      filename: file.name,
      tenant_id: 'tenant-123'
    })
  }).then(r => r.json());

  // 2) Готовим форму для прямой загрузки
  const formData = new FormData();
  Object.entries(presignResp.fields).forEach(([k, v]) => formData.append(k, v));
  formData.append('file', file);

  // 3) Отправляем файл напрямую в S3-совместимое хранилище
  const xhr = new XMLHttpRequest();
  const uploadPromise = new Promise((resolve, reject) => {
    xhr.upload.addEventListener('progress', (e) => {
      if (e.lengthComputable) {
        const pct = Math.round((e.loaded / e.total) * 100);
        console.log('progress', pct);
      }
    });
    xhr.onreadystatechange = () => {
      if (xhr.readyState === 4) {
        if (xhr.status === 204 || xhr.status === 201) resolve();
        else reject(new Error('Upload failed: ' + xhr.status));
      }
    };
    xhr.open('POST', presignResp.url, true);
    xhr.send(formData);
  });

  await uploadPromise;

  // 4) Сообщаем серверу, что загрузка завершена (он сверит параметры и вернёт ссылку)
  const finalize = await fetch('/api/uploads/finalize', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ upload_id: presignResp.upload_id })
  }).then(r => r.json());

  console.log('Download URL:', finalize.download_url);
  return finalize;
}

Замечания:

• Если хранилище отвечает 201/204 — файл принят. Но всё равно вызываем finalize, чтобы приложение убедилось в корректности и активировало дальнейшие шаги.
• В случае ошибки по размеру/типу сервер корректно отклонит финализацию.

Выдача файлов: подписанные ссылки, заголовки, CDN

Лучший способ отдачи закрытого контента — короткоживущая подписанная ссылка:

• не держит ваше приложение в тракте трафика;
• ограничивает время доступа и параметры ответа (имя файла, тип);
• легко кэшируется CDN без раскрытия бакета наружу (например, по приватному origin с подписями CDN).

Рекомендации:

• Для многих мелких файлов (иконки, публичные превью) используйте CDN с публичным кэшем и строгими заголовками кэширования.
• Для приватных загрузок — подписанные URL на несколько десятков секунд.
• Добавляйте ResponseContentDisposition=attachment; filename="..." для безопасного скачивания.
• Ставьте X-Content-Type-Options: nosniff при проксировании через свой сервер.

Экономия: классы хранения, срок жизни, дедупликация

• Переход классов хранения: переводите «хвосты» (давние файлы) в более дешёвые классы (Infrequent Access/Cold/Archive) по политике Lifecycle.
• Срок жизни: удаляйте сырьё из uploads через 1–7 дней после успешной обработки, quarantine — по внутренним SLA.
• Сжатие: для текстовых форматов (CSV, JSON, логов) храните и/или отдавайте сжатые варианты.
• Дедупликация: сохраняйте контрольную сумму файла (например, SHA‑256) в метаданных и не дублируйте одинаковые файлы между сущностями — сэкономите место.
• Избегайте двойного трафика: не проксируйте скачивание через приложение, если контент закрытый — используйте подписанные URL или CDN со своей подписью.

Учёт в продукте: состояния файла и интеграция в модель данных

Сущность Upload обычно содержит:

• id, tenant_id, user_id;
• оригинальное имя, конечный ключ в clean, тип, размер;
• статус: pending → uploading → scanning → processing → ready | blocked;
• причину блокировки (вирус, нарушение размера/типа, ошибка конвертации);
• контрольные суммы (sha256), даты создания/обновления;
• связь с бизнес‑сущностью (например, document_id).

Такой подход упрощает аудит, повторную обработку и очистку хвостов. При ошибке антивируса можно повторно отправить в обработку, не тревожа пользователя.

Тестирование: проверка сценариев и инструменты

• Локальные интеграционные тесты с LocalStack/MinIO.
• Негативные сценарии: превышение размера, неверный тип, отмена загрузки, порванная сеть, истёкшая подпись.
• Большие файлы и много мелких одновременно — проверка параллелизма и лимитов браузера.
• Прогрев CDN и проверка заголовков кэширования/безопасности.
• Нагрузочные тесты на выдачу подписанных ссылок (это дешёвый, но частый путь).

Частые ошибки и как их избежать

• Долгая подпись (часами) — украдут и будут заливать мусор. Делайте минуты.
• Публичный бакет — утечка мгновенно. Отключайте публичный доступ целиком.
• Доверие client‑side полям: размер/тип должен проверяться на стороне сервера и/или хранилища.
• Отсутствие антивируса — риск вредоносных вложений.
• Смешивание сырья и готового контента в одном месте — сложнее политики и очистка.
• Неправильный CORS — «Request has been blocked by CORS policy». Разрешайте только свой origin и нужные методы.
• Неуказанный Content-Disposition — браузер может попытаться отрендерить бинарь прямо в окне.

Чеклист внедрения

• Закрытый бакет, запрет незащищённого транспорта
• CORS на ваш домен и нужные методы
• Подписанная загрузка с жёсткими условиями и коротким TTL
• Таблица Upload с состояниями и метаданными
• События после загрузки → очередь → антивирус → перенос в clean
• Выдача подписанными ссылками/через CDN, корректные заголовки
• Политики жизненного цикла: очистка сырья, перевод в «холод»
• Тесты на ошибки сети, размеры, истечение подписей

Бонус: пример рабочего обработчика антивируса (контейнер/ВМ)

Ниже — упрощённый рабочий скрипт: читает события об объектах из очереди (SQS), скачивает файл, запускает clamscan, по результату переносит в clean или quarantine. В проде упакуйте вместе с ClamAV в контейнер, настройте регулярное обновление баз сигнатур.

# requirements:
#   boto3==1.34.50
#   mypy-boto3-sqs (необязательно, для типов)
# Требуется установленный clamscan в окружении контейнера.

import json
import os
import shutil
import subprocess
import tempfile
from pathlib import Path

import boto3

AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
SQS_URL = os.getenv("SQS_URL")
UPLOAD_BUCKET = os.getenv("UPLOAD_BUCKET", "my-company-uploads")
CLEAN_BUCKET = os.getenv("CLEAN_BUCKET", "my-company-clean")
QUARANTINE_BUCKET = os.getenv("QUARANTINE_BUCKET", "my-company-quarantine")

s3 = boto3.client("s3", region_name=AWS_REGION)
sqs = boto3.client("sqs", region_name=AWS_REGION)


def process_record(bucket: str, key: str) -> None:
    with tempfile.TemporaryDirectory() as td:
        local = Path(td) / "obj.bin"
        s3.download_file(bucket, key, str(local))
        # Запуск ClamAV
        res = subprocess.run(["clamscan", "--stdout", "--no-summary", str(local)], capture_output=True, text=True)
        infected = res.returncode == 1  # 1 — найден вирус, 0 — чисто, >1 — ошибка
        if res.returncode > 1:
            raise RuntimeError(f"ClamAV error: {res.stderr}")

        dest_bucket = QUARANTINE_BUCKET if infected else CLEAN_BUCKET
        s3.copy_object(
            Bucket=dest_bucket,
            Key=key,
            CopySource={"Bucket": bucket, "Key": key},
            ServerSideEncryption="AES256",
            MetadataDirective="COPY",
        )
        # Удаляем сырьё после успешного копирования
        s3.delete_object(Bucket=bucket, Key=key)
        print(f"Moved {key} -> {dest_bucket} (infected={infected})")


def main():
    while True:
        resp = sqs.receive_message(QueueUrl=SQS_URL, MaxNumberOfMessages=10, WaitTimeSeconds=20)
        msgs = resp.get("Messages", [])
        if not msgs:
            continue
        for m in msgs:
            try:
                body = json.loads(m["Body"])  # S3->SQS сообщение в формате EventBridge/S3
                # Нормализуем: находим записи S3
                records = body.get("Records", [])
                for r in records:
                    b = r["s3"]["bucket"]["name"]
                    k = r["s3"]["object"]["key"]
                    process_record(b, k)
                sqs.delete_message(QueueUrl=SQS_URL, ReceiptHandle=m["ReceiptHandle"])  # ack
            except Exception as e:
                print("Error processing message:", e)
                # Сообщение уедет в DLQ по политике повторов

if __name__ == "__main__":
    main()

Итог: простой, надёжный и экономичный конвейер загрузки и выдачи файлов, который растёт вместе с вашим продуктом, не превращая бэкенд в «трубу» для гигабайт трафика и не жертвуя безопасностью.