Merge pull request #90 from dsaub/security-fixes

Security fixes: image validation, email masking, quantity limits

tienda/forms.py

@@ -1,7 +1,18 @@
 from django import forms
 from django.core.exceptions import ValidationError
+from django.core.validators import FileExtensionValidator, MinLengthValidator, MaxLengthValidator
 from .models import Category
 
+ALLOWED_IMAGE_EXTENSIONS = ['jpg', 'jpeg', 'png', 'gif', 'webp']
+ALLOWED_MIME_TYPES = ['image/jpeg', 'image/png', 'image/gif', 'image/webp']
+
+def validate_image_file(value):
+    ext = value.name.split('.')[-1].lower()
+    if ext not in ALLOWED_IMAGE_EXTENSIONS:
+        raise ValidationError(f'Tipo de archivo no permitido. Allowed: {", ".join(ALLOWED_IMAGE_EXTENSIONS)}')
+    if hasattr(value, 'content_type') and value.content_type not in ALLOWED_MIME_TYPES:
+        raise ValidationError(f'Tipo MIME no permitido. Allowed: {", ".join(ALLOWED_MIME_TYPES)}')
+
 
 class ProductForm(forms.Form): 
     name = forms.CharField(
@@ -61,6 +72,7 @@ class ProductForm(forms.Form):
     primary_image = forms.ImageField(
         label="Imagen Principal",
         required = False,
+        validators=[validate_image_file],
         widget = forms.ClearableFileInput(
             attrs = {
                 'class': 'form-control',
@@ -108,6 +120,7 @@ class ProductEditForm(forms.Form):
     primary_image = forms.ImageField(
         label="Imagen Principal (opcional)",
         required=False,
+        validators=[validate_image_file],
         widget=forms.ClearableFileInput(attrs={'class': 'form-control', 'accept': 'image/*'})
     )
 
@@ -116,6 +129,7 @@ class SecondaryImageForm(forms.Form):
     image = forms.ImageField(
         label="Seleccionar Imagen",
         required = True,
+        validators=[validate_image_file],
         widget = forms.ClearableFileInput(
             attrs = {
                 'class': 'form-control',
@@ -190,7 +204,9 @@ class UserRegisterForm(forms.Form):
     password = forms.CharField(
         label = "Contraseña",
         max_length = 255,
+        min_length = 8,
         required = True,
+        validators=[MinLengthValidator(8)],
         widget = forms.PasswordInput(
             attrs = {
                 'class': 'form-control'
@@ -200,7 +216,9 @@ class UserRegisterForm(forms.Form):
     password_confirm = forms.CharField(
         label = "Verificar Contraseña",
         max_length = 255,
+        min_length = 8,
         required = True,
+        validators=[MinLengthValidator(8)],
         widget = forms.PasswordInput(
             attrs = {
                 'class': 'form-control'

tienda/models.py

@@ -3,10 +3,13 @@ from __future__ import annotations
 import unicodedata
 from django.db import models
 from django.contrib.auth.models import User, AbstractUser
+from django.core.validators import MaxValueValidator
 from django.utils.crypto import get_random_string
 from .vars import VAT_RATE, TRANSACTION_CODE_PREFIX, TRANSACTION_CODE_LENGTH, TRANSACTION_CODE_ALPHABET
 import random, string
 
+MAX_QUANTITY = 9999
+
 
 def generate_transaction_code() -> str:
     while True:
@@ -154,11 +157,16 @@ class StockReservation(models.Model):
 class StockReservationItem(models.Model):
     reservation = models.ForeignKey(StockReservation, on_delete=models.CASCADE, related_name="items")
     product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name="stock_reservation_items")
-    quantity = models.PositiveIntegerField(default=1)
+    quantity = models.PositiveIntegerField(default=1, validators=[MaxValueValidator(MAX_QUANTITY)])
 
     class Meta:
         unique_together = ("reservation", "product")
 
+    def clean(self):
+        from django.core.exceptions import ValidationError
+        if self.quantity is not None and self.quantity > MAX_QUANTITY:
+            raise ValidationError(f'La cantidad no puede exceder {MAX_QUANTITY} unidades.')
+
     def __str__(self):
         return f"{self.quantity}x {self.product.name} (reserva {self.reservation_id})"
 
@@ -190,7 +198,7 @@ class Cart(models.Model):
 class CartItem(models.Model):
     cart = models.ForeignKey(Cart, on_delete=models.CASCADE, related_name='items')
     product = models.ForeignKey(Product, on_delete=models.CASCADE)
-    quantity = models.PositiveIntegerField(default=1)
+    quantity = models.PositiveIntegerField(default=1, validators=[MaxValueValidator(MAX_QUANTITY)])
     added_at = models.DateTimeField(auto_now_add=True)
     
     class Meta:
@@ -265,7 +273,7 @@ class OrderItem(models.Model):
     product = models.ForeignKey(Product, on_delete=models.SET_NULL, null=True, blank=True)
     product_name = models.CharField(max_length=200, default="")
     seller = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='order_items_to_fulfill')
-    quantity = models.PositiveIntegerField(default=1)
+    quantity = models.PositiveIntegerField(default=1, validators=[MaxValueValidator(MAX_QUANTITY)])
     unit_price = models.FloatField(default=0)
     total_price = models.FloatField(default=0)
     status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=STATUS_PENDING)

tienda/views.py

@@ -43,6 +43,17 @@ STOCK_RESERVATION_SESSION_KEY = "stock_reservation_id"
 STOCK_RESERVATION_PAYMENT_SESSION_KEY = "stock_reservation_payment_method"
 
 
+def _mask_email(email: str) -> str:
+    if not email or '@' not in email:
+        return "***"
+    local, domain = email.rsplit('@', 1)
+    if len(local) <= 2:
+        masked_local = local[0] + '*'
+    else:
+        masked_local = local[0] + '*' * (len(local) - 2) + local[-1]
+    return f"{masked_local}@{domain}"
+
+
 def _invalidate_product_cache(product_ids):
     unique_product_ids = {product_id for product_id in product_ids if product_id is not None}
     if not unique_product_ids:
@@ -235,7 +246,7 @@ def login(request: HttpRequest):
                 user: User = User.objects.get(email=email)
                 username = user.username
             except User.DoesNotExist:
-                audit_logger.warning("LOGIN FAILED email=%s reason=user_not_found ip=%s", email, client_ip)
+                audit_logger.warning("LOGIN FAILED email=%s reason=user_not_found ip=%s", _mask_email(email), client_ip)
                 messages.error(request, "El email o la contraseña es incorrecta")
                 return render(request, "tienda/login.html", {"form": form})
             if user.registration_status == User.RegisterStatus.BANNED:
@@ -254,7 +265,7 @@ def login(request: HttpRequest):
                     logins = int(data)
                 
                 if logins >= 5:
-                    audit_logger.info("LOGIN FAILED email=%s reason=rate_limited", email)
+                    audit_logger.info("LOGIN FAILED email=%s reason=rate_limited", _mask_email(email))
                     messages.error(request, "Has sufrido de Rate Limit por fallar 5 veces la contraseña")
                     return render(request, "tienda/login.html", {"form": form})
                 logins+=1
@@ -262,7 +273,7 @@ def login(request: HttpRequest):
                 messages.error(request, "El email o la contraseña es incorrecta")
                 return render(request, "tienda/login.html", {"form": form})
             if user.registration_status == User.RegisterStatus.CONFIRMATION_REQUIRED:
-                audit_logger.info("LOGIN_FAILED email=%s reason=not_verified", email)
+                audit_logger.info("LOGIN_FAILED email=%s reason=not_verified", _mask_email(email))
                 messages.error(request, "No se puede iniciar sesión porque no has verificado tu cuenta, comprueba tu email. Si eliminaste el email pero querias verificarte, contacta con el soporte tecnico")
                 return render(request, "tienda/login.html", {"form": form})
             auth_login(request, user)
@@ -272,7 +283,7 @@ def login(request: HttpRequest):
             else:
                 request.session.set_expiry(1209600)
             
-            audit_logger.info("LOGIN_SUCCESS user_id=%s email=%s ip=%s remember=%s", user.id, user.email, client_ip, bool(remember))
+            audit_logger.info("LOGIN_SUCCESS user_id=%s email=%s ip=%s remember=%s", user.id, _mask_email(user.email), client_ip, bool(remember))
             tasks.enviar_correo_bienvenida.delay(user.email, f"{user.first_name} {user.last_name}")
             messages.success(request, f"¡Bienvenido {user.first_name or user.username}!")
             return redirect("index")
@@ -332,7 +343,7 @@ def register(request: HttpRequest):
 
             # Validación email
             if User.objects.filter(email=email).exists():
-                audit_logger.warning("REGISTER_FAILED email=%s reason=email_exists ip=%s", email, client_ip)
+                audit_logger.warning("REGISTER_FAILED email=%s reason=email_exists ip=%s", _mask_email(email), client_ip)
                 messages.error(request, "Ya existe un usuario con este correo electrónico")
                 return render(request, "tienda/register.html", {"form":form})
             
@@ -352,7 +363,7 @@ def register(request: HttpRequest):
                 "REGISTER_SUCCESS user_id=%s username=%s email=%s ip=%s",
                 user.id,
                 user.username,
-                user.email,
+                _mask_email(user.email),
                 client_ip,
             )
         
@@ -370,7 +381,7 @@ def logout(request: HttpRequest):
     email = request.user.email if request.user.is_authenticated else None
     client_ip = _get_client_ip(request)
     auth_logout(request)
-    audit_logger.info("LOGOUT user_id=%s email=%s ip=%s", user_id, email, client_ip)
+    audit_logger.info("LOGOUT user_id=%s email=%s ip=%s", user_id, _mask_email(email) if email else "***", client_ip)
     messages.success(request, "Has cerrado sesión exitosamente.")
     return redirect("index")