DEV Community

Deepika Pusala
Deepika Pusala

Posted on

Instagram Clone Backend — Part 4: Security & Authentication (Bearer Tokens, Argon2id, Enumeration Defense, Actor Extraction & Input Protection)

#backend #python #security #tutorial

Security is the most critical foundation of any social media application. In an Instagram clone, our backend handles sensitive credentials, private photos, and personal interactions. A single security flaw can expose millions of accounts to credential stuffing, session hijacking, or account enumeration attacks.

In this tutorial, we will explore the production-grade security architecture we implemented in our backend, covering:

  1. Modern Password Hashing with Argon2id (OWASP recommended)
  2. Opaque Bearer Token Sessions with SHA-256 Storage Hashing
  3. Account Enumeration Defense via Dummy Verifications
  4. Actor Extraction with FastAPI Dependencies
  5. Input Protection & Header Sanitization

1. Password Hashing: Why Argon2id?

For years, developers relied on MD5, SHA-256, or bcrypt for password hashing. Today, Argon2id (the winner of the Password Hashing Competition) is the gold standard recommended by OWASP.

Why Argon2id?

  • Memory-hard & CPU-intensive: Unlike SHA-256, Argon2id is deliberately expensive to compute, rendering brute-force attacks using GPUs and dedicated ASICs economically infeasible.
  • Hybrid resistance: Argon2id combines Argon2d (resists GPU cracking) and Argon2i (resists side-channel and cache-timing attacks).

The Async Thread-Pool Challenge

Because Argon2id is intentionally CPU-intensive, running it directly inside an async function would block Python's event loop, stalling all other users!

To solve this, we offload the hashing computation to a background worker thread using asyncio.to_thread:

import asyncio
from argon2 import PasswordHasher

# Configured according to OWASP guidelines:
# 64 MB memory cost, 3 iterations, 4 parallel threads
_ph = PasswordHasher(
    time_cost=3,
    memory_cost=65536,
    parallelism=4,
    hash_len=32,
    salt_len=16,
)

async def hash_password(password: str) -> str:
    """Hashes password in a worker thread to keep the event loop non-blocking."""
    return await asyncio.to_thread(_ph.hash, password)

async def verify_password(password: str, hashed_password: str) -> bool:
    """Verifies password safely without stalling concurrent server requests."""
    try:
        await asyncio.to_thread(_ph.verify, hashed_password, password)
        return True
    except Exception:
        return False
Enter fullscreen mode Exit fullscreen mode

2. Timing Attack Mitigation & Account Enumeration Defense

Have you ever noticed that poorly built websites say:

  • "User does not exist" when entering an incorrect email, but
  • "Incorrect password" when entering a valid email?

This allows attackers to compile lists of registered users—a vulnerability called Account Enumeration.

Even if you return a generic message like "Invalid credentials", an attacker can measure the server response time:

  • Non-existent user: Returns in ~5ms (DB lookup fails immediately).
  • Existing user: Takes ~80ms (DB lookup succeeds, then expensive Argon2id hash verification runs).

By measuring response latency, attackers know which accounts exist!

The Solution: Dummy Password Verification

In our backend, if a username or email is not found in the database, we intentionally execute a dummy hash verification against a pre-computed hash. This burns the exact same amount of CPU time as a real verification:

# Pre-computed dummy hash created at server boot
_DUMMY_HASH = _ph.hash("dummy_password_for_timing_mitigation_123")

async def dummy_verify_password(password: str) -> None:
    """
    Executes a dummy hash verification to burn approximately equal CPU time
    when a user account is not found, mitigating timing enumeration attacks.
    """
    try:
        await asyncio.to_thread(_ph.verify, _DUMMY_HASH, password)
    except Exception:
        pass
Enter fullscreen mode Exit fullscreen mode

The Unified Login Handler: 

async def authenticate_user(db: AsyncSession, username_or_email: str, password: str):
    user = await get_user_by_identifier(db, username_or_email)

    if not user:
        # User not found: run dummy verification to equalize response timing
        await dummy_verify_password(password)
        raise HTTPException(status_code=401, detail="Invalid username or password")

    # User exists: verify against actual stored hash
    is_valid = await verify_password(password, user.hashed_password)
    if not is_valid:
        raise HTTPException(status_code=401, detail="Invalid username or password")

    return user
Enter fullscreen mode Exit fullscreen mode

Whether the user exists or not, the endpoint takes the exact same time to respond!

3. High-Entropy Bearer Tokens & SHA-256 Storage Hashing

For session management, we use cryptographically secure opaque Bearer tokens.

The Golden Security Rule: Never Store Plain Tokens in the Database

If an attacker manages to obtain a read-only SQL injection or a database backup dump, they could use raw stored session tokens to impersonate all logged-in users.

To prevent this, we follow the same pattern used by Git and GitHub:

  1. Raw Token: Sent only once to the client client-side upon login.
  2. Token Hash: Only the deterministic SHA-256 hash of the token is stored in the database. 
