overlay face function

pyproject.toml

@@ -83,6 +83,8 @@ async-timeout = "^5.0.1"
 aiohttp = "^3.11.11"
 scout-apm = "^3.3.0"
 newrelic = "^10.4.0"
+opencv-python = "^4.8.0"
+numpy = "^1.24.0"
 
 [tool.poetry.group.dev.dependencies]
 black = "^24.8.0"

website/models.py

@@ -7,6 +7,8 @@
 from enum import Enum
 from urllib.parse import urlparse
 
+import cv2
+import numpy as np
 import requests
 from annoying.fields import AutoOneToOneField
 from captcha.fields import CaptchaField
@@ -32,6 +34,8 @@
 from mdeditor.fields import MDTextField
 from rest_framework.authtoken.models import Token
 
+from .views.privacy import overlay_faces
+
 logger = logging.getLogger(__name__)
 
 
@@ -263,12 +267,39 @@ def get_or_set_x_url(self, name):
 
 def validate_image(fieldfile_obj):
     try:
+        fieldfile_obj.file.seek(0)
+
+        # Read image data
         filesize = fieldfile_obj.file.size
-    except:
-        filesize = fieldfile_obj.size
+        np_img = np.frombuffer(fieldfile_obj.file.read(), np.uint8)
+        img = cv2.imdecode(np_img, cv2.IMREAD_UNCHANGED)
+
+        if img is None:
+            raise ValueError("Failed to decode image")
+
+        # Apply face blurring
+        img_with_faces_hidden = overlay_faces(img)
+
+        # Encode back to file
+        ext = os.path.splitext(fieldfile_obj.name)[1].lower()
+        if ext in [".jpg", ".jpeg"]:
+            _, buffer = cv2.imencode(".jpg", img_with_faces_hidden, [cv2.IMWRITE_JPEG_QUALITY, 90])
+        else:
+            _, buffer = cv2.imencode(".png", img_with_faces_hidden)
+
+        # Reset file with processed image
+        fieldfile_obj.file = ContentFile(buffer.tobytes())
+        fieldfile_obj.file.seek(0)
+
+    except Exception as e:
+        print(f"Image validation error: {str(e)}")
+        return False
+
     megabyte_limit = 3.0
     if filesize > megabyte_limit * 1024 * 1024:
-        raise ValidationError("Max file size is %sMB" % str(megabyte_limit))
+        raise ValidationError(f"Max file size is {str(megabyte_limit)}MB")
+
+    return True
 
 
 class Hunt(models.Model):

website/views/issue.py

@@ -2,10 +2,13 @@
 import io
 import json
 import os
+import time
 import uuid
 from datetime import datetime, timezone
 from urllib.parse import urlparse
 
+import cv2
+import numpy as np
 import requests
 import six
 import tweepy
@@ -70,6 +73,8 @@
     safe_redirect_request,
 )
 
+from .privacy import overlay_faces
+
 
 @login_required(login_url="/accounts/login")
 def like_issue(request, issue_pk):
@@ -355,24 +360,39 @@ def newhome(request, template="new_home.html"):
 
 
 def delete_issue(request, id):
+    tokenauth = False
+
+    # Check for token in POST or GET
+    token_key = request.POST.get("token") or request.GET.get("token")
+
+    if token_key:
+        try:
+            token = Token.objects.get(key=token_key)
+            request.user = User.objects.get(id=token.user_id)
+            tokenauth = True
+        except (Token.DoesNotExist, User.DoesNotExist):
+            tokenauth = False
+
     try:
-        # TODO: Refactor this for a direct query instead of looping through all tokens
-        for token in Token.objects.all():
-            if request.POST["token"] == token.key:
-                request.user = User.objects.get(id=token.user_id)
-                tokenauth = True
-    except Token.DoesNotExist:
-        tokenauth = False
+        issue = Issue.objects.get(id=id)
+    except Issue.DoesNotExist:
+        messages.error(request, "Issue not found")
+        return redirect("/")
 
-    issue = Issue.objects.get(id=id)
     if request.user.is_superuser or request.user == issue.user or tokenauth:
