Snippet Quota Security

Welcome to the lesson on Flawed Business Logic in Snippet Quota Management. In this lesson, we will explore how business logic flaws can lead to security vulnerabilities in web applications. Business logic is crucial, as it dictates how an application behaves and processes data. When flawed, it can open doors to various security issues that are often missed by automated security scanners.

In this lesson, we'll focus on snippet quota management, a common feature in web applications, and learn how to identify and fix vulnerabilities related to it. Let's dive in! 🚀

Business logic represents the custom rules and processes that govern how an application operates. It's the code that enforces the "rules of the business." Unlike common vulnerabilities like SQL Injection or Cross-Site Scripting, which often have well-known patterns, business logic flaws are unique to each application's specific features and requirements.

When implementing features like snippet management, business logic determines crucial aspects such as:

• Who can create snippets?
• How many snippets a user can create (a numerical limit).
• What the size limitations for individual snippets are.
• How the total storage quota is calculated and enforced for each user.

Oversights in these rules can lead to serious security vulnerabilities. For instance, if we don't validate storage quotas, a malicious user could potentially exhaust the server's storage capacity by uploading an unlimited amount of data. This can cause a Denial of Service (DoS) for other users, or even incur significant financial costs in a cloud environment. In the following sections, we'll examine a specific example of flawed business logic in snippet management and learn how to properly secure it.

Let's look at a simple function that demonstrates one such vulnerability next.

Let's examine code that demonstrates how the absence of size checks and user quota limits can lead to vulnerabilities. The following example uses Python with the FastAPI framework. The code defines an endpoint that allows an authenticated user to create a new code snippet.

# VULNERABLE: no size checks or per-user quotas
@router.post("")
async def create_snippet(request: Request, db: AsyncSession = Depends(get_db)):
    # Step 1: Get the current authenticated user.
    authorization = request.headers.get('authorization')
    user = await get_current_user(authorization, db)
    
    # Step 2: Get the snippet data from the request body.
    data = await request.json()
    
    # Step 3: Create a new Snippet object with the user's data.
    # Notice there are no checks on the size of data['content'].
    snippet = Snippet(
        id=str(uuid.uuid4()),
        title=data['title'],
        content=data['content'],
        language=data['language'],
        userId=user.id
    )
    
    # Step 4: Save the new snippet to the database.
    db.add(snippet)
    await db.commit()
    await db.refresh(snippet)
    
    return {
        "id": snippet.id,
        "title": snippet.title,
        "content": snippet.content,
        "language": snippet.language,
        "userId": str(snippet.userId)
    }

In this code, the application allows users to create snippets without any validation on the size of the content field. Furthermore, it doesn't check the total storage already consumed by the user. This oversight means a single user can create an unlimited number of snippets of any size, which can be exploited to overload the system's storage.

To understand the impact of this vulnerability, let's see how an attacker might exploit it. An attacker can write a simple script to automate the process of creating snippets, quickly filling up the server's disk space. This is a form of resource exhaustion attack.

Here's a Python script that demonstrates this attack by repeatedly sending requests to create new, moderately large snippets:

#!/usr/bin/env python3
import requests
import sys

BASE_URL = "http://localhost:3000"
API_URL = f"{BASE_URL}/api"

def create_spam_snippets(token):
    """Create a massive number of snippets to exhaust storage"""
    print("Starting spam attack...")
    
    headers = {"Authorization": f"Bearer {token}"}
    
    # Loop 1000 times to create 1000 snippets.
    for i in range(1, 1001):
        # Each snippet's content is a 10,000-character string.
        payload = {
            "title": f"spam-{i}",
            "content": "A" * 10000,
            "language": "plaintext"
        }
        
        try:
            # Send the POST request to the vulnerable endpoint.
            response = requests.post(
                f"{API_URL}/snippets",
                json=payload,
                headers=headers
            )
            if i % 100 == 0:
                print(f"Created {i} snippets...")
        except Exception as e:
            print(f"Error at snippet {i}: {e}")
            sys.exit(1)
    
    print("Spam completed")

