feat: upload file in contact form
This commit is contained in:
+44
-2
@@ -1,11 +1,13 @@
|
||||
from django.contrib import admin
|
||||
from django.http import HttpResponseRedirect
|
||||
from django.urls import reverse
|
||||
from django.http import FileResponse, Http404, HttpResponseRedirect
|
||||
from django.urls import path, reverse
|
||||
from django.utils.html import format_html
|
||||
|
||||
from .models import (
|
||||
AboutSection,
|
||||
AboutSectionItem,
|
||||
ContactSubmission,
|
||||
ContactSubmissionAttachment,
|
||||
CustomPage,
|
||||
CustomPageSection,
|
||||
CustomPageSectionItem,
|
||||
@@ -83,6 +85,21 @@ class AboutSectionAdmin(admin.ModelAdmin):
|
||||
)
|
||||
|
||||
|
||||
class ContactSubmissionAttachmentInline(admin.TabularInline):
|
||||
model = ContactSubmissionAttachment
|
||||
extra = 0
|
||||
can_delete = False
|
||||
fields = ("original_filename", "attachment_link", "uploaded_at")
|
||||
readonly_fields = ("original_filename", "attachment_link", "uploaded_at")
|
||||
|
||||
@admin.display(description="File")
|
||||
def attachment_link(self, obj):
|
||||
if not obj.file:
|
||||
return "—"
|
||||
url = reverse("admin:pages_contactattachment_download", args=[obj.pk])
|
||||
return format_html('<a href="{}">Download</a>', url)
|
||||
|
||||
|
||||
@admin.register(ContactSubmission)
|
||||
class ContactSubmissionAdmin(admin.ModelAdmin):
|
||||
list_display = ("name", "title", "email", "submitted_at", "is_read")
|
||||
@@ -90,11 +107,36 @@ class ContactSubmissionAdmin(admin.ModelAdmin):
|
||||
search_fields = ("name", "title", "description", "email")
|
||||
list_editable = ("is_read",)
|
||||
readonly_fields = ("name", "title", "description", "email", "submitted_at")
|
||||
inlines = [ContactSubmissionAttachmentInline]
|
||||
fieldsets = (
|
||||
(None, {"fields": ("name", "title", "email", "description")}),
|
||||
("Meta", {"fields": ("submitted_at", "is_read")}),
|
||||
)
|
||||
|
||||
def get_urls(self):
|
||||
urls = super().get_urls()
|
||||
custom_urls = [
|
||||
path(
|
||||
"attachment/<int:attachment_id>/download/",
|
||||
self.admin_site.admin_view(self.download_attachment),
|
||||
name="pages_contactattachment_download",
|
||||
),
|
||||
]
|
||||
return custom_urls + urls
|
||||
|
||||
def download_attachment(self, request, attachment_id):
|
||||
try:
|
||||
attachment = ContactSubmissionAttachment.objects.get(pk=attachment_id)
|
||||
except ContactSubmissionAttachment.DoesNotExist as exc:
|
||||
raise Http404 from exc
|
||||
if not attachment.file:
|
||||
raise Http404
|
||||
return FileResponse(
|
||||
attachment.file.open("rb"),
|
||||
as_attachment=True,
|
||||
filename=attachment.original_filename,
|
||||
)
|
||||
|
||||
|
||||
@admin.register(FAQEntry)
|
||||
class FAQEntryAdmin(admin.ModelAdmin):
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
import os
|
||||
import uuid
|
||||
import zipfile
|
||||
from pathlib import PurePosixPath
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.files.storage import FileSystemStorage
|
||||
from django.core.files.uploadedfile import UploadedFile
|
||||
from django.utils.text import get_valid_filename
|
||||
from PIL import Image
|
||||
|
||||
ALLOWED_EXTENSIONS = frozenset(
|
||||
{
|
||||
".png",
|
||||
".jpg",
|
||||
".jpeg",
|
||||
".gif",
|
||||
".webp",
|
||||
".bmp",
|
||||
".zip",
|
||||
".log",
|
||||
".txt",
|
||||
}
|
||||
)
|
||||
|
||||
IMAGE_EXTENSIONS = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"})
|
||||
TEXT_EXTENSIONS = frozenset({".log", ".txt"})
|
||||
ARCHIVE_EXTENSIONS = frozenset({".zip"})
|
||||
|
||||
MAX_FILE_SIZE = getattr(settings, "CONTACT_ATTACHMENT_MAX_SIZE", 10 * 1024 * 1024)
|
||||
MAX_ATTACHMENTS = getattr(settings, "CONTACT_ATTACHMENT_MAX_COUNT", 3)
|
||||
MAX_ZIP_ENTRIES = 100
|
||||
MAX_ZIP_UNCOMPRESSED_SIZE = 50 * 1024 * 1024
|
||||
|
||||
|
||||
class ContactAttachmentStorage(FileSystemStorage):
|
||||
def __init__(self):
|
||||
super().__init__(location=settings.CONTACT_UPLOAD_ROOT)
|
||||
|
||||
|
||||
contact_attachment_storage = ContactAttachmentStorage()
|
||||
|
||||
|
||||
def contact_attachment_upload_to(instance, _filename):
|
||||
ext = os.path.splitext(instance.original_filename)[1].lower()
|
||||
if ext not in ALLOWED_EXTENSIONS:
|
||||
ext = ".bin"
|
||||
subdir = str(instance.submission_id) if instance.submission_id else "pending"
|
||||
return f"{subdir}/{uuid.uuid4().hex}{ext}"
|
||||
|
||||
|
||||
def _extension(filename):
|
||||
return os.path.splitext(filename)[1].lower()
|
||||
|
||||
|
||||
def _read_header(uploaded_file, size=32):
|
||||
uploaded_file.seek(0)
|
||||
header = uploaded_file.read(size)
|
||||
uploaded_file.seek(0)
|
||||
return header
|
||||
|
||||
|
||||
def _validate_magic(header, ext):
|
||||
if ext == ".png" and not header.startswith(b"\x89PNG\r\n\x1a\n"):
|
||||
raise ValidationError("File content does not match its type.")
|
||||
if ext in (".jpg", ".jpeg") and not header.startswith(b"\xff\xd8\xff"):
|
||||
raise ValidationError("File content does not match its type.")
|
||||
if ext == ".gif" and not (
|
||||
header.startswith(b"GIF87a") or header.startswith(b"GIF89a")
|
||||
):
|
||||
raise ValidationError("File content does not match its type.")
|
||||
if ext == ".webp" and not (
|
||||
len(header) >= 12 and header[0:4] == b"RIFF" and header[8:12] == b"WEBP"
|
||||
):
|
||||
raise ValidationError("File content does not match its type.")
|
||||
if ext == ".bmp" and not header.startswith(b"BM"):
|
||||
raise ValidationError("File content does not match its type.")
|
||||
if ext == ".zip" and not (
|
||||
header.startswith(b"PK\x03\x04")
|
||||
or header.startswith(b"PK\x05\x06")
|
||||
or header.startswith(b"PK\x07\x08")
|
||||
):
|
||||
raise ValidationError("File content does not match its type.")
|
||||
|
||||
|
||||
def _validate_image(uploaded_file):
|
||||
try:
|
||||
uploaded_file.seek(0)
|
||||
with Image.open(uploaded_file) as img:
|
||||
img.verify()
|
||||
uploaded_file.seek(0)
|
||||
with Image.open(uploaded_file) as img:
|
||||
img.load()
|
||||
except Exception as exc:
|
||||
raise ValidationError("Invalid or corrupted image file.") from exc
|
||||
finally:
|
||||
uploaded_file.seek(0)
|
||||
|
||||
|
||||
def _validate_text(uploaded_file):
|
||||
uploaded_file.seek(0)
|
||||
data = uploaded_file.read()
|
||||
uploaded_file.seek(0)
|
||||
if b"\x00" in data:
|
||||
raise ValidationError("Text files must not contain binary data.")
|
||||
try:
|
||||
data.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
try:
|
||||
data.decode("latin-1")
|
||||
except UnicodeDecodeError as exc:
|
||||
raise ValidationError("Text file is not valid UTF-8 or Latin-1.") from exc
|
||||
|
||||
|
||||
def _validate_zip(uploaded_file):
|
||||
uploaded_file.seek(0)
|
||||
if not zipfile.is_zipfile(uploaded_file):
|
||||
raise ValidationError("Invalid ZIP archive.")
|
||||
uploaded_file.seek(0)
|
||||
|
||||
total_uncompressed = 0
|
||||
entry_count = 0
|
||||
|
||||
with zipfile.ZipFile(uploaded_file, "r") as archive:
|
||||
for info in archive.infolist():
|
||||
entry_count += 1
|
||||
if entry_count > MAX_ZIP_ENTRIES:
|
||||
raise ValidationError("ZIP archive contains too many files.")
|
||||
|
||||
name = info.filename
|
||||
if name.startswith("/") or ".." in PurePosixPath(name).parts:
|
||||
raise ValidationError("ZIP archive contains unsafe paths.")
|
||||
|
||||
if info.flag_bits & 0x1:
|
||||
raise ValidationError("Encrypted ZIP archives are not allowed.")
|
||||
|
||||
total_uncompressed += info.file_size
|
||||
if total_uncompressed > MAX_ZIP_UNCOMPRESSED_SIZE:
|
||||
raise ValidationError("ZIP archive uncompressed size is too large.")
|
||||
|
||||
uploaded_file.seek(0)
|
||||
|
||||
|
||||
def validate_contact_attachment(uploaded_file):
|
||||
if uploaded_file.size > MAX_FILE_SIZE:
|
||||
max_mb = MAX_FILE_SIZE // (1024 * 1024)
|
||||
raise ValidationError(f"File exceeds the maximum size of {max_mb} MB.")
|
||||
|
||||
name = get_valid_filename(os.path.basename(uploaded_file.name))
|
||||
if not name:
|
||||
raise ValidationError("Invalid file name.")
|
||||
|
||||
ext = _extension(name)
|
||||
if ext not in ALLOWED_EXTENSIONS:
|
||||
raise ValidationError(
|
||||
"File type not allowed. Permitted: images (PNG, JPG, GIF, WebP, BMP), "
|
||||
"ZIP, LOG, or TXT."
|
||||
)
|
||||
|
||||
header = _read_header(uploaded_file)
|
||||
if ext in IMAGE_EXTENSIONS or ext in ARCHIVE_EXTENSIONS:
|
||||
_validate_magic(header, ext)
|
||||
|
||||
if ext in IMAGE_EXTENSIONS:
|
||||
_validate_image(uploaded_file)
|
||||
elif ext in TEXT_EXTENSIONS:
|
||||
_validate_text(uploaded_file)
|
||||
elif ext in ARCHIVE_EXTENSIONS:
|
||||
_validate_zip(uploaded_file)
|
||||
|
||||
return name
|
||||
|
||||
|
||||
def validate_contact_attachments(files):
|
||||
if not files:
|
||||
return []
|
||||
|
||||
if len(files) > MAX_ATTACHMENTS:
|
||||
raise ValidationError(f"You can attach at most {MAX_ATTACHMENTS} files.")
|
||||
|
||||
validated = []
|
||||
for uploaded_file in files:
|
||||
if not isinstance(uploaded_file, UploadedFile):
|
||||
raise ValidationError("Invalid upload.")
|
||||
original_name = validate_contact_attachment(uploaded_file)
|
||||
validated.append((uploaded_file, original_name))
|
||||
return validated
|
||||
@@ -1,9 +1,13 @@
|
||||
from django import forms
|
||||
from django.core.exceptions import ValidationError
|
||||
|
||||
from .contact_uploads import validate_contact_attachments
|
||||
from .models import ContactSubmission
|
||||
|
||||
|
||||
class ContactForm(forms.ModelForm):
|
||||
attachments = forms.Field(required=False)
|
||||
|
||||
class Meta:
|
||||
model = ContactSubmission
|
||||
fields = ["name", "title", "description", "email"]
|
||||
@@ -22,3 +26,16 @@ class ContactForm(forms.ModelForm):
|
||||
}
|
||||
),
|
||||
}
|
||||
|
||||
def __init__(self, *args, file_list=None, **kwargs):
|
||||
self.file_list = file_list
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def clean(self):
|
||||
cleaned_data = super().clean()
|
||||
try:
|
||||
cleaned_data["attachments"] = validate_contact_attachments(self.file_list)
|
||||
except ValidationError as exc:
|
||||
self.add_error("attachments", exc)
|
||||
cleaned_data["attachments"] = []
|
||||
return cleaned_data
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
# Generated by Django 5.0.2 on 2026-06-07 13:23
|
||||
|
||||
import apps.pages.contact_uploads
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('pages', '0012_aboutsectionitem_icon'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='ContactSubmissionAttachment',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('file', models.FileField(storage=apps.pages.contact_uploads.ContactAttachmentStorage(), upload_to=apps.pages.contact_uploads.contact_attachment_upload_to)),
|
||||
('original_filename', models.CharField(max_length=255)),
|
||||
('uploaded_at', models.DateTimeField(auto_now_add=True)),
|
||||
('submission', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='attachments', to='pages.contactsubmission')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'Contact Attachment',
|
||||
'verbose_name_plural': 'Contact Attachments',
|
||||
'ordering': ['uploaded_at'],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -3,6 +3,10 @@ from django.db import models
|
||||
from django.utils.text import slugify
|
||||
|
||||
from apps.core.utils import CONTENT_FORMAT_CHOICES, FORMAT_MARKDOWN, FORMAT_PLAIN, render_content
|
||||
from apps.pages.contact_uploads import (
|
||||
contact_attachment_storage,
|
||||
contact_attachment_upload_to,
|
||||
)
|
||||
|
||||
RESERVED_PAGE_SLUGS = frozenset({
|
||||
"admin",
|
||||
@@ -177,6 +181,28 @@ class ContactSubmission(models.Model):
|
||||
return f"{self.name} — {self.title}"
|
||||
|
||||
|
||||
class ContactSubmissionAttachment(models.Model):
|
||||
submission = models.ForeignKey(
|
||||
ContactSubmission,
|
||||
on_delete=models.CASCADE,
|
||||
related_name="attachments",
|
||||
)
|
||||
file = models.FileField(
|
||||
upload_to=contact_attachment_upload_to,
|
||||
storage=contact_attachment_storage,
|
||||
)
|
||||
original_filename = models.CharField(max_length=255)
|
||||
uploaded_at = models.DateTimeField(auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
ordering = ["uploaded_at"]
|
||||
verbose_name = "Contact Attachment"
|
||||
verbose_name_plural = "Contact Attachments"
|
||||
|
||||
def __str__(self):
|
||||
return self.original_filename
|
||||
|
||||
|
||||
class FAQEntry(models.Model):
|
||||
question = models.CharField(max_length=500)
|
||||
answer = models.TextField()
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
import io
|
||||
import zipfile
|
||||
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from django.test import Client, TestCase, override_settings
|
||||
from django.urls import reverse
|
||||
from PIL import Image
|
||||
|
||||
from apps.pages.contact_uploads import (
|
||||
MAX_ATTACHMENTS,
|
||||
MAX_FILE_SIZE,
|
||||
validate_contact_attachments,
|
||||
)
|
||||
from apps.pages.models import ContactSubmission, ContactSubmissionAttachment
|
||||
|
||||
|
||||
def _png_file(name="screenshot.png"):
|
||||
buffer = io.BytesIO()
|
||||
Image.new("RGB", (8, 8), color="red").save(buffer, format="PNG")
|
||||
buffer.seek(0)
|
||||
return SimpleUploadedFile(name, buffer.read(), content_type="image/png")
|
||||
|
||||
|
||||
def _zip_file(name="logs.zip", entries=None):
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
|
||||
for entry_name, content in (entries or {"app.log": "line one\n"}).items():
|
||||
archive.writestr(entry_name, content)
|
||||
buffer.seek(0)
|
||||
return SimpleUploadedFile(name, buffer.read(), content_type="application/zip")
|
||||
|
||||
|
||||
class ContactUploadValidationTest(TestCase):
|
||||
def test_accepts_valid_png(self):
|
||||
validated = validate_contact_attachments([_png_file()])
|
||||
self.assertEqual(len(validated), 1)
|
||||
self.assertEqual(validated[0][1], "screenshot.png")
|
||||
|
||||
def test_accepts_valid_zip(self):
|
||||
validated = validate_contact_attachments([_zip_file()])
|
||||
self.assertEqual(len(validated), 1)
|
||||
|
||||
def test_accepts_valid_text_log(self):
|
||||
uploaded = SimpleUploadedFile(
|
||||
"error.log",
|
||||
b"2026-06-07 ERROR something failed\n",
|
||||
content_type="text/plain",
|
||||
)
|
||||
validated = validate_contact_attachments([uploaded])
|
||||
self.assertEqual(validated[0][1], "error.log")
|
||||
|
||||
def test_rejects_executable_extension(self):
|
||||
uploaded = SimpleUploadedFile(
|
||||
"malware.exe",
|
||||
b"MZfake",
|
||||
content_type="application/octet-stream",
|
||||
)
|
||||
with self.assertRaises(Exception):
|
||||
validate_contact_attachments([uploaded])
|
||||
|
||||
def test_rejects_php_disguised_as_png(self):
|
||||
uploaded = SimpleUploadedFile(
|
||||
"image.png",
|
||||
b"<?php echo 'bad'; ?>",
|
||||
content_type="image/png",
|
||||
)
|
||||
with self.assertRaises(Exception):
|
||||
validate_contact_attachments([uploaded])
|
||||
|
||||
def test_rejects_oversized_file(self):
|
||||
uploaded = SimpleUploadedFile(
|
||||
"big.log",
|
||||
b"x" * (MAX_FILE_SIZE + 1),
|
||||
content_type="text/plain",
|
||||
)
|
||||
with self.assertRaises(Exception):
|
||||
validate_contact_attachments([uploaded])
|
||||
|
||||
def test_rejects_too_many_files(self):
|
||||
files = [_png_file(f"shot-{index}.png") for index in range(MAX_ATTACHMENTS + 1)]
|
||||
with self.assertRaises(Exception):
|
||||
validate_contact_attachments(files)
|
||||
|
||||
def test_rejects_zip_with_path_traversal(self):
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, "w") as archive:
|
||||
archive.writestr("../escape.txt", "bad")
|
||||
buffer.seek(0)
|
||||
uploaded = SimpleUploadedFile(
|
||||
"bad.zip",
|
||||
buffer.read(),
|
||||
content_type="application/zip",
|
||||
)
|
||||
with self.assertRaises(Exception):
|
||||
validate_contact_attachments([uploaded])
|
||||
|
||||
|
||||
@override_settings(
|
||||
CONTACT_UPLOAD_ROOT=__import__("pathlib").Path(__file__).resolve().parents[3]
|
||||
/ "test_private_uploads"
|
||||
)
|
||||
class ContactViewUploadTest(TestCase):
|
||||
def setUp(self):
|
||||
self.client = Client(enforce_csrf_checks=True)
|
||||
self.url = reverse("pages:contact")
|
||||
|
||||
def _start_session(self):
|
||||
response = self.client.get(self.url)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.csrf_token = response.cookies["csrftoken"].value
|
||||
captcha_question = response.context["captcha_question"]
|
||||
left, right = captcha_question.split(" + ")
|
||||
return int(left) + int(right)
|
||||
|
||||
def _post_contact(self, captcha_answer, attachments=None, extra=None):
|
||||
payload = {
|
||||
"name": "Test User",
|
||||
"title": "Upload test",
|
||||
"description": "Testing attachments",
|
||||
"email": "test@example.com",
|
||||
"captcha_answer": captcha_answer,
|
||||
}
|
||||
if attachments is not None:
|
||||
payload["attachments"] = attachments
|
||||
if extra:
|
||||
payload.update(extra)
|
||||
return self.client.post(
|
||||
self.url,
|
||||
data=payload,
|
||||
HTTP_X_CSRFTOKEN=self.csrf_token,
|
||||
)
|
||||
|
||||
def test_contact_submission_with_png_attachment(self):
|
||||
captcha_answer = self._start_session()
|
||||
response = self._post_contact(captcha_answer, attachments=_png_file())
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue(response.json()["success"])
|
||||
submission = ContactSubmission.objects.get(title="Upload test")
|
||||
self.assertEqual(submission.attachments.count(), 1)
|
||||
attachment = submission.attachments.first()
|
||||
self.assertEqual(attachment.original_filename, "screenshot.png")
|
||||
self.assertTrue(attachment.file.storage.exists(attachment.file.name))
|
||||
|
||||
def test_contact_submission_rejects_invalid_attachment(self):
|
||||
captcha_answer = self._start_session()
|
||||
response = self._post_contact(
|
||||
captcha_answer,
|
||||
attachments=SimpleUploadedFile(
|
||||
"bad.exe",
|
||||
b"MZ",
|
||||
content_type="application/octet-stream",
|
||||
),
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertFalse(response.json()["success"])
|
||||
self.assertIn("attachments", response.json()["errors"])
|
||||
self.assertEqual(ContactSubmission.objects.count(), 0)
|
||||
self.assertEqual(ContactSubmissionAttachment.objects.count(), 0)
|
||||
+14
-2
@@ -12,6 +12,7 @@ from .forms import ContactForm
|
||||
from .models import (
|
||||
AboutSection,
|
||||
ContactSubmission,
|
||||
ContactSubmissionAttachment,
|
||||
CustomPage,
|
||||
FAQEntry,
|
||||
HeroSection,
|
||||
@@ -87,7 +88,10 @@ class ContactView(View):
|
||||
)
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
form = ContactForm(request.POST)
|
||||
form = ContactForm(
|
||||
request.POST,
|
||||
file_list=request.FILES.getlist("attachments"),
|
||||
)
|
||||
expected = request.session.get("captcha_answer")
|
||||
captcha_question = self._new_captcha(request)
|
||||
|
||||
@@ -98,7 +102,15 @@ class ContactView(View):
|
||||
pass
|
||||
|
||||
if form.is_valid() and captcha_ok:
|
||||
form.save()
|
||||
submission = form.save()
|
||||
for uploaded_file, original_name in form.cleaned_data.get(
|
||||
"attachments", []
|
||||
):
|
||||
attachment = ContactSubmissionAttachment(
|
||||
submission=submission,
|
||||
original_filename=original_name,
|
||||
)
|
||||
attachment.file.save(original_name, uploaded_file, save=True)
|
||||
return JsonResponse({"success": True})
|
||||
|
||||
errors: dict = {}
|
||||
|
||||
Reference in New Issue
Block a user