Anti Bot Review Protection

Introduction

Welcome to the third lesson of the "A04: Insecure Design" course! In this lesson, we'll explore the importance of anti-bot verification in review submission systems. Bots can be used to submit fake reviews, which can significantly impact businesses and user trust. Our goal is to understand how the absence of anti-bot measures can lead to vulnerabilities and to learn how to implement effective defenses to protect against such attacks. Let's dive in! 🤖

Understanding the Vulnerability

Insecure design refers to a category of weaknesses that stem from missing or ineffective security controls. It's not about a bug in the implementation but a flaw in the design itself—a failure to think like an attacker and anticipate how a feature could be misused. A review submission system without anti-bot protection is a classic example. The system is designed with the "happy path" in mind, assuming that only legitimate users will submit genuine reviews. This design fails to account for the "unhappy path" where automated scripts, or bots, are used to abuse the system. Review bombing and fake review submissions have become significant global issues affecting businesses across various platforms. For instance, in 2021, Amazon reported removing over 200 million suspected fake reviews before they were seen by customers. These automated attacks can severely impact businesses, with studies showing that a one-star difference in ratings can affect revenue by 5-9%. Consider an e-commerce platform where competitors use bots to submit thousands of negative reviews for a product. This not only misleads potential customers but also manipulates the platform's recommendation algorithms. Real-world examples include cases where restaurants have been targeted by competitors using bots to lower their ratings on popular review platforms, leading to significant business losses. The core vulnerability is the system's implicit trust in all incoming requests, which we will now examine in code.

Vulnerable Code

Here's an example of a vulnerable review submission endpoint that lacks any form of protection against automated attacks, written using FastAPI. This code correctly validates the format of the incoming data but fails to validate the source or intent of the request. Python# VULNERABLE: No rate limiting or CAPTCHA verification @router.post("/add-review") async def create_review(request: Request): data = await request.json() product_id = data.get("productId") rating = data.get("rating") comment = data.get("comment") if product_id is None or rating is None or comment is None: raise HTTPException(status_code=400, detail="Missing fields") # Pretend to persist review (omitted); vulnerability is lack of bot controls return {"productId": product_id, "rating": rating, "comment": comment} This endpoint accepts review submissions without any mechanisms to distinguish a human user from a bot. It doesn't check how many reviews have been submitted from one source, nor does it require any proof of humanity (like a CAPTCHA). An attacker could easily create a script to flood this endpoint with fake reviews, potentially overwhelming the system and compromising the integrity of the review system.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Implementing Rate Limiting

To defend against automated attacks, we'll first implement rate limiting. The theory behind rate limiting is to enforce a policy on the frequency of requests from a single source, typically identified by its IP address. While determined attackers can use proxies to cycle through different IPs, this still serves as an effective first line of defense against simple scripts.

Here is an example using a simple in-memory approach in FastAPI to limit review submissions from a single IP address.

Python

# Simple in-memory rate limiter per IP for demo purposes
WINDOW_SECONDS = 24 * 60 * 60
MAX_REQUESTS = 10
_requests = defaultdict(list)

@router.post("/add-review")
async def create_review(request: Request):
    # Identify the user by their IP address
    ip = request.client.host if request.client else "unknown"
    now = time()
    
    # Remove timestamps older than our window
    _requests[ip] = [t for t in _requests[ip] if now - t < WINDOW_SECONDS]
    
    # Block the request if the limit is exceeded
    if len(_requests[ip]) >= MAX_REQUESTS:
        raise HTTPException(status_code=429, detail="Too many reviews submitted. Try tomorrow.")
    
    # Record the timestamp of the current request
    _requests[ip].append(now)

    data = await request.json()
    product_id = data.get("productId")
    rating = data.get("rating")
    comment = data.get("comment")
    # ... (rest of the logic)

# Simple in-memory rate limiter per IP for demo purposes
WINDOW_SECONDS = 24 * 60 * 60
MAX_REQUESTS = 10
_requests = defaultdict(list)

@router.post("/add-review")
async def create_review(request: Request):
    # Identify the user by their IP address
    ip = request.client.host if request.client else "unknown"
    now = time()
    
    # Remove timestamps older than our window
    _requests[ip] = [t for t in _requests[ip] if now - t < WINDOW_SECONDS]
    
    # Block the request if the limit is exceeded
    if len(_requests[ip]) >= MAX_REQUESTS:
        raise HTTPException(status_code=429, detail="Too many reviews submitted. Try tomorrow.")
    
    # Record the timestamp of the current request
    _requests[ip].append(now)

    data = await request.json()
    product_id = data.get("productId")
    rating = data.get("rating")
    comment = data.get("comment")
    # ... (rest of the logic)

In this secured code, we track request timestamps for each IP address. We allow a maximum of 5 review submissions per IP within a 24-hour window.

