imperva-demo/backend/main.py

import os
import sqlite3
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import PlainTextResponse
from pydantic import BaseModel

app = FastAPI(title="Imperva Demo Leaky API", version="1.0.0")

DB_FILE = "demo.db"
BOARDING_PASS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "boarding_passes")

FLIGHTS = [
    {"id": "OUT101", "origin": "Earth (JFK)", "destination": "Lunar Gateway", "depart": "2026-08-14 09:00", "price": 128000},
    {"id": "OUT202", "origin": "Earth (LAX)", "destination": "Mars Colony Alpha", "depart": "2026-09-02 03:30", "price": 890000},
    {"id": "OUT303", "origin": "Earth (LHR)", "destination": "Europa Research Station", "depart": "2026-10-21 14:15", "price": 2450000},
    {"id": "OUT404", "origin": "Lunar Gateway", "destination": "Mars Colony Alpha", "depart": "2026-08-30 22:00", "price": 610000},
    {"id": "OUT505", "origin": "Earth (NRT)", "destination": "Titan Outpost", "depart": "2026-11-11 06:45", "price": 3980000},
]

def init_db():
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT,
            email TEXT,
            role TEXT
        )
    """)
    cursor.execute("SELECT COUNT(*) FROM users")
    if cursor.fetchone()[0] == 0:
        cursor.executemany("""
            INSERT INTO users (username, email, role) VALUES (?, ?, ?)
        """, [
            ("admin", "admin@flytoouter.space", "administrator"),
            ("jdoe", "jdoe@flytoouter.space", "engineer"),
            ("alice", "alice@flytoouter.space", "analyst")
        ])
    conn.commit()
    conn.close()

init_db()

class LoginRequest(BaseModel):
    username: str
    password: str

@app.get("/api/v1/health")
def health_check():
    """Endpoint for uptime and monitoring checks."""
    return {"status": "ok", "service": "imperva-demo-backend"}

@app.post("/api/v1/login")
def login(credentials: LoginRequest):
    """
    Intentionally lacks rate limiting or lockout protection.
    Demo Use: Showcases Imperva Advanced Bot Protection / Credential Stuffing mitigation.
    """
    if credentials.username == "admin" and credentials.password == "Secret123!":
        return {"status": "authenticated", "token": "demo-jwt-token-xyz789"}
    return {"status": "failed", "message": "Invalid credentials"}

@app.get("/api/v1/users")
def search_users(q: str = Query("", description="Search username")):
    """
    INTENTIONAL VULNERABILITY: Raw string concatenation for SQL query.
    Demo Use: Triggers Imperva default SQL Injection (SQLi) detection rules when passing payloads like `' OR '1'='1`.
    """
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    query = f"SELECT id, username, email, role FROM users WHERE username LIKE '%{q}%'"
    try:
        cursor.execute(query)
        rows = cursor.fetchall()
        conn.close()
        return [{"id": r[0], "username": r[1], "email": r[2], "role": r[3]} for r in rows]
    except Exception as e:
        conn.close()
        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")

@app.get("/api/v1/search")
def search_flights(term: str = Query("", description="Destination search term")):
    """
    INTENTIONAL VULNERABILITY: the search term is echoed back verbatim in
    `query_echo` and the frontend renders it unescaped (v-html).
    Demo Use: Triggers Imperva default Cross-Site Scripting (XSS) detection rules
    when passing payloads like `<script>alert(1)</script>`.
    """
    term_lower = term.lower()
    results = [
        f for f in FLIGHTS
        if term_lower in f["origin"].lower() or term_lower in f["destination"].lower()
    ] if term else FLIGHTS

    return {
        "query_echo": f"Showing results for: {term}" if term else "Showing all flights",
        "results": results,
    }


@app.get("/api/v1/download")
def download_boarding_pass(file: str = Query(..., description="Boarding pass filename")):
    """
    INTENTIONAL VULNERABILITY: raw path concatenation, no traversal sanitization.
    Demo Use: Triggers Imperva default Path Traversal / Illegal Resource Access
    detection rules when passing payloads like `../../../../etc/passwd`.
    """
    path = os.path.join(BOARDING_PASS_DIR, file)
    try:
        with open(path, "r") as f:
            content = f.read()
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail="Boarding pass not found")
    except IsADirectoryError:
        raise HTTPException(status_code=400, detail="Not a file")
    return PlainTextResponse(content)


@app.get("/api/v1/hidden-config")
def hidden_config():
    """
    Unauthenticated endpoint exposing internal application state.
    Demo Use: API Discovery & custom URI blocking rules.
    """
    return {
        "env": "production-demo",
        "db_connection": "sqlite:///demo.db",
        "internal_api_key": "sk_live_demo_998877665544332211",
        "debug_mode": True
    }