Tagg

Posted on Apr 3

15 Security Practices I Applied to My FastAPI Side Project

#security #api #python #webdev

⚠️ Disclaimer
These are practical measures I applied to my side project.
For enterprise-grade security, consult professionals.

Why This Post?

I recently launched a K-Beauty Cosmetic Ingredients API on RapidAPI. Before going live, I spent significant time hardening security.

This post shares 15 practical security measures I implemented, with:

✅ Real code examples
✅ OWASP Top 10 mapping
✅ Lessons learned

Whether you're deploying on RapidAPI, AWS, or anywhere else — these patterns apply.

Authentication & Access Control
Server Hardening
Database Security
Monitoring & Alerting
Resilience & Error Handling
OWASP Top 10 Mapping
Lessons Learned

1. Authentication & Access Control

1.1 RapidAPI Proxy Secret Validation

RapidAPI acts as a proxy. But what stops someone from calling your server directly?

Proxy Secret — a shared secret that only RapidAPI knows.

from fastapi import Request, HTTPException

RAPIDAPI_PROXY_SECRET = os.environ.get("RAPIDAPI_PROXY_SECRET")

@app.middleware("http")
async def verify_rapidapi_proxy(request: Request, call_next):
    # Skip for health checks
    if request.url.path == "/health":
        return await call_next(request)

    # Verify secret
    proxy_secret = request.headers.get("X-RapidAPI-Proxy-Secret")
    if proxy_secret != RAPIDAPI_PROXY_SECRET:
        raise HTTPException(status_code=403, detail="Forbidden")

    return await call_next(request)

🔐 OWASP A01: Broken Access Control — Direct access blocked.

1.2 Tier-Based Access Control

Different subscription tiers get different features:

@app.middleware("http")
async def extract_tier(request: Request, call_next):
    tier_header = request.headers.get("X-RapidAPI-Subscription", "BASIC")
    request.state.tier = tier_header.upper()
    return await call_next(request)

# In endpoint:
def search_partial(request: Request, q: str):
    if request.state.tier == "BASIC":
        raise HTTPException(
            status_code=403, 
            detail="Partial search requires PRO or ULTRA tier"
        )
    # ... continue

Tier	Features
BASIC	Exact search only
PRO	+ Partial search, more fields
ULTRA	+ All fields, highest limits

1.3 Rate Limiting with SlowAPI

Prevent abuse and ensure fair usage:

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/v1/ingredient/inci")
@limiter.limit("60/minute")
def search_by_inci(request: Request, q: str):
    # ...

🔐 OWASP A07: Identification and Authentication Failures — Brute force prevention.

2. Server Hardening

2.1 Docker Non-Root User

Never run containers as root:

# Create non-root user
RUN groupadd -r apigroup && useradd -r -g apigroup apiuser

# Set ownership
COPY --chown=apiuser:apigroup . /app

# Switch to non-root user
USER apiuser

CMD ["gunicorn", "main:app", "-w", "2", "-k", "uvicorn.workers.UvicornWorker"]

🔐 OWASP A04: Insecure Design — Principle of least privilege.

2.2 Sensitive File Permissions

.env files should only be readable by owner:

chmod 600 config/.env

-rw-------  1 ubuntu ubuntu  .env    # Only owner can read/write

🔐 OWASP A05: Security Misconfiguration — Sensitive data protected.

2.3 Environment Separation

Different configs for different environments:

ENV = os.environ.get("ENV", "dev")

if ENV == "prod":
    DEBUG = False
    LOG_LEVEL = "WARNING"
else:
    DEBUG = True
    LOG_LEVEL = "DEBUG"

2.4 HSTS Header

Force HTTPS connections:

@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    return response

🔐 OWASP A02: Cryptographic Failures — Encrypted transport enforced.

3. Database Security

3.1 OS-Level Read-Only Mode

API only reads data. Why allow writes?

Instead of relying on SQLite's PRAGMA query_only, I used OS-level read-only mode via URI parameter — a more robust approach:

def get_db():
    # file: URI with ?mode=ro enforces read-only at OS level
    conn = sqlite3.connect(
        f"file:{DB_PATH}?mode=ro",
        uri=True,
        timeout=5.0
    )
    return conn

This is stronger than PRAGMA query_only because:

OS-level enforcement — even if SQLite has a bug, the OS blocks writes
Cannot be bypassed — no way to disable via SQL commands

🔐 OWASP A08: Software and Data Integrity Failures — No unauthorized modifications.

3.2 Parameterized Queries

Never concatenate user input into SQL:

# ❌ BAD - SQL Injection vulnerable
cursor.execute(f"SELECT * FROM ingredients WHERE name = '{user_input}'")

# ✅ GOOD - Parameterized
cursor.execute("SELECT * FROM ingredients WHERE name = ?", (user_input,))

🔐 OWASP A03: Injection — SQL injection prevented.

3.3 LIKE Wildcard Escaping

User input in LIKE queries can abuse wildcards (%, _):

def escape_like_wildcards(value: str) -> str:
    """Escape LIKE wildcards to prevent injection."""
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

# Usage
escaped_q = escape_like_wildcards(user_input)
cursor.execute(
    "SELECT * FROM ingredients WHERE inci_name LIKE ? ESCAPE '\\\\'",
    (f"{escaped_q}%",)
)

3.4 Query Timeout

Prevent long-running queries from blocking:

conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True, timeout=5.0)