# Using default user for demo purposes.    
def get_auth_token():
    """Login to get authentication token"""
    response = requests.post(
        f"{API_URL}/auth/login",
        json={"username": "admin", "password": "codesignal"}
    )
    
    if response.status_code != 200:
        print(f"Login failed: {response.text}")
        sys.exit(1)
    
    return response.json()["token"]    

if __name__ == "__main__":
    token = get_auth_token()
    create_spam_snippets(token)

This script first authenticates to get a valid token. Then, it enters a loop that sends 1,000 requests to the server. Each request creates a new snippet with 10,000 characters. Without size checks or quota limits, the server's storage can quickly become overwhelmed, leading to potential downtime or data loss for all users.

Another approach is to create a single, extremely large snippet. An attacker could modify the script to send just one request with a content field containing millions of characters, demonstrating how the lack of a per-snippet size check allows a user to exhaust a large amount of storage with just one malicious request.

The first step in securing our business logic is to enforce rules on individual actions. In this case, we need to implement a size check to ensure that each snippet does not exceed a reasonable limit. This prevents the "single large snippet" attack vector. In Python, you can check the size of a string in bytes using len(content.encode('utf-8')), which correctly handles multi-byte characters.

Here's how you can add a maximum snippet size check to the create_snippet function:

MAX_SNIPPET_SIZE = 1 * 1024 * 1024  # 1MB

@router.post("")
async def create_snippet(request: Request, db: AsyncSession = Depends(get_db)):
    authorization = request.headers.get('authorization')
    user = await get_current_user(authorization, db)
    
    data = await request.json()
    title = data['title']
    content = data['content']
    language = data['language']
    
    # Check snippet size before creating the object.
    content_bytes = len(content.encode('utf-8'))
    if content_bytes > MAX_SNIPPET_SIZE:
        # If the content is too large, reject the request with a 413 status code.
        raise HTTPException(status_code=413, detail="Snippet too large")
    
    snippet = Snippet(
        id=str(uuid.uuid4()),
        title=title,
        content=content,
        language=language,
        userId=user.id
    )
    
    db.add(snippet)
    await db.commit()
    await db.refresh(snippet)
    
    return {
        "id": snippet.id,
        "title": snippet.title,
        "content": snippet.content,
        "language": snippet.language,
        "userId": str(snippet.userId)
    }

In this updated code, we define a MAX_SNIPPET_SIZE constant of 1MB. Before creating the Snippet object, we calculate the byte size of the incoming content. If it exceeds our defined limit, we immediately stop processing and return a 413 Payload Too Large error response. This effectively prevents the creation of oversized snippets.

While application-level size checks are essential for enforcing business rules, another crucial layer of protection is implementing request size limits at the web server or framework level. This acts as a first line of defense, rejecting excessively large requests before they even reach your application logic. This can save server resources like CPU and memory that would otherwise be used to process the large request.

For example, in FastAPI (and Starlette), you can configure the Uvicorn server to set a maximum request size:

# This command tells Uvicorn to reject any request with a body larger than 1MB.
uvicorn main:app --limit-max-request 1048576  # 1MB

Other frameworks have similar configurations. In Flask, you can set the MAX_CONTENT_LENGTH configuration variable:

app = Flask(__name__)
# Flask will reject requests with a content length greater than 1MB.
app.config['MAX_CONTENT_LENGTH'] = 1 * 1024 * 1024  # 1MB

In Django, you can set the DATA_UPLOAD_MAX_MEMORY_SIZE in your settings file:

# Django's setting to limit the size of in-memory request bodies.
DATA_UPLOAD_MAX_MEMORY_SIZE = 1048576  # 1MB

Setting a global request size limit is a robust, low-level defense. It works in tandem with your per-snippet size checks. The server-level limit blocks grossly oversized requests, while your route-level checks enforce the specific business rules for your application's features.