[ Client / Browser ]                           [ Database ]
        |                                            |
        |---- Login Credentials -------------------->|
        |<--- Raw Token: sf_abc123... (Never stored)-|
        |                                            |
        |                                 Stores: sha256(raw_token)
        |                                 "e3b0c44298fc1c149afbf4c8..."
        |
        |---- Authorization: Bearer sf_abc123... --->|
                                          Hashes incoming raw token
                                          Matches stored hash in DB!
Enter fullscreen mode Exit fullscreen mode

Token Generation Implementation: 

import hashlib
import secrets
from typing import Tuple

def hash_token(raw_token: str) -> str:
    """Computes SHA-256 hash (64 hex characters) for database lookup."""
    return hashlib.sha256(raw_token.strip().encode("utf-8")).hexdigest()

def generate_session_token() -> Tuple[str, str]:
    """
    Generates a cryptographically random session token (256-bit entropy).
    Returns (raw_token, token_hash).
    """
    # High-entropy random token with prefix
    raw_token = f"sf_{secrets.token_urlsafe(32)}"
    token_hash = hash_token(raw_token)
    return raw_token, token_hash
Enter fullscreen mode Exit fullscreen mode

4. Actor Extraction via Dependency Injection

In an Instagram backend, every protected endpoint needs to know: "Who is making this request?" (The Actor).

Instead of parsing headers and validating database sessions manually in every endpoint, FastAPI provides an elegant Actor Extraction pattern using Depends:

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

security_scheme = HTTPBearer(auto_error=False)

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security_scheme),
    db: AsyncSession = Depends(get_db)
) -> User:
    """
    Extracts, hashes, and validates Bearer token against active DB sessions.
    Returns the authenticated User actor model.
    """
    if not credentials or credentials.scheme.lower() != "bearer":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required"
        )

    # 1. Compute SHA-256 of the presented bearer token
    token_hash = hash_token(credentials.credentials)

    # 2. Look up the active session in the database
    query = (
        select(User)
        .join(Session, Session.user_id == User.id)
        .where(
            Session.token_hash == token_hash,
            Session.is_revoked == False,
            Session.expires_at > func.now()
        )
    )
    result = await db.execute(query)
    user = result.scalar_one_or_none()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Session expired or invalid"
        )

    return user
Enter fullscreen mode Exit fullscreen mode

How Endpoints Consume the Authenticated Actor:

Notice how clean the route handler becomes:

@router.post("/posts")
async def create_new_post(
    post_data: CreatePostSchema,
    current_user: User = Depends(get_current_user),  # Actor extracted!
    db: AsyncSession = Depends(get_db)
):
    # current_user is guaranteed to be authenticated and valid
    post = Post(
        caption=post_data.caption,
        image_url=post_data.image_url,
        user_id=current_user.id
    )
    db.add(post)
    await db.commit()
    return {"message": "Post published", "post_id": post.id}
Enter fullscreen mode Exit fullscreen mode

5. Input Protection & Header Sanitization

Security is not just about passwords; it includes sanitizing every piece of metadata that enters the system.

1. Request ID Sanitization

Every request is tracked across logs using a unique X-Request-Id. If clients send their own request ID, we must sanitize it with a strict Regex whitelist to prevent log-injection or header-injection attacks:

import re
import uuid

# Only allow safe alphanumeric and punctuation characters (1 to 128 characters)
REQUEST_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_\-\.:]{1,128}$")

def sanitize_or_generate_request_id(incoming_header: str | None) -> str:
    """Validates client header against safe ASCII pattern or generates UUID4."""
    if incoming_header and REQUEST_ID_PATTERN.match(incoming_header):
        return incoming_header
    return str(uuid.uuid4())
Enter fullscreen mode Exit fullscreen mode

2. Pydantic Schema Bounds

Every incoming payload is strictly validated using Pydantic:

  • Post captions are limited to 2,200 characters (matching Instagram's limit).
  • Usernames are restricted to alphanumeric characters and underscores (^[a-zA-Z0-9_.]+$).
  • Passwords require minimum length and complexity.

Summary of Security Architecture

Security Threat Countermeasure In Our Backend
GPU/ASIC Brute-Force Argon2id memory-hard password hashing (64MB, 3 iterations).
Event-Loop Blocking Worker threads via asyncio.to_thread for CPU-heavy crypto operations.
Timing & User Enumeration dummy_verify_password equalizes execution timing on 401 errors.
DB Session Leaks Opaque sf_ tokens; only SHA-256 hashes stored in database.
Tampered Requests Type-safe Pydantic input schemas and strict Header Regex sanitization.
Session Tracking Clean get_current_user dependency injection for actor resolution.

In the next post, we will explore API Architecture & Lifecycle: Middleware, Context Variables, and CORS!

Top comments (0)