Initial commit: InviteMe production redesign
This commit is contained in:
@@ -0,0 +1,121 @@
|
||||
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
|
||||
Reference in New Issue
Block a user