CWE-566: Authorization Bypass Through User-Controlled SQL Primary Key - Python
Overview
Authorization bypass through user-controlled keys (also known as Insecure Direct Object Reference or IDOR) happens when a Python web application uses a user-supplied identifier - a user ID, document ID, or account number - to retrieve a resource without verifying that the authenticated user may access it. Changing that identifier in a URL parameter or request body then returns another user's data, which is horizontal privilege escalation.
Python's web frameworks (Django, Flask, FastAPI) make database queries easy to write, and the authorization check is the part that gets left out. Using request parameters directly in ORM queries (User.query.get(user_id) or Document.objects.get(id=doc_id)) with no ownership filter is a common shape in Python codebases. Django provides some built-in protections through its permission system; Flask and FastAPI require developers to implement authorization logic explicitly.
The three frameworks fail in the same place through different APIs, so a fix in one does not transfer by pattern-matching. Flask's Model.query.get(id), Django's get_object_or_404(Model, id=...) and a FastAPI handler's db.query(Model).filter(Model.id == id).first() all resolve a primary key the caller supplied, and none of them consults the caller's identity. Django's permission framework is the closest thing to a built-in answer and it stops short of this: has_perm('app.change_document') says the user may change documents, not that they may change this document, which is what object-level authorization has to decide.
Primary Defence: Always include ownership verification in database queries. Use patterns like Document.query.filter_by(id=doc_id, owner_id=current_user.id) (Flask), Document.objects.filter(id=doc_id, owner=request.user) (Django), or repository methods that combine resource ID with user ID filters. Never use bare get(id) or objects.get(id=x): the scope to the authenticated user belongs in the query itself.
Common Vulnerable Patterns
Direct Database Lookup Without Authorization Check
# VULNERABLE - No ownership verification in Flask
from flask import Flask, jsonify, request
from models import Document, db
@app.route('/api/document/<int:doc_id>')
def get_document(doc_id):
# Directly retrieves ANY document by ID
doc = Document.query.get_or_404(doc_id)
return jsonify({
'id': doc.id,
'title': doc.title,
'content': doc.content,
'owner_id': doc.owner_id
})
# Attack example:
# User A's document: GET /api/document/123 → Returns User A's data
# Attacker tries: GET /api/document/124 → Returns User B's data!
# Attacker tries: GET /api/document/125 → Returns User C's data!
# Sequential enumeration exposes ALL documents
Why this is vulnerable: Document.query.get_or_404(doc_id) retrieves any document matching the provided ID - nothing compares the document's owner_id with the authenticated user. An attacker changes the doc_id in the URL to reach any document in the database, and walking the IDs in sequence reaches all of them.
Django ORM Query Without Permission Check
# VULNERABLE - Missing authorization in Django view
from django.http import JsonResponse
from django.shortcuts import get_object_or_404
from .models import Order
def get_order(request, order_id):
# Retrieves order without checking if user owns it
order = get_object_or_404(Order, id=order_id)
return JsonResponse({
'order_id': order.id,
'total': str(order.total),
'items': [item.name for item in order.items.all()],
'shipping_address': order.shipping_address
})
# Attack example:
# User logged in, makes request: GET /orders/5001/
# User's own order ID is 5001, but tries: GET /orders/5002/
# Result: Returns another user's order with private shipping address!
Why this is vulnerable: Django's get_object_or_404(Order, id=order_id) returns the order if it exists, regardless of who owns it. Any authenticated user can reach another user's order details, including the shipping address and order total, by changing the order_id URL parameter.
FastAPI Endpoint with Missing Authorization
# VULNERABLE - Path parameter used directly without auth check
from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from . import models, database
@app.get("/api/users/{user_id}/profile")
async def get_user_profile(
user_id: int,
db: Session = Depends(database.get_db)
):
# No check if current user can access this profile
user = db.query(models.User).filter(
models.User.id == user_id
).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return {
"user_id": user.id,
"email": user.email, # PII exposure!
"full_name": user.full_name,
"phone": user.phone,
"ssn": user.ssn # Critical data leak!
}
# Attack example:
# Attacker enumerates: GET /api/users/1/profile → User 1's SSN
# GET /api/users/2/profile → User 2's SSN
# GET /api/users/3/profile → User 3's SSN
# Mass PII theft through sequential access
Why this is vulnerable: The endpoint queries the database for the user_id path parameter with no check comparing it against the authenticated user's ID. Any authenticated user can walk the whole user table by incrementing that parameter, and each response carries the SSN, email address and phone number the handler returns.
File Access Without Owner Validation
# VULNERABLE - file located by a caller-supplied row ID, with no ownership check
from flask import send_file, abort
from flask_login import login_required
@app.route('/download/<int:file_id>')
@login_required
def download_file(file_id):
# Primary key straight from the URL - the row is fetched for whoever
# asks, and only then handed to the file system
record = UploadedFile.query.get(file_id)
if record is None:
abort(404)
# No check that record.owner_id == current_user.id
return send_file(record.storage_path, as_attachment=True,
download_name=record.original_name)
# Attack example:
# User uploads their document, which is stored as file 123
# Attacker requests: GET /download/124
# Attacker requests: GET /download/125
# Result: downloads every other user's uploads by walking the ID
Why this is vulnerable: query.get(file_id) looks the row up by primary key alone. @login_required establishes that the caller is signed in, not that this file is theirs, so the only thing standing between an attacker and the whole upload table is knowing that IDs are sequential. Fix it the same way as any other row - UploadedFile.query.filter_by(id=file_id, owner_id=current_user.id).first() - so the file system is only reached for a row the caller could already have.
Two related mistakes are not CWE-566 and need their own fix. Taking the filename from the request rather than from the row makes the endpoint a path-traversal read (CWE-22) regardless of the ownership check. And serving the upload directory through the web server, or handing out long-lived pre-signed URLs, routes around this view entirely - an ownership check here is correct and irrelevant if the bytes are reachable without it.
Batch Operation Without Individual Authorization
# VULNERABLE - Bulk delete without per-item authorization
from django.views.decorators.http import require_http_methods
from django.http import JsonResponse
from .models import Document
import json
@require_http_methods(["DELETE"])
def bulk_delete_documents(request):
data = json.loads(request.body)
doc_ids = data.get('document_ids', [])
# Deletes ALL specified documents without ownership check!
deleted_count = Document.objects.filter(
id__in=doc_ids
).delete()[0]
return JsonResponse({'deleted': deleted_count})
# Attack example:
# POST /api/documents/bulk-delete
# Body: {"document_ids": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]}
# Result: Deletes ANY documents, not just attacker's documents!
# Attacker can delete entire database by enumerating IDs
Why this is vulnerable: Document.objects.filter(id__in=doc_ids).delete() deletes every document matching the provided IDs, because the query carries no ownership filter (owner=request.user). An attacker deletes any document in the system by listing its ID in the request, and a request that enumerates the ID space empties the table for every user.
Secure Patterns
Filter Query by Current User (Primary Pattern)
# SECURE - Scope query to authenticated user
from flask import Flask, jsonify, abort
from flask_login import login_required, current_user
from models import Document, db
@app.route('/api/document/<int:doc_id>')
@login_required
def get_document(doc_id):
# Filter by BOTH id AND owner_id
doc = Document.query.filter_by(
id=doc_id,
owner_id=current_user.id # Ensures user owns this document
).first()
if not doc:
# Return 404 for both non-existent and unauthorized
# Don't reveal whether document exists but user can't access it
abort(404, "Document not found")
return jsonify({
'id': doc.id,
'title': doc.title,
'content': doc.content
})
Why this works: With the ownership check (owner_id=current_user.id) in the query filter, the database returns only documents belonging to the authenticated user. An attacker who knows another user's document ID gets no rows back, so the escalation is stopped at the data access layer. Returning a 404 for both non-existent and unauthorized resources prevents information leakage about which document IDs exist in the system.
Explicit Authorization Check with Dedicated Function
# SECURE - Separate authorization logic for reusability
from flask import Flask, jsonify, abort
from flask_login import login_required, current_user
from models import Document, DocumentShare, db
from functools import wraps
from sqlalchemy import or_
# Permission name -> the share flag that grants it. Deny-by-default depends on
# this being a lookup: a chain of `if require_permission == ...` comparisons
# falls through to "granted" for any name it does not recognise, so a typo at
# the route lets a read-only share reach a mutating handler
PERMISSION_FLAGS = {
'read': 'can_read',
'write': 'can_write',
'delete': 'can_delete',
}
def authorize_document_access(require_permission='read'):
"""Decorator to verify user can access document"""
if require_permission not in PERMISSION_FLAGS:
# Raised at import, not at request time
raise ValueError(f"Unknown permission: {require_permission!r}")
def decorator(f):
@wraps(f)
def wrapper(doc_id, *args, **kwargs):
# One query settles visibility: owned, or shared with this
# caller carrying read. Fetching by id and only then looking for
# a share costs a denied request one more round trip than a
# missing id, so the two 404s stay distinguishable by response
# time even with identical status and body - measured at 2
# queries against 1 on SQLAlchemy 2.0
doc = db.session.execute(
db.select(Document).where(
Document.id == doc_id,
or_(
Document.owner_id == current_user.id,
Document.shared_access.any(
user_id=current_user.id, can_read=True
),
),
)
).scalar_one_or_none()
if doc is None:
# Missing, or not visible to this caller - one branch, so
# there is one response and nothing to tell the two apart
abort(404)
# Past the visibility floor the caller is already entitled to know
# this document exists, so the second query leaks nothing: what it
# decides is the operation, not existence
if doc.owner_id != current_user.id:
shared_access = db.session.execute(
db.select(DocumentShare).where(
DocumentShare.document_id == doc.id,
DocumentShare.user_id == current_user.id,
)
).scalar_one_or_none()
# The flag is looked up rather than compared against, so an
# unrecognised permission denies instead of falling through.
# shared_access cannot be None here - the query above returns
# a non-owned row only when it has a readable share - but
# denying on it keeps that assumption from becoming a 500
if shared_access is None or not getattr(
shared_access, PERMISSION_FLAGS[require_permission]
):
abort(403, f"Not authorized to {require_permission} this document")
# User is authorized, attach document to kwargs
kwargs['document'] = doc
return f(doc_id, *args, **kwargs)
return wrapper
return decorator
@app.route('/api/document/<int:doc_id>')
@login_required
@authorize_document_access(require_permission='read')
def get_document(doc_id, document=None):
return jsonify({
'id': document.id,
'title': document.title,
'content': document.content,
'is_owner': document.owner_id == current_user.id
})
@app.route('/api/document/<int:doc_id>', methods=['DELETE'])
@login_required
@authorize_document_access(require_permission='delete')
def delete_document(doc_id, document=None):
db.session.delete(document)
db.session.commit()
return jsonify({'message': 'Document deleted'}), 200
Why this works: The decorator gives every endpoint one authorization path, so a new route cannot acquire a weaker check by omission - it either carries the decorator and receives an authorized document, or it receives nothing. The permission argument keeps each route's requirement visible at the route rather than implied by the handler's name.
The two denial paths differ on purpose. Not visible at all - not owned, and no share carrying can_read - aborts 404, the same as an ID that was never issued, so the endpoint cannot be used to test which document IDs exist. Denied at the permission level - a read-only share on a DELETE - aborts 403, because the caller can already see the document and the status code tells them nothing new.
Note that can_read gates visibility rather than being one case among three, and that it does so inside the query. Checking only that a share row exists would grant read access that the flag explicitly denies.
Keeping visibility in the query is also what makes the two 404 answers cost the same. Fetching the document by id and only then looking for a share spends an extra round trip on the denial path, so a caller can separate "no such document" from "not yours" by response time even though the status and body match - which is the timing tell the Testing section below asks you to remove. Measured on SQLAlchemy 2.0, this shape answers a missing id, an unshared document and a write-without-read share in one query each, against two for the fetch-then-check version. The second query runs only once visibility is settled, where its cost reveals nothing the caller was not already entitled to know.
Django Class-Based Views with Permission Mixins
# SECURE - Django with built-in permission system
from django.views.generic import DetailView, UpdateView, DeleteView
from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
from django.http import Http404
from .models import Order
class UserOwnsOrderMixin(UserPassesTestMixin):
"""Mixin to ensure user owns the order"""
def test_func(self):
order = self.get_object()
# Check if current user owns this order
return order.user == self.request.user
def handle_no_permission(self):
# Http404, not PermissionDenied - the default redirect-to-login is
# wrong for an authenticated user, and a 403 would confirm the
# order exists
raise Http404
class OrderDetailView(LoginRequiredMixin, UserOwnsOrderMixin, DetailView):
model = Order
template_name = 'orders/detail.html'
context_object_name = 'order'
def get_queryset(self):
# Additional safety: scope queryset to current user
return Order.objects.filter(user=self.request.user)
class OrderUpdateView(LoginRequiredMixin, UserOwnsOrderMixin, UpdateView):
model = Order
fields = ['shipping_address', 'notes']
template_name = 'orders/edit.html'
def get_queryset(self):
# Only allow updating own orders
return Order.objects.filter(user=self.request.user)
class OrderDeleteView(LoginRequiredMixin, UserOwnsOrderMixin, DeleteView):
model = Order
success_url = '/orders/'
def get_queryset(self):
# Only allow deleting own orders
return Order.objects.filter(user=self.request.user)
Why this works: The get_queryset override is the load-bearing part. Scoping the queryset to self.request.user means get_object() cannot return another user's order at all, on every view that inherits it rather than on the ones somebody remembered. UserPassesTestMixin then covers the cases where a plain queryset filter is too blunt (shared orders, admin overrides), and handle_no_permission raises Http404 rather than the default PermissionDenied so a denial is indistinguishable from a missing record.
Trace the order of events, because it is not the one the code reads like. UserPassesTestMixin.dispatch calls test_func() before the view's get(), and test_func calls self.get_object() - so for another user's order the Http404 is raised inside the test, out of the scoped queryset, and handle_no_permission never runs at all. Two things follow. The denial is a 404 because of get_queryset, not because of handle_no_permission, so removing the queryset override changes the status code even though the mixin looks unchanged. And a subclass that widens get_queryset() - to add shared orders, say - moves the whole authorization decision into test_func and starts routing denials through handle_no_permission instead, without a line of either changing. Whenever the queryset stops being the check, confirm test_func still is.
FastAPI Dependency Injection for Authorization
# SECURE - FastAPI with dependency injection pattern
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from typing import Annotated
from . import models, database, auth
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: Session = Depends(database.get_db)
) -> models.User:
"""Verify JWT token and return current user"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
user_id = auth.decode_token(token)
if user_id is None:
raise credentials_exception
user = db.query(models.User).filter(models.User.id == user_id).first()
if user is None:
raise credentials_exception
return user
async def get_authorized_document(
doc_id: int,
current_user: Annotated[models.User, Depends(get_current_user)],
db: Session = Depends(database.get_db)
) -> models.Document:
"""Verify user can access document"""
doc = db.query(models.Document).filter(
models.Document.id == doc_id,
models.Document.owner_id == current_user.id # Authorization check
).first()
if not doc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Document not found"
)
return doc
@app.get("/api/documents/{doc_id}")
async def get_document(
document: Annotated[models.Document, Depends(get_authorized_document)]
):
"""Get document - authorization handled by dependency"""
return {
"id": document.id,
"title": document.title,
"content": document.content,
"created_at": document.created_at
}
@app.delete("/api/documents/{doc_id}")
async def delete_document(
document: Annotated[models.Document, Depends(get_authorized_document)],
db: Session = Depends(database.get_db)
):
"""Delete document - authorization handled by dependency"""
db.delete(document)
db.commit()
return {"message": "Document deleted successfully"}
Why this works: FastAPI resolves dependencies before the endpoint function runs, so get_authorized_document does the authentication (via get_current_user) and the authorization (the owner_id filter) first; if either fails it raises and the endpoint body never executes. Every endpoint declaring the same dependency gets the same check without restating it, while other entry points still need equivalent checks of their own.
Using UUIDs with Authorization (Defense in Depth)
# SECURE - UUIDs + Authorization checks
from flask import Flask, jsonify, abort
from flask_login import login_required, current_user
from datetime import datetime, timezone
from models import db
import uuid
class Document(db.Model):
# Use UUID instead of sequential integer
id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
owner_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text)
# datetime.utcnow is deprecated from Python 3.12 and returns a naive
# value. The column has to be timezone=True to keep the offset the
# aware default carries - on a naive column Postgres discards it
created_at = db.Column(
db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
@app.route('/api/document/<doc_id>')
@login_required
def get_document(doc_id):
# Validate UUID format first
try:
uuid.UUID(doc_id)
except ValueError:
abort(400, "Invalid document ID format")
# STILL perform authorization check - UUIDs are NOT sufficient!
doc = Document.query.filter_by(
id=doc_id,
owner_id=current_user.id
).first()
if not doc:
abort(404, "Document not found")
return jsonify({
'id': doc.id,
'title': doc.title,
'content': doc.content
})
# Example document IDs:
# Instead of: /api/document/123, /api/document/124, /api/document/125
# Use: /api/document/a3f7c8e1-4b9d-4e3a-8f7c-1234567890ab
# uuid.uuid4() creates high-entropy IDs that are hard to enumerate, but authorization is still required
Why this works: UUIDv4 values from uuid.uuid4() and other high-entropy opaque IDs make sequential enumeration computationally infeasible. Unlike sequential IDs (1, 2, 3, 4...), an attacker cannot guess valid random UUIDs by incrementing or pattern matching. However, UUIDs alone are NOT a security control - they only reduce enumeration risk. The code still includes explicit authorization checks (owner_id=current_user.id) because UUIDs can be exposed through logs, URLs shared between users, browser history, or other side channels. Ownership verification is what still refuses access once an ID is known.
Bulk Operations with Per-Item Authorization
# SECURE - Verify ownership for each item in bulk operation
from django.views.decorators.http import require_http_methods
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_protect
from django.contrib.auth.decorators import login_required
from .models import Document
import json
@login_required
@require_http_methods(["DELETE"])
@csrf_protect
def bulk_delete_documents(request):
data = json.loads(request.body)
doc_ids = data.get('document_ids', [])
if not doc_ids or len(doc_ids) > 100: # Cap the batch size
return JsonResponse(
{'error': 'Invalid request'},
status=400
)
# Only delete documents owned by current user
deleted_count = Document.objects.filter(
id__in=doc_ids,
owner=request.user # Authorization check!
).delete()[0]
# Return count of actually deleted documents
# May be less than requested if user didn't own all of them
return JsonResponse({
'deleted': deleted_count,
'requested': len(doc_ids)
})
# Alternative with per-ID feedback:
@login_required
@require_http_methods(["DELETE"])
@csrf_protect
def bulk_delete_documents_detailed(request):
data = json.loads(request.body)
requested_ids = data.get('document_ids', [])
# Reject an over-length batch rather than truncating it. Slicing to
# [:100] drops the rest silently, and the response - which reports a
# 'skipped' list - would then describe a batch the caller did not send
if not requested_ids or len(requested_ids) > 100:
return JsonResponse({'error': 'Invalid request'}, status=400)
# Ownership is part of the query, so another user's document is never
# loaded and cannot be told apart from an ID that does not exist
owned = Document.objects.filter(id__in=requested_ids, owner=request.user)
deleted_ids = list(owned.values_list('id', flat=True))
owned.delete()
# SECURE - one 'skipped' list. Separate 'not_found' and 'not_authorized'
# lists would let a single batch request enumerate the whole table
return JsonResponse({
'deleted': deleted_ids,
'skipped': [i for i in requested_ids if i not in set(deleted_ids)],
})
Why this works: Bulk operations are secured by including the ownership filter (owner=request.user) in the queryset used for deletion. The database only deletes documents that match both the ID list AND the ownership requirement, so another user's ID in the batch is a no-op rather than a deletion. The batch-size cap (maximum 100 items) bounds how many rows one request can touch - it is a limit on blast radius, not rate limiting, which belongs in front of the endpoint. Returning the deleted count against the requested count is safe because the difference does not say which IDs were rejected, or whether they existed. The per-ID variant is only safe because it reports one skipped list: splitting that into not_found and not_authorized would tell the caller which IDs exist, letting a single batch request enumerate the table that the ownership filter was added to protect.
Framework-Specific Guidance
Django - Permission-Based Access Control
Django provides a permission system to build the authorization on:
# SECURE - Django with permissions and object-level permissions
from django.contrib.auth.decorators import login_required
from django.contrib.auth.models import User
from django.db import models
from django.db.models import Q
from django.shortcuts import get_object_or_404
from django.http import JsonResponse
# Model with owner field
class Document(models.Model):
owner = models.ForeignKey(User, on_delete=models.CASCADE)
title = models.CharField(max_length=200)
content = models.TextField()
is_public = models.BooleanField(default=False)
# user_can_access() and the viewset's get_queryset() both read this,
# so it has to be a real field and not an assumed one
shared_with = models.ManyToManyField(
User, related_name='shared_documents', blank=True
)
class Meta:
permissions = [
("view_private_document", "Can view private documents"),
]
def user_can_access(self, user):
"""Object-level permission check"""
if self.owner == user:
return True
if self.is_public:
return True
# Check sharing permissions
return self.shared_with.filter(id=user.id).exists()
# View with permission checking
@login_required
def get_document(request, doc_id):
# The authorization is the queryset, not a check after it.
# get_object_or_404 takes a queryset as its first argument, so a plain
# function view can scope the lookup exactly as the viewset below does -
# and it raises Http404 from the same place whether the row is missing or
# not visible to this caller. .distinct() because the shared_with join
# can return the same document more than once
visible = Document.objects.filter(
Q(owner=request.user) | Q(is_public=True) | Q(shared_with=request.user)
).distinct()
doc = get_object_or_404(visible, id=doc_id)
return JsonResponse({
'id': doc.id,
'title': doc.title,
'content': doc.content,
'is_owner': doc.owner == request.user
})
# Django Rest Framework with object permissions
from rest_framework import viewsets, permissions
from rest_framework.exceptions import PermissionDenied
class IsOwnerOrReadOnly(permissions.BasePermission):
"""Object-level permission to only allow owners to edit"""
def has_object_permission(self, request, view, obj):
# Read permissions for GET, HEAD, OPTIONS
if request.method in permissions.SAFE_METHODS:
return obj.user_can_access(request.user)
# Write permissions only for owner
return obj.owner == request.user
class DocumentViewSet(viewsets.ModelViewSet):
queryset = Document.objects.all()
serializer_class = DocumentSerializer
permission_classes = [permissions.IsAuthenticated, IsOwnerOrReadOnly]
def get_queryset(self):
# Scope to documents user can access
user = self.request.user
return Document.objects.filter(
models.Q(owner=user) |
models.Q(is_public=True) |
models.Q(shared_with=user)
).distinct()
def perform_create(self, serializer):
# Automatically set owner to current user
serializer.save(owner=self.request.user)
Why this works: The user_can_access method puts object-level authorization on the model, so the same rule serves every view rather than being re-derived at each call site.
Both examples in this section scope the query, and they reach it differently
only because DRF hands a viewset a get_queryset() hook and a plain function
view has none. That is not a reason to fall back on fetch-then-check:
get_object_or_404 takes a queryset as its first argument, so the same
predicate goes in either way.
The fetch-then-check version is worth knowing as a shape to recognise rather
than to write, because its cost is not where you would look for it. It answers
404 for both cases and the bodies are genuinely identical - unlike Flask's
abort(404, "message"), Django's Http404 does not carry its message into the
response when DEBUG=False, and both paths render the same 179-byte page. What
differs is the work. Measured on this model, a missing id costs 1 query and
an existing-but-denied document costs 3, because user_can_access runs the
owner test and then the shared_with lookup only for rows that were found. The
status matches, the body matches, and the response time announces which of the
two happened - the timing tell the Testing section below asks you to remove.
Scoping the queryset makes it 1 query for every case, denied and missing alike.
Keep user_can_access on the model even so. The queryset decides visibility;
that method is what the DRF permission class calls to decide writes, and a
rule stated once on the model is one a later view can reuse instead of
re-deriving. In the DRF viewset, the get_queryset override is what actually stops the enumeration: a document outside that queryset is never retrieved, so DRF raises 404 before has_object_permission is reached and a denial is indistinguishable from a missing row. IsOwnerOrReadOnly then narrows what an accessible document may be used for - read for shared and public documents, write for the owner only.
That ordering is worth keeping in mind, because it is what makes DRF's default 403 harmless here. A viewset that leaves queryset = Document.objects.all() in place without the get_queryset override reaches has_object_permission for every ID in the table and answers 403 for the ones that exist, which hands back the enumeration the permission class was added to prevent.
Flask - Custom Authorization Decorators
Flask leaves authorization to you, and a decorator keeps it in one place:
# SECURE - Flask with comprehensive authorization system
from flask import Flask, abort, g, jsonify, request
from flask_login import LoginManager, login_required, current_user
from functools import wraps
from models import Document, DocumentShare, db
from sqlalchemy import or_
app = Flask(__name__)
login_manager = LoginManager(app)
def not_found():
"""The single response for every miss and every denial.
A per-model message on one branch and a bare abort(404) on the other
produce different response bodies for the same status code, which is
the existence oracle the 404 was chosen to close.
"""
abort(404, "Not found")
# Permission name -> the share flag that grants it, so an unrecognised
# permission has nothing to look up and is denied rather than falling through
PERMISSION_FLAGS = {
'read': 'can_read',
'write': 'can_write',
'delete': 'can_delete',
}
# Authorization decorator
def authorize_resource(model_class, param_name='id', permission='read'):
"""
Decorator to authorize access to a resource
Args:
model_class: SQLAlchemy model class
param_name: URL parameter name containing resource ID
permission: Required permission level ('read', 'write', 'delete')
"""
if permission not in PERMISSION_FLAGS:
# Raised at import, so a typo at the route is a startup failure
# rather than a mutating request that quietly succeeds
raise ValueError(f"Unknown permission: {permission!r}")
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
resource_id = kwargs.get(param_name)
# Visibility goes in the query, so a resource this caller may
# not see is never returned and a denial costs exactly what a
# missing id costs. Loading by primary key and checking afterwards
# spends an extra round trip on the denial path only, which is a
# timing tell between two responses that are otherwise identical.
# .any() accepts keywords, so this stays generic over model_class
visible = [model_class.owner_id == current_user.id]
if hasattr(model_class, 'shared_access'):
# can_read is the floor, and it is in the query: a share row
# granting write without read does not make this visible
visible.append(
model_class.shared_access.any(
user_id=current_user.id, can_read=True
)
)
resource = db.session.execute(
db.select(model_class).where(
model_class.id == resource_id, or_(*visible)
)
).scalar_one_or_none()
if resource is None:
not_found()
if resource.owner_id == current_user.id:
# Owner has all permissions
g.authorized_resource = resource
return f(*args, **kwargs)
# Shared, and already known to carry read. This second query
# decides the operation rather than existence, so its cost falls
# on a path where the caller may already know the resource exists
access = resource.shared_access.filter_by(
user_id=current_user.id
).first()
# Past the floor the caller can already see the resource, so a
# denial here reveals nothing new about existence - 403 is correct
if access is None or not getattr(access, PERMISSION_FLAGS[permission]):
abort(403, f"Not authorized to {permission} this resource")
g.authorized_resource = resource
return f(*args, **kwargs)
return wrapper
return decorator
# Usage in routes
@app.route('/api/document/<int:id>')
@login_required
@authorize_resource(Document, param_name='id', permission='read')
def get_document(id):
doc = g.authorized_resource
return jsonify({
'id': doc.id,
'title': doc.title,
'content': doc.content
})
@app.route('/api/document/<int:id>', methods=['PUT'])
@login_required
@authorize_resource(Document, param_name='id', permission='write')
def update_document(id):
doc = g.authorized_resource
data = request.get_json()
doc.title = data.get('title', doc.title)
doc.content = data.get('content', doc.content)
db.session.commit()
return jsonify({'message': 'Document updated'})
@app.route('/api/document/<int:id>', methods=['DELETE'])
@login_required
@authorize_resource(Document, param_name='id', permission='delete')
def delete_document(id):
doc = g.authorized_resource
db.session.delete(doc)
db.session.commit()
return jsonify({'message': 'Document deleted'})
Why this works: The decorator gives every endpoint one authorization path, so a route either carries it and receives an authorized resource in g, or it receives nothing. The permission argument keeps each route's requirement visible at the route rather than implied by the handler's name, and checking ownership before shared access means an owner never depends on a share row existing.
The permission is looked up in PERMISSION_FLAGS rather than compared against a chain of if permission == ... branches, which is what makes an unrecognised name deny. A chain has an implicit final branch - fall through everything and the handler runs - so permission='update' at the route would clear the read floor and then match nothing, and a read-only share would reach a mutating handler with a 200. Validating the argument in the decorator factory turns that typo into a startup ValueError instead of a request that quietly succeeds.
can_read gates visibility rather than being one permission among three, the same way it does in the authorize_document_access decorator above and in the FastAPI dependency below. Testing permission == 'write' and access.can_write on its own would admit a share row carrying write without read, and that caller would then learn the resource exists from a 200 on the update route - the floor is what makes the two denial paths mean different things. Past it, a 403 is correct: the caller can already see the resource, so refusing the operation tells them nothing new.
Both invisibility refusals go through not_found() on purpose. The status code is only half of what a caller can see: Flask renders abort(404, "Document not found") and a bare abort(404) into different bodies - the second carries Werkzeug's default "The requested URL was not found on the server" text - so a decorator that describes the missing case and stays silent on the denied one answers the enumeration question in the response body while looking uniform in the status line. One helper, one description, both branches.
FastAPI - Comprehensive Dependency Authorization
# SECURE - FastAPI with advanced dependency patterns
from fastapi import FastAPI, Depends, HTTPException, status, Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes
from sqlalchemy import and_, or_
from sqlalchemy.orm import Session
from typing import Annotated, Optional
from pydantic import BaseModel
from . import models, database, auth
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="token",
scopes={
"documents:read": "Read documents",
"documents:write": "Write documents",
"documents:delete": "Delete documents",
}
)
class DocumentUpdate(BaseModel):
"""Request body for the update route below.
Named explicitly rather than taking the raw body: a model with only the
fields a caller may set is what keeps `owner_id` out of reach, the same
concern as the JavaScript page's note on `doc.set(req.body)`
"""
title: str
content: str
class AuthorizationError(HTTPException):
def __init__(self, detail: str):
super().__init__(
status_code=status.HTTP_403_FORBIDDEN,
detail=detail
)
async def get_current_user(
security_scopes: SecurityScopes,
token: Annotated[str, Depends(oauth2_scheme)],
db: Session = Depends(database.get_db)
) -> models.User:
"""Authenticate user and verify scopes"""
authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": authenticate_value},
)
# Decode and verify token
payload = auth.decode_token(token)
if payload is None:
raise credentials_exception
user_id: int = payload.get("sub")
token_scopes: list = payload.get("scopes", [])
# Verify required scopes
for scope in security_scopes.scopes:
if scope not in token_scopes:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
headers={"WWW-Authenticate": authenticate_value},
)
user = db.query(models.User).filter(models.User.id == user_id).first()
if user is None:
raise credentials_exception
return user
class DocumentAccess:
"""Dependency resolving a document the caller is allowed to act on"""
def __init__(self, permission: str):
if permission not in ('read', 'write', 'delete'):
raise ValueError(f"Unknown permission: {permission}")
self.permission = permission
async def __call__(
self,
doc_id: int,
current_user: Annotated[models.User, Security(get_current_user)],
db: Session = Depends(database.get_db)
) -> models.Document:
# Visibility is part of the query: owned, or shared with this user.
# A document that is neither is never loaded, so the response is
# identical to the one for an ID that does not exist
row = (
db.query(models.Document, models.DocumentShare)
.outerjoin(
models.DocumentShare,
and_(
models.DocumentShare.document_id == models.Document.id,
models.DocumentShare.user_id == current_user.id,
# can_read is the visibility floor - a share without it
# does not entitle the caller to know the document exists
models.DocumentShare.can_read.is_(True),
),
)
.filter(
models.Document.id == doc_id,
or_(
models.Document.owner_id == current_user.id,
models.DocumentShare.id.isnot(None),
),
)
.first()
)
if row is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Document not found"
)
doc, share = row
# Owner holds every permission
if doc.owner_id == current_user.id:
return doc
# Shared. 403 is correct from here on: the caller is already entitled
# to know this document exists, so the status code reveals nothing new
if not getattr(share, f"can_{self.permission}"):
raise AuthorizationError(
f"Your access to this document does not include {self.permission}"
)
# Attach permission info to document
doc.user_permission = {
'can_read': share.can_read,
'can_write': share.can_write,
'can_delete': share.can_delete
}
return doc
# Usage with different permission requirements
@app.get("/api/documents/{doc_id}")
async def get_document(
document: Annotated[
models.Document,
Security(DocumentAccess('read'), scopes=["documents:read"])
]
):
"""Get document - owner, or a share carrying can_read"""
return {
"id": document.id,
"title": document.title,
"content": document.content,
"is_owner": not hasattr(document, 'user_permission')
}
@app.put("/api/documents/{doc_id}")
async def update_document(
document: Annotated[
models.Document,
Security(DocumentAccess('write'), scopes=["documents:write"])
],
data: DocumentUpdate,
db: Session = Depends(database.get_db)
):
"""Update document - owner, or a share carrying can_write"""
document.title = data.title
document.content = data.content
db.commit()
db.refresh(document)
return {"message": "Document updated", "document": document}
@app.delete("/api/documents/{doc_id}")
async def delete_document(
document: Annotated[
models.Document,
Security(DocumentAccess('delete'), scopes=["documents:delete"])
],
db: Session = Depends(database.get_db)
):
"""Delete document - owner, or a share carrying can_delete"""
db.delete(document)
db.commit()
return {"message": "Document deleted successfully"}
Why this works: The dependency resolves the document and the caller's authority in a single scoped query, so an endpoint annotated with it cannot be handed a document the caller has no claim to - the parameter is either an authorized Document or the request never reaches the function body. Naming the permission at the call site (DocumentAccess('write')) means each route states what it needs, and a new route cannot inherit a weaker check by omission.
The two denial paths are deliberately different, and the split is the part worth copying:
- Not visible at all - not owned, and no share carrying
can_read- is folded into the query's join andWHEREclause and answered 404, the same as an ID that was never issued. Fetching the document first and then deciding would answer 404 for a missing ID and 403 for someone else's, which is the enumeration oracle the ownership check exists to prevent. - Visible but not permitted at this level - a
can_readshare on aPUT- is answered 403. The caller can already see the document, so the status code tells them nothing about existence they did not already know, and a 404 here would be actively unhelpful.
The OAuth2 scope check in get_current_user sits in front of both and is also a legitimate 403: a missing documents:write scope is a property of the token, not of any particular document, so it cannot vary with whether a given ID exists.
Note where each flag is enforced. can_read sits in the join condition, because it decides whether the caller may know the document exists at all - a DocumentShare with can_read false is a real state in this model, and admitting that row into the query would leak existence through the 403. can_write and can_delete are checked afterwards with getattr(share, f"can_{self.permission}"), once visibility is already established. A model where a share can carry write without read has no sensible answer here; if that state is reachable, either forbid it with a constraint or decide explicitly which flag governs visibility.
Common Pitfalls
- Fixing a hand-written Django view with
get_object_or_404(Order, id=order_id, user=request.user), but leaving a DRFModelViewSet'sget_queryset()unoverridden - the auto-generated list/retrieve/update/destroy actions still queryOrder.objects.all()and remain unscoped even though the manually written view next to it is correct. - Setting
permission_classes = [IsAuthenticated]on a DRF viewset and treating that as the authorization fix.IsAuthenticatedonly confirms the caller is logged in; object-level authorization needshas_object_permissionon a custompermissions.BasePermission, and that method is only invoked for detail routes (retrieve/update/destroy) - alistaction still leaks every row unless the queryset itself is filtered. - Using a FastAPI
Depends(get_current_user)dependency and assuming dependency injection also enforces ownership.Depends()only injects the authenticated principal into the endpoint; the endpoint function still has to compareresource.owner_id == current_user.idexplicitly, since nothing about the dependency mechanism checks object-level access. - Adding
Document.query.filter_by(id=doc_id, owner_id=current_user.id)to the main Flask blueprint, while a Flask-Admin or internal-tooling blueprint mounted for support staff queries the same model directly through its own view functions without the same filter.
Testing
Authorization bugs are invisible to a scanner, because only the application knows which user is supposed to own which record. Every test below therefore needs at least two users and a resource belonging to one of them:
- A user can read, update and delete their own resource.
- User B's request for user A's resource ID is indistinguishable from a request for an ID that was never issued: same status, same body, no timing tell. What the fix has to remove is the difference between the two, so assert on the pair rather than on either one alone.
- 404 for both is the default and the easiest to keep true. A uniform 403 also passes, provided nothing in the handler answers 403 only when the record happens to exist - which is the usual way this regresses.
- Sequential IDs around a known-good one are not reachable.
- Bulk operations submitted with a mix of owned and unowned IDs affect only the owned ones, and do not partially apply before failing.
- Unauthenticated requests are rejected with 401.
- Malformed IDs (non-numeric, negative, oversized, null) return 400, not 500. A 500 usually means the ID reached the ORM before any check.
- Where sharing exists, a user granted read access cannot write.
Run these against the service or query layer as well as the view. A check that lives only in the view is bypassed by any other caller, including management commands and Celery tasks.