4. Monitoring & Alerting

4.1 Telegram Alerts on Server Start

Know immediately when your server restarts:

def alert_server_started(version: str, env: str, db_count: int):
    message = f"""
🚀 <b>API Server Started</b>
├ Version: {version}
├ Environment: {env}
├ DB Records: {db_count:,}
└ Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
    send_telegram_message(message)

4.2 Real-Time Error Alerts

500 errors go straight to Telegram:

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    alert_error(
        endpoint=request.url.path,
        error_type=type(exc).__name__,
        error_message=str(exc)[:500]
    )
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"}
    )

🔐 OWASP A09: Security Logging and Monitoring Failures — Real-time visibility.

4.3 Preventing Duplicate Alerts

Multiple Gunicorn workers = multiple startup alerts? Not anymore:

lock_file = os.path.join(LOG_DIR, ".startup_alert_lock")

try:
    # Clean up stale lock files (server restart case)
    # Race Condition defense: ignore if another worker deleted it first
    try:
        if os.path.exists(lock_file):
            file_age = time.time() - os.path.getmtime(lock_file)
            if file_age > 10:
                os.remove(lock_file)
    except OSError:
        pass  # Another worker deleted it first — continue

    # Atomic file creation - only first worker succeeds
    fd = os.open(lock_file, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    os.close(fd)
    alert_server_started(version, env, count)  # First worker sends
except FileExistsError:
    pass  # Other workers skip

This handles:

TOCTOU (Time-of-check to time-of-use) — Atomic O_EXCL operation
Race conditions — OSError catch for concurrent access

4.4 Structured JSON Logging

Logs that are easy to parse and analyze:

import logging
import json

def log_api_request(endpoint, query, status_code, response_time_ms):
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "endpoint": endpoint,
        "query": query,
        "status_code": status_code,
        "response_time_ms": response_time_ms
    }
    logger.info(json.dumps(log_entry))

5. Resilience & Error Handling

5.1 Graceful Logging Failures

Log failures shouldn't crash your API:

async def safe_log(request, endpoint, query, status_code):
    try:
        await asyncio.to_thread(
            log_api_request, request, endpoint, query, status_code
        )
    except Exception:
        pass  # Log failure doesn't affect API response

5.2 Telegram Retry Logic

Network is unreliable. Retry with backoff:

from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter

retry_strategy = Retry(
    total=3,
    backoff_factor=0.5,           # 0.5s, 1s, 2s
    status_forcelist=[429, 500, 502, 503, 504],
    respect_retry_after_header=True  # Honor 429 Retry-After
)

session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=retry_strategy))

5.3 Input Sanitization for Alerts

User input in Telegram messages? Escape it:

import html

def alert_error(endpoint: str, error_message: str):
    safe_message = html.escape(error_message)[:4000]  # Telegram limit
    send_telegram_message(f"Error at {endpoint}: {safe_message}")

5.4 Docker Auto-Restart

Container crashes? Auto-recover:

docker run -d \
  --restart=always \
  --name myapi \
  myapi:latest

5.5 Health Check Endpoint

For load balancers and monitoring:

@app.get("/health")
def health_check():
    return {
        "status": "healthy",
        "version": "2.3.0",
        "timestamp": datetime.utcnow().isoformat()
    }

HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

6. OWASP Top 10 Mapping

Here's how our security measures map to OWASP Top 10 (2021):

OWASP	Risk	Our Mitigation
A01	Broken Access Control	✅ Proxy Secret, Tier-based access
A02	Cryptographic Failures	✅ HSTS header, HTTPS enforced
A03	Injection	✅ Parameterized SQL, LIKE escaping
A04	Insecure Design	✅ Non-root Docker, read-only DB
A05	Security Misconfiguration	✅ .env permissions, env separation
A06	Vulnerable Components	⚠️ Regular dependency updates needed
A07	Auth Failures	✅ Rate limiting, secret validation
A08	Data Integrity	✅ OS-level read-only database
A09	Logging Failures	✅ JSON logging, Telegram alerts
A10	SSRF	➖ N/A (no external requests)

Coverage: 9/10 categories addressed ✅

7. Lessons Learned

1. Security is Layers

No single measure is enough. Defense in depth:

Network → Proxy Secret
Application → Rate limiting, input validation
Data → Read-only, parameterized queries
Infrastructure → Non-root, auto-restart
Monitoring → Real-time alerts

2. Fail Gracefully

Security logging shouldn't break your API. Wrap everything in try-except.

3. Assume Breach

Design as if attackers will get in:

Read-only database (can't modify)
Non-root container (limited damage)
Monitoring (you'll know quickly)

4. Automate Everything

Auto-restart on crash
Auto-retry on network failure
Auto-alert on errors

5. Test Your Security

Before launch:

# Test direct access (should fail)
curl http://your-server:8000/v1/stats
# Expected: 403 Forbidden

# Test with valid secret (should work)
curl -H "X-RapidAPI-Proxy-Secret: your-secret" http://your-server:8000/v1/stats
# Expected: 200 OK

Wrapping Up

Security isn't a feature — it's a foundation. These 15 measures took extra time upfront, but give me confidence that the API can handle production traffic safely.

Resources:

Try the API

If you're curious about the API itself:

🔗 K-Beauty Cosmetic Ingredients API

📂 GitHub - Example Code

Questions about any of these security measures? Drop a comment below!

DEV Community