InviteMe Developer
122 lines
5.0 KiB
Python
122 lines
5.0 KiB
Python
import io
|
|
import os
|
|
import re
|
|
import secrets
|
|
import unicodedata
|
|
import warnings
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from PIL import Image, ImageOps
|
|
|
|
|
|
ALLOWED_IMAGE_EXTENSIONS = {"jpg", "jpeg", "png", "webp"}
|
|
_FORMAT_EXTENSIONS = {
|
|
"JPEG": {"jpg", "jpeg"},
|
|
"PNG": {"png"},
|
|
"WEBP": {"webp"},
|
|
}
|
|
|
|
|
|
def validate_password(password, min_length=12):
|
|
"""Return a user-facing error or None when a password is acceptable."""
|
|
if len(password) < min_length:
|
|
return f"Пароль должен содержать не менее {min_length} символов"
|
|
|
|
groups = sum([
|
|
any(c.islower() for c in password),
|
|
any(c.isupper() for c in password),
|
|
any(c.isdigit() for c in password),
|
|
any(not c.isalnum() for c in password),
|
|
])
|
|
if groups < 3:
|
|
return "Используйте минимум три группы: строчные, заглавные, цифры и спецсимволы"
|
|
return None
|
|
|
|
|
|
def safe_upload_filename(filename, max_length=160):
|
|
"""Keep Unicode names while removing path and control characters."""
|
|
if not filename:
|
|
raise ValueError("Пустое имя файла")
|
|
|
|
normalized = unicodedata.normalize("NFKC", str(filename)).replace("\\", "/")
|
|
name = normalized.rsplit("/", 1)[-1].strip().strip(".")
|
|
name = "".join(ch for ch in name if unicodedata.category(ch)[0] != "C")
|
|
name = re.sub(r"[^\w.()\- ]", "_", name, flags=re.UNICODE)
|
|
name = re.sub(r"\s+", " ", name).strip()
|
|
if not name or name in {".", ".."}:
|
|
raise ValueError("Небезопасное имя файла")
|
|
|
|
suffix = Path(name).suffix
|
|
stem_limit = max_length - len(suffix)
|
|
name = f"{Path(name).stem[:stem_limit]}{suffix}"
|
|
return name
|
|
|
|
|
|
def image_extension(filename):
|
|
return Path(filename).suffix.lower().lstrip(".")
|
|
|
|
|
|
def validate_zip_size(stream, max_files, max_uncompressed_bytes):
|
|
"""Reject compressed containers that could expand beyond safe limits."""
|
|
position = stream.tell()
|
|
try:
|
|
with zipfile.ZipFile(stream, "r") as archive:
|
|
members = [member for member in archive.infolist() if not member.is_dir()]
|
|
if len(members) > max_files:
|
|
raise ValueError("В архиве слишком много файлов")
|
|
if sum(member.file_size for member in members) > max_uncompressed_bytes:
|
|
raise ValueError("Архив слишком большой после распаковки")
|
|
except zipfile.BadZipFile as exc:
|
|
raise ValueError("Некорректный ZIP-контейнер") from exc
|
|
finally:
|
|
stream.seek(position)
|
|
|
|
|
|
def save_validated_image(stream, destination, max_bytes, max_pixels):
|
|
"""Validate and re-encode a single raster image, stripping metadata."""
|
|
filename = Path(destination).name
|
|
extension = image_extension(filename)
|
|
if extension not in ALLOWED_IMAGE_EXTENSIONS:
|
|
raise ValueError("Разрешены только JPG, PNG и WebP")
|
|
|
|
data = stream.read(max_bytes + 1)
|
|
if len(data) > max_bytes:
|
|
raise ValueError("Файл изображения слишком большой")
|
|
if not data:
|
|
raise ValueError("Файл изображения пуст")
|
|
|
|
Image.MAX_IMAGE_PIXELS = max_pixels
|
|
try:
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("error", Image.DecompressionBombWarning)
|
|
with Image.open(io.BytesIO(data)) as source:
|
|
image_format = (source.format or "").upper()
|
|
if image_format not in _FORMAT_EXTENSIONS:
|
|
raise ValueError("Неподдерживаемый формат изображения")
|
|
if extension not in _FORMAT_EXTENSIONS[image_format]:
|
|
raise ValueError("Расширение файла не соответствует его содержимому")
|
|
if getattr(source, "n_frames", 1) != 1:
|
|
raise ValueError("Анимированные изображения не поддерживаются")
|
|
source.load()
|
|
image = ImageOps.exif_transpose(source)
|
|
image = image.convert("RGB") if image_format == "JPEG" else image.copy()
|
|
except (Image.DecompressionBombError, Image.DecompressionBombWarning):
|
|
raise ValueError("Слишком большое разрешение изображения")
|
|
except (OSError, SyntaxError) as exc:
|
|
raise ValueError("Файл не является корректным изображением") from exc
|
|
|
|
destination = Path(destination)
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
temp_path = destination.with_name(f".{destination.name}.{secrets.token_hex(8)}.uploading")
|
|
save_options = {"optimize": True}
|
|
if image_format == "JPEG":
|
|
save_options.update({"quality": 90, "progressive": True})
|
|
try:
|
|
image.save(temp_path, format=image_format, **save_options)
|
|
os.replace(temp_path, destination)
|
|
finally:
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
return destination
|