-        screenshots = issue.screenshots.all()
-        for screenshot in screenshots:
-            screenshot.delete()
-        issue.delete()
-        messages.success(request, "Issue deleted")
-        if tokenauth:
-            return JsonResponse("Deleted", safe=False)
+        try:
+            screenshots = issue.screenshots.all()
+            for screenshot in screenshots:
+                screenshot.delete()
+            issue.delete()
+            messages.success(request, "Issue deleted")
+            if tokenauth:
+                return JsonResponse("Deleted", safe=False)
+        except Exception as e:
+            messages.error(request, f"Error deleting issue: {str(e)}")
+    else:
+        messages.error(request, "Permission denied")
+
     return redirect("/")
 
 
@@ -1003,11 +1023,55 @@ def create_issue(self, form):
 
             # Save screenshots
             for screenshot in self.request.FILES.getlist("screenshots"):
-                filename = screenshot.name
-                extension = filename.split(".")[-1]
-                screenshot.name = (filename[:10] + str(uuid.uuid4()))[:40] + "." + extension
-                default_storage.save(f"screenshots/{screenshot.name}", screenshot)
-                IssueScreenshot.objects.create(image=f"screenshots/{screenshot.name}", issue=obj)
+                try:
+                    screenshot.seek(0)
+
+                    # Read image data
+                    image_data = screenshot.read()
+                    np_img = np.frombuffer(image_data, np.uint8)
+                    img = cv2.imdecode(np_img, cv2.IMREAD_UNCHANGED)
+
+                    if img is None:
+                        raise ValueError("Failed to decode image")
+
+                    # Process image - original returned if no faces
+                    img_with_faces_hidden = overlay_faces(img)
+
+                    # Save processed image
+                    filename = screenshot.name
+                    extension = filename.split(".")[-1].lower()
+
+                    if extension not in ["jpg", "jpeg", "png"]:
+                        extension = "jpg"
+
+                    new_filename = f"{filename[:10]}_{uuid.uuid4()}_{int(time.time())}.{extension}"
+
+                    # Encode and save image
+                    if extension in ["jpg", "jpeg"]:
+                        _, buffer = cv2.imencode(
+                            ".jpg", img_with_faces_hidden, [cv2.IMWRITE_JPEG_QUALITY, 90]
+                        )
+                    else:
+                        _, buffer = cv2.imencode(".png", img_with_faces_hidden)
+
+                    if buffer is None:
+                        raise ValueError("Failed to encode processed image")
+
+                    processed_image = ContentFile(buffer.tobytes())
+                    saved_path = default_storage.save(
+                        f"screenshots/{new_filename}", processed_image
+                    )
+
+                    # Create screenshot object
+                    IssueScreenshot.objects.create(image=saved_path, issue=obj)
+
+                except Exception as e:
+                    messages.error(self.request, f"Error processing image: {str(e)}")
+                    return render(
+                        self.request,
+                        "report.html",
+                        {"form": self.get_form(), "captcha_form": CaptchaForm()},
+                    )
 
             # Handle team members
             team_members_id = [

website/views/privacy.py

@@ -0,0 +1,32 @@
+import cv2
+
+
+def overlay_faces(img, color=(255, 255, 255)):
+    """Apply white rectangles over detected faces in an image."""
+    if img is None:
+        raise ValueError("Invalid image input - image could not be decoded")
+
+    if len(img.shape) == 2:
+        img = cv2.cvtColor(img, cv2.COLOR_GRAY2BGR)
+    elif len(img.shape) == 3 and img.shape[2] == 4:
+        img = cv2.cvtColor(img, cv2.COLOR_BGRA2BGR)
+
+    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+
+    face_cascade = cv2.CascadeClassifier(
+        cv2.data.haarcascades + "haarcascade_frontalface_default.xml"
+    )
+    if face_cascade.empty():
+        return img
+
+    # Detect faces with tuned parameters
+    faces = face_cascade.detectMultiScale(
+        gray, scaleFactor=1.1, minNeighbors=5, minSize=(30, 30), flags=cv2.CASCADE_SCALE_IMAGE
+    )
+
+    if len(faces) == 0:
+        return img
+
+    for x, y, w, h in faces:
+        cv2.rectangle(img, (x, y), (x + w, y + h), color, thickness=cv2.FILLED)
+    return img