Initial commit: Imperva WAF demo lab
Flight-booking app scaffolded from the source running on the GCP host, plus the four-screen frontend and three OWASP attack endpoints (SQLi, reflected XSS, path traversal) built and locally verified against a live uvicorn/vite dev pair. Includes deploy.sh (local build -> ship over ssh -> swap on host) and docs/ carried over from the handoff.
This commit is contained in:
commit
4d60fdd139
28 files changed
+2609
No files matched your search
+131
@@ -0,0 +1,131 @@
|
||||
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
|
||||
}
|
||||
Reference in new issue
Block a user