Implementing a per-snippet size limit is a good start, but it doesn't solve the "death by a thousand cuts" problem where an attacker creates many small, valid-sized snippets to exhaust storage. To prevent this, we must implement a total storage quota for each user. This involves calculating the user's current storage usage and preventing new snippet creation if the quota would be exceeded.

Here's how you can add a user quota check to the create_snippet function:

MAX_SNIPPET_SIZE = 1 * 1024 * 1024      # 1MB
USER_QUOTA_BYTES = 10 * 1024 * 1024     # 10MB

@router.post("")
async def create_snippet(request: Request, db: AsyncSession = Depends(get_db)):
    authorization = request.headers.get('authorization')
    user = await get_current_user(authorization, db)
    
    data = await request.json()
    title = data['title']
    content = data['content']
    language = data['language']
    
    # 1. Check individual snippet size.
    content_bytes = len(content.encode('utf-8'))
    if content_bytes > MAX_SNIPPET_SIZE:
        raise HTTPException(status_code=413, detail="Snippet too large")
    
    # 2. Check if the new snippet will exceed the user's total quota.
    # First, get all existing snippets for the user.
    result = await db.execute(select(Snippet).where(Snippet.userId == user.id))
    existing_snippets = result.scalars().all()
    # Then, calculate their total size.
    current_usage = sum(len(s.content.encode('utf-8')) for s in existing_snippets)
    
    # If current usage plus the new snippet's size exceeds the quota, reject it.
    if current_usage + content_bytes > USER_QUOTA_BYTES:
        raise HTTPException(status_code=413, detail="Storage quota exceeded")
    
    snippet = Snippet(
        id=str(uuid.uuid4()),
        title=title,
        content=content,
        language=language,
        userId=user.id
    )
    
    db.add(snippet)
    await db.commit()
    await db.refresh(snippet)
    
    return {
        "id": snippet.id,
        "title": snippet.title,
        "content": snippet.content,
        "language": snippet.language,
        "userId": str(snippet.userId)
    }

In this final, secured version, we've added a USER_QUOTA_BYTES constant (10MB in this example). After validating the new snippet's size, we perform a second check:

1. We query the database to retrieve all existing snippets for the current user.
2. We calculate the current_usage by summing the byte sizes of all their snippets.
3. We check if this current_usage plus the size of the new snippet (content_bytes) would exceed the USER_QUOTA_BYTES.
4. If it would, we return a 413 error, preventing the user from exceeding their allocated storage.

This multi-layered approach—combining server-level request limits, per-snippet size validation, and per-user total storage quotas—provides a robust defense against resource exhaustion attacks.

While our implementation effectively prevents resource exhaustion attacks, the quota checking mechanism has an important limitation: it queries and sums all user snippets on every create operation. This O(n) approach works fine for learning and small-scale applications, but won't scale to production workloads where users might have thousands of snippets.

For production systems, consider these scalable alternatives:

Denormalized Counter (Recommended): Add a storage_used field to the User model that tracks total bytes. Increment it when creating snippets and decrement when deleting them. This makes quota checks O(1) constant-time operations:

# In User model
storage_used = Column(Integer, default=0)

# In create_snippet
if user.storage_used + content_bytes > USER_QUOTA_BYTES:
    raise HTTPException(status_code=413, detail="Storage quota exceeded")
user.storage_used += content_bytes

Database Aggregation: Use SQL SUM() aggregations pushed down to the database level rather than fetching all records into application memory.

Background Jobs: Calculate quotas asynchronously in background tasks and cache the results, refreshing periodically.

For this course, we'll continue using the direct summation approach as it's easier to understand and sufficient for learning the security principles.

In this lesson, we explored the importance of business logic in web applications and how flaws in snippet quota management can lead to serious vulnerabilities. We learned how to identify these flaws, write an exploit script to demonstrate the impact, and implement effective, multi-layered solutions to mitigate them.

As you move on to the practice exercises, remember the key points from this lesson: always validate inputs against your business rules, enforce limits on both individual actions and aggregate usage, and consider defenses at multiple levels of your application stack. Keep up the great work! 🎉