Integrity Failures in Django
How Integrity Failures Manifests in Django
Integrity failures in Django applications typically occur when data validation, authorization checks, or business logic constraints are bypassed or improperly implemented. These vulnerabilities allow attackers to manipulate data in ways that violate the intended application logic.
One common manifestation is Model Integrity Bypass through Django's ORM. Consider a Django model with a custom save() method that enforces business rules:
class Account(models.Model):
user = models.ForeignKey(User, on_delete=models.CASCADE)
balance = models.DecimalField(max_digits=10, decimal_places=2)
is_active = models.BooleanField(default=True)
def save(self, *args, **kwargs):
if self.balance < 0 and self.is_active:
raise ValidationError('Cannot have negative balance')
super().save(*args, **kwargs)
An attacker can bypass this validation by directly manipulating the database or using Django's update() method, which skips the save() method entirely:
# This bypasses the save() validation
Account.objects.filter(id=123).update(balance=-1000.00)
Another Django-specific integrity failure occurs with ForeignKey Integrity. Django's ORM provides convenient access to related objects, but this can be exploited:
# Vulnerable: No authorization check
class OrderViewSet(viewsets.ModelViewSet):
def retrieve(self, request, pk=None):
order = get_object_or_404(Order, pk=pk)
return Response({
'order': order,
'customer': order.customer.username # Potential data exposure
})
An attacker could enumerate order IDs to access other users' order information, as there's no check that the requesting user owns the order.
Transaction Integrity Failures are also prevalent in Django. When multiple database operations should succeed or fail together but don't:
from django.db import transaction
@transaction.atomic
def transfer_funds(from_account_id, to_account_id, amount):
from_account = Account.objects.select_for_update().get(id=from_account_id)
to_account = Account.objects.select_for_update().get(id=to_account_id)
from_account.balance -= amount
from_account.save()
# Simulated failure (e.g., network error, validation error)
if amount > 1000:
raise ValueError('Transfer too large')
to_account.balance += amount
to_account.save()
The issue here is that the first save() commits before the transaction is complete, potentially leaving the system in an inconsistent state if the second operation fails.
Django-Specific Detection
Detecting integrity failures in Django requires both static code analysis and dynamic runtime testing. middleBrick's Django-specific scanning looks for several critical patterns:
ORM Method Analysis: middleBrick identifies usage of update(), bulk_create(), and bulk_update() methods that bypass model validation:
from django.db.models import F
# middleBrick flags this as high risk
Account.objects.filter(user=request.user).update(balance=F('balance') - amount)
Serializer Validation Gaps: When using Django REST Framework, middleBrick checks for serializers that don't properly validate input:
class OrderSerializer(serializers.ModelSerializer):
class Meta:
model = Order
fields = '__all__'
# middleBrick detects missing validation for critical fields
Permission Logic Analysis: middleBrick examines view permissions and object-level permissions:
class OrderViewSet(viewsets.ModelViewSet):
permission_classes = [IsAuthenticated] # middleBrick flags this as insufficient
def get_queryset(self):
return Order.objects.filter(user=self.request.user)
The scanner also tests for race conditions by sending concurrent requests that could exploit transaction timing issues. For example:
# middleBrick simulates:
# 1. Request A: Starts transfer of $1000
# 2. Request B: Attempts transfer of $1000 from same account
# 3. Request A: Completes
# 4. Request B: Should fail but might succeed due to race
Middleware Integrity Checks are another focus area. middleBrick analyzes custom middleware that might modify requests or responses without proper validation:
class IntegrityMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# middleBrick checks if this properly validates session integrity
if not self.validate_session_integrity(request):
return HttpResponseForbidden()
return self.get_response(request)
Django-Specific Remediation
Remediating integrity failures in Django requires a multi-layered approach using Django's built-in security features and best practices.
Model-Level Protection: Use Django's model validation and signals to enforce integrity:
from django.db.models import signals
from django.core.exceptions import ValidationError
class Account(models.Model):
user = models.ForeignKey(User, on_delete=models.CASCADE)
balance = models.DecimalField(max_digits=10, decimal_places=2)
is_active = models.BooleanField(default=True)
def clean(self):
if self.balance < 0 and self.is_active:
raise ValidationError('Cannot have negative balance')
def save(self, *args, **kwargs):
self.full_clean()
super().save(*args, **kwargs)
# Signal to prevent direct database manipulation
def validate_account_integrity(sender, instance, **kwargs):
if instance.balance < 0:
raise ValidationError('Account integrity violation detected')
signals.pre_save.connect(validate_account_integrity, sender=Account)
Transaction Management: Always use atomic transactions for operations that must succeed or fail together:
from django.db import transaction
from django.db.models import F
@transaction.atomic
def transfer_funds_atomic(from_account_id, to_account_id, amount):
from_account = Account.objects.select_for_update().get(id=from_account_id)
to_account = Account.objects.select_for_update().get(id=to_account_id)
if from_account.balance < amount:
raise ValidationError('Insufficient funds')
# Use F() expressions to avoid race conditions
Account.objects.filter(id=from_account_id).update(
balance=F('balance') - amount
)
Account.objects.filter(id=to_account_id).update(
balance=F('balance') + amount
)
DRF Permission Classes: Implement robust permission checking:
from rest_framework import permissions
class IsOwnerOrReadOnly(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
if request.method in permissions.SAFE_METHODS:
return True
return obj.user == request.user
class OrderViewSet(viewsets.ModelViewSet):
queryset = Order.objects.all()
serializer_class = OrderSerializer
permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]
def get_queryset(self):
return self.queryset.filter(user=self.request.user)
Input Validation with Pydantic or Django Validators:
from django.core.validators import MinValueValidator
class Transaction(models.Model):
amount = models.DecimalField(
max_digits=10,
decimal_places=2,
validators=[MinValueValidator(0.01)]
)
description = models.CharField(max_length=200)
def clean(self):
if self.amount > 10000:
raise ValidationError('Transaction amount too large')
API Rate Limiting to prevent brute-force integrity attacks:
from rest_framework.throttling import UserRateThrottle
class HighPriorityThrottle(UserRateThrottle):
rate = '10/minute'
class OrderViewSet(viewsets.ModelViewSet):
throttle_classes = [HighPriorityThrottle]