First, we get the client's IP address.
Next, we clean up our list of timestamps for that IP, keeping only those within the last 24 hours (our WINDOW_SECONDS).
Then, we check if the number of recent requests is at or above our MAX_REQUESTS limit. If it is, we reject the request with an HTTP 429 Too Many Requests status code.
If the request is allowed, we record its timestamp. This simple mechanism effectively mitigates the curl script from the previous section, as the script would be blocked after the first five requests.

Implementing Email Verification

While rate limiting helps control the frequency of attacks, it doesn't prevent a determined attacker from creating multiple accounts. This is where email verification becomes a critical second layer of defense. By requiring users to verify their email addresses before they can submit reviews, we significantly raise the barrier for creating fake accounts at scale.

Why Email Verification Matters for Anti-Bot Protection

Email verification serves several crucial purposes in preventing bot attacks:

Account Creation Friction: Creating fake accounts requires access to valid email addresses. While free email services exist, creating hundreds or thousands of accounts becomes time-consuming and detectable.
Traceability: Verified email addresses provide a way to trace abusive behavior back to specific accounts and potentially identify patterns of abuse.
Cost Barrier: Many email providers have their own anti-bot measures, rate limits, and CAPTCHA requirements, adding layers of difficulty for attackers.
Deterrence: The verification step signals to potential abusers that the platform takes security seriously and has measures in place to detect and prevent fake accounts.

Database Model for Email Verification

To implement email verification, we need to extend our User model to track verification status. Here's what the model looks like:

Python

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String, unique=True, nullable=False)
    password = Column(String, nullable=False)
    role = Column(Enum(Role), default=Role.USER)
    email = Column(String, unique=True, nullable=False)
    email_verified = Column(Boolean, default=False)  # Key field for verification
    
    snippets = relationship("Snippet", back_populates="author")
    reviews = relationship("Review", back_populates="author")

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String, unique=True, nullable=False)
    password = Column(String, nullable=False)
    role = Column(Enum(Role), default=Role.USER)
    email = Column(String, unique=True, nullable=False)
    email_verified = Column(Boolean, default=False)  # Key field for verification
    
    snippets = relationship("Snippet", back_populates="author")
    reviews = relationship("Review", back_populates="author")

The email_verified field is a boolean flag that defaults to False when a user registers. This flag is only set to True after the user successfully completes the email verification process.

Checking Email Verification in Review Submission

Now that we have the ability to track verification status, we can enforce this requirement when users attempt to submit reviews. This check should happen early in the request processing, right after authentication:

Python

@router.post("/add-review")
async def create_review(
    request: Request,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    # Rate limiting logic (from previous section)...
    
    # Check if user's email is verified
    if not current_user.email_verified:
        raise HTTPException(
            status_code=403, 
            detail="Email must be verified to submit reviews"
        )
    
    # Continue with review submission...

@router.post("/add-review")
async def create_review(
    request: Request,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    # Rate limiting logic (from previous section)...
    
    # Check if user's email is verified
    if not current_user.email_verified:
        raise HTTPException(
            status_code=403, 
            detail="Email must be verified to submit reviews"
        )
    
    # Continue with review submission...

This verification check happens after rate limiting but before accepting the review data. By returning a 403 Forbidden status code, we clearly communicate that the user is authenticated but lacks the necessary permissions due to unverified status.

Implementing the Email Verification Endpoint

To allow users to verify their email addresses, we need to create a verification endpoint. In a production system, this would typically involve sending a verification token via email, but for learning purposes, we'll create a simplified endpoint:

Python

@router.post("/verify-email")
async def verify_email(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    """Verify the current user's email address"""
    # In production, this would validate a token sent via email
    current_user.email_verified = True
    await db.commit()
    
    return {
        "message": "Email verified successfully",
        "username": current_user.username
    }

@router.post("/verify-email")
async def verify_email(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    """Verify the current user's email address"""
    # In production, this would validate a token sent via email
    current_user.email_verified = True
    await db.commit()
    
    return {
        "message": "Email verified successfully",
        "username": current_user.username
    }

In a real-world application, this endpoint would:

Accept a verification token as a parameter
Validate that the token is valid, unexpired, and matches the user's email
Only then set email_verified to True

The simplified version shown here focuses on demonstrating the verification mechanism without the complexity of token generation and email sending, which we covered in the password recovery lesson.

How Email Verification Complements Rate Limiting

Email verification and rate limiting work together to create a robust defense:

Rate limiting prevents a single source (IP address) from overwhelming the system with requests
Email verification prevents attackers from easily creating multiple accounts to bypass IP-based rate limits

An attacker would need to:

Create multiple user accounts (time-consuming)
Verify the email for each account (requires access to multiple email addresses)
Distribute their attack across these accounts
Still be limited by per-IP rate limits

This multi-layered approach makes attacks significantly more difficult and expensive to execute.

Implementing Additional Verification Checks

With rate limiting and email verification in place, we can add another layer of protection by implementing checks that validate the legitimacy of the request itself. This includes verifying that the user is human and that their comment isn't obvious spam.

Below is an example of how you might add a CAPTCHA check and a simple spam filter to the endpoint.

Python

async def verify_captcha(token: str) -> bool:
    # Placeholder for CAPTCHA verification with a provider like hCaptcha or reCAPTCHA
    # In a real app, this would make an API call to the provider's server
    return token == "dev-ok"

def is_spam(text: str) -> bool:
    # Placeholder for a simple spam check. Real spam filters are far more complex.
    return isinstance(text, str) and len(text) > 0 and text.count("http") > 2

@router.post("/add-review")
async def create_review(request: Request):
    # --- Rate Limiting Logic (from previous step) ---
    ip = request.client.host if request.client else "unknown"
    now = time()
    _requests[ip] = [t for t in _requests[ip] if now - t < WINDOW_SECONDS]
    if len(_requests[ip]) >= MAX_REQUESTS:
        raise HTTPException(status_code=429, detail="Too many reviews submitted. Try tomorrow.")
    _requests[ip].append(now)

    # --- Email Verification Check (from previous step) ---
    if not current_user.email_verified:
        raise HTTPException(status_code=403, detail="Email must be verified to submit reviews")

    # --- Additional Verification Logic ---
    data = await request.json()
    product_id = data.get("productId")
    rating = data.get("rating")
    comment = data.get("comment")
    captcha_token = data.get("captchaToken")

    # 1. CAPTCHA Verification
    if not await verify_captcha(captcha_token or ""):
        raise HTTPException(status_code=400, detail="Invalid CAPTCHA")

    # 2. Spam Detection
    if is_spam(comment):
        raise HTTPException(status_code=400, detail="Comment detected as spam")

    # 3. Basic Data Validation
    if not isinstance(rating, int) or rating < 1 or rating > 5:
        raise HTTPException(status_code=400, detail="Rating must be between 1 and 5")

    # Pretend to persist review (omitted); focus is on design controls
    return {"productId": product_id, "rating": rating, "comment": comment, "status": "ok"}

async def verify_captcha(token: str) -> bool:
    # Placeholder for CAPTCHA verification with a provider like hCaptcha or reCAPTCHA
    # In a real app, this would make an API call to the provider's server
    return token == "dev-ok"

def is_spam(text: str) -> bool:
    # Placeholder for a simple spam check. Real spam filters are far more complex.
    return isinstance(text, str) and len(text) > 0 and text.count("http") > 2

@router.post("/add-review")
async def create_review(request: Request):
    # --- Rate Limiting Logic (from previous step) ---
    ip = request.client.host if request.client else "unknown"
    now = time()
    _requests[ip] = [t for t in _requests[ip] if now - t < WINDOW_SECONDS]
    if len(_requests[ip]) >= MAX_REQUESTS:
        raise HTTPException(status_code=429, detail="Too many reviews submitted. Try tomorrow.")
    _requests[ip].append(now)

    # --- Email Verification Check (from previous step) ---
    if not current_user.email_verified:
        raise HTTPException(status_code=403, detail="Email must be verified to submit reviews")

    # --- Additional Verification Logic ---
    data = await request.json()
    product_id = data.get("productId")
    rating = data.get("rating")
    comment = data.get("comment")
    captcha_token = data.get("captchaToken")

    # 1. CAPTCHA Verification
    if not await verify_captcha(captcha_token or ""):
        raise HTTPException(status_code=400, detail="Invalid CAPTCHA")

    # 2. Spam Detection
    if is_spam(comment):
        raise HTTPException(status_code=400, detail="Comment detected as spam")

    # 3. Basic Data Validation
    if not isinstance(rating, int) or rating < 1 or rating > 5:
        raise HTTPException(status_code=400, detail="Rating must be between 1 and 5")

    # Pretend to persist review (omitted); focus is on design controls
    return {"productId": product_id, "rating": rating, "comment": comment, "status": "ok"}

Here, we've added several new checks that run after the rate limiter and email verification:

CAPTCHA Verification: We call a (placeholder) function to verify a CAPTCHA token submitted with the request. This helps ensure the user is human.
Spam Detection: We use a simple function to check if the comment contains an excessive number of links, a common characteristic of spam.
Data Validation: We ensure the rating is a valid integer within the expected range.

These checks work together to create a much more secure design.

Introduction

Understanding the Vulnerability

Vulnerable Code

Join the 1M+ learners on CodeSignal

Exploiting the Vulnerability

Defensive Measures

Implementing Rate Limiting

Implementing Email Verification

Implementing Additional Verification Checks

Combining Anti-Bot Measures

Conclusion and Next Steps