API নিরাপত্তা ও OWASP API Top 10

mkdir -p ~/projects/api-security
cd ~/projects/api-security
uv init --bare -q
uv add -q "fastapi[standard]" pyjwt httpx2
uv add -q --dev pytest
# with pip: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]" pyjwt httpx2 pytest

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.testclient import TestClient


def current_user(x_user: Annotated[str | None, Header()] = None) -> str:
    if x_user is None:
        raise HTTPException(401, "Not authenticated")
    return x_user


User = Annotated[str, Depends(current_user)]


def show(label, resp):
    print(f"{label:11} {resp.status_code} {resp.json()}")

NOTES = {1: {"owner": "nadia", "text": "Q3 budget"}, 2: {"owner": "rafi", "text": "salary table"}}


def make_notes_app(safe: bool) -> FastAPI:
    app = FastAPI()

    @app.get("/notes/{note_id}")
    def read_note(note_id: int, user: User):
        note = NOTES.get(note_id)
        if note is None or (safe and note["owner"] != user):
            raise HTTPException(404, "Note not found")
        return {"id": note_id, "text": note["text"]}

    return app


for safe in (False, True):
    client = TestClient(make_notes_app(safe))
    show("fixed" if safe else "vulnerable", client.get("/notes/2", headers={"X-User": "nadia"}))

vulnerable  200 {'id': 2, 'text': 'salary table'}
fixed       404 {'detail': 'Note not found'}

import jwt

SECRET = "tutorial-only-secret-change-me-0123456789"      # from the environment in real code
FAR = 4102444800                                           # 2100-01-01, keeps the output stable


def make_auth_app(safe: bool) -> FastAPI:
    app = FastAPI()

    def subject(authorization: Annotated[str | None, Header()] = None) -> str:
        if not authorization or not authorization.startswith("Bearer "):
            raise HTTPException(401, "Missing bearer token")
        token = authorization[7:]
        try:
            if safe:
                claims = jwt.decode(token, SECRET, algorithms=["HS256"], issuer="bitbyte-api",
                                    options={"require": ["exp", "sub", "iss"]})
            else:
                claims = jwt.decode(token, options={"verify_signature": False})
        except jwt.InvalidTokenError:
            raise HTTPException(401, "Invalid token") from None
        return claims["sub"]

    @app.get("/me")
    def me(sub: Annotated[str, Depends(subject)]):
        return {"user": sub}

    return app


good = {"sub": "nadia", "iss": "bitbyte-api", "exp": FAR}
tokens = {
    "genuine": jwt.encode(good, SECRET, algorithm="HS256"),
    "forged": jwt.encode({**good, "sub": "admin"}, "attacker-key-attacker-key-attacker-key-1", algorithm="HS256"),
    "expired": jwt.encode({**good, "exp": 1}, SECRET, algorithm="HS256"),
}
for safe in (False, True):
    client = TestClient(make_auth_app(safe))
    for name, token in tokens.items():
        r = client.get("/me", headers={"Authorization": f"Bearer {token}"})
        print("fixed     " if safe else "vulnerable", f"{name:8}", r.status_code, r.json())

vulnerable genuine  200 {'user': 'nadia'}
vulnerable forged   200 {'user': 'admin'}
vulnerable expired  200 {'user': 'nadia'}
fixed      genuine  200 {'user': 'nadia'}
fixed      forged   401 {'detail': 'Invalid token'}
fixed      expired  401 {'detail': 'Invalid token'}

from pydantic import BaseModel, ConfigDict


class ProfilePatch(BaseModel):
    model_config = ConfigDict(extra="forbid")          # unknown fields are an error, not ignored
    display_name: str | None = None


class ProfileOut(BaseModel):
    id: int
    display_name: str


def make_profile_app(safe: bool) -> FastAPI:
    app = FastAPI()
    row = {"id": 1, "display_name": "Nadia", "is_admin": False, "password_hash": "$argon2id$v=19$..."}

    if safe:
        @app.get("/profile", response_model=ProfileOut)
        def read():
            return row

        @app.patch("/profile", response_model=ProfileOut)
        def update(body: ProfilePatch):
            row.update(body.model_dump(exclude_unset=True))
            return row
    else:
        @app.get("/profile")
        def read():
            return row

        @app.patch("/profile")
        def update(body: dict):
            row.update(body)
            return row

    return app


for safe in (False, True):
    client = TestClient(make_profile_app(safe))
    label = "fixed" if safe else "vulnerable"
    show(label, client.get("/profile"))
    r = client.patch("/profile", json={"is_admin": True})
    print(f"{label:11} PATCH is_admin ->", r.status_code, r.json().get("is_admin", r.json().get("detail", [{}])[0].get("type")))

vulnerable  200 {'id': 1, 'display_name': 'Nadia', 'is_admin': False, 'password_hash': '$argon2id$v=19$...'}
vulnerable  PATCH is_admin -> 200 True
fixed       200 {'id': 1, 'display_name': 'Nadia'}
fixed       PATCH is_admin -> 422 extra_forbidden

from fastapi import Request
from fastapi.responses import JSONResponse
from pydantic import Field

MAX_BODY = 10_000


class LooseJob(BaseModel):
    text: str
    max_tokens: int = 256


class StrictJob(BaseModel):
    text: str = Field(min_length=1, max_length=2000)
    max_tokens: int = Field(256, ge=1, le=1024)


def make_llm_app(safe: bool) -> FastAPI:
    app = FastAPI()
    Job = StrictJob if safe else LooseJob

    if safe:
        @app.middleware("http")
        async def limit_body(request: Request, call_next):
            if int(request.headers.get("content-length", 0)) > MAX_BODY:
                return JSONResponse({"detail": "Body too large"}, status_code=413)
            return await call_next(request)

    @app.post("/summarise")
    def summarise(job: Job):
        return {"estimated_tokens": len(job.text) // 4 + job.max_tokens}

    return app


for safe in (False, True):
    client = TestClient(make_llm_app(safe))
    label = "fixed" if safe else "vulnerable"
    r = client.post("/summarise", json={"text": "x" * 5000, "max_tokens": 1_000_000})
    print(f"{label:11} big request ->", r.status_code, str(r.json())[:60])
    r = client.post("/summarise", json={"text": "x" * 20_000})
    print(f"{label:11} huge body   ->", r.status_code, str(r.json())[:60])

vulnerable  big request -> 200 {'estimated_tokens': 1001250}
vulnerable  huge body   -> 200 {'estimated_tokens': 5256}
fixed       big request -> 422 {'detail': [{'type': 'string_too_long', 'loc': ['body', 'tex
fixed       huge body   -> 413 {'detail': 'Body too large'}

ROLES = {"nadia": "user", "root": "admin"}


def require_admin(user: User):
    if ROLES.get(user) != "admin":
        raise HTTPException(403, "Admin only")


def make_admin_app(safe: bool) -> FastAPI:
    app = FastAPI()

    @app.delete("/admin/users/{uid}", dependencies=[Depends(require_admin)] if safe else [])
    def delete_user(uid: int):
        return {"deleted": uid}

    return app


for safe in (False, True):
    client = TestClient(make_admin_app(safe))
    for who in ("nadia", "root"):
        show(("fixed" if safe else "vulnerable") + f"/{who}", client.delete("/admin/users/9", headers={"X-User": who}))

vulnerable/nadia 200 {'deleted': 9}
vulnerable/root 200 {'deleted': 9}
fixed/nadia 403 {'detail': 'Admin only'}
fixed/root  200 {'deleted': 9}

import threading


def make_bonus_app(safe: bool) -> FastAPI:
    app = FastAPI()
    claimed, balances, lock = set(), {}, threading.Lock()

    @app.post("/credits/claim")
    def claim(user: User):
        with lock:
            if safe and user in claimed:
                raise HTTPException(409, "Bonus already claimed")
            claimed.add(user)
            balances[user] = balances.get(user, 0) + 50
            return {"balance": balances[user]}

    return app


for safe in (False, True):
    client = TestClient(make_bonus_app(safe))
    codes = [client.post("/credits/claim", headers={"X-User": "bot-1"}) for _ in range(3)]
    print("fixed     " if safe else "vulnerable", [r.status_code for r in codes], codes[-1].json())

vulnerable [200, 200, 200] {'balance': 150}
fixed      [200, 409, 409] {'detail': 'Bonus already claimed'}

import ipaddress
import socket
from urllib.parse import urlsplit


class UnsafeURL(ValueError):
    pass


def check_url(url: str) -> str:
    """Return the public IP address to connect to, or raise UnsafeURL."""
    parts = urlsplit(url)
    if parts.scheme not in {"http", "https"}:
        raise UnsafeURL("scheme not allowed")
    if parts.username or parts.password:
        raise UnsafeURL("credentials in URL")
    if not parts.hostname:
        raise UnsafeURL("no host")
    try:
        infos = socket.getaddrinfo(parts.hostname, parts.port or 443, type=socket.SOCK_STREAM)
    except socket.gaierror:
        raise UnsafeURL("host does not resolve") from None
    for info in infos:
        ip = ipaddress.ip_address(info[4][0].split("%")[0])
        if ip.version == 6 and ip.ipv4_mapped:
            ip = ip.ipv4_mapped                    # ::ffff:127.0.0.1 is really 127.0.0.1
        if not ip.is_global or ip.is_multicast:        # multicast 224.0.0.1 counts as global in Python
            raise UnsafeURL(f"address {ip} is not public")
    return infos[0][4][0]


for url in [
    "https://8.8.8.8/doc",
    "http://169.254.169.254/latest/meta-data/",
    "http://2130706433/admin",
    "http://127.0.0.1/",
    "http://10.0.0.5/",
    "http://[::1]/",
    "http://[::ffff:127.0.0.1]/",
    "file:///etc/passwd",
]:
    try:
        print("ok      ", url, "->", check_url(url))
    except UnsafeURL as exc:
        print("BLOCKED ", url, "->", exc)

ok       https://8.8.8.8/doc -> 8.8.8.8
BLOCKED  http://169.254.169.254/latest/meta-data/ -> address 169.254.169.254 is not public
BLOCKED  http://2130706433/admin -> address 127.0.0.1 is not public
BLOCKED  http://127.0.0.1/ -> address 127.0.0.1 is not public
BLOCKED  http://10.0.0.5/ -> address 10.0.0.5 is not public
BLOCKED  http://[::1]/ -> address ::1 is not public
BLOCKED  http://[::ffff:127.0.0.1]/ -> address 127.0.0.1 is not public
BLOCKED  file:///etc/passwd -> scheme not allowed

from urllib.parse import urljoin

import httpx2 as httpx


def fetch_public(url: str, client: httpx.Client, max_hops: int = 3) -> str:
    for _ in range(max_hops):
        check_url(url)                                          # every hop is checked again
        response = client.get(url, follow_redirects=False, timeout=5)
        if response.is_redirect:
            url = urljoin(url, response.headers["location"])
            continue
        return response.text[:100_000]                          # cap what we keep
    raise UnsafeURL("too many redirects")


def upstream(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/start":
        return httpx.Response(302, headers={"location": "http://169.254.169.254/latest/meta-data/"})
    return httpx.Response(200, text="hello")


client = httpx.Client(transport=httpx.MockTransport(upstream))
print(fetch_public("https://8.8.8.8/hello", client))
try:
    fetch_public("https://8.8.8.8/start", client)
except UnsafeURL as exc:
    print("blocked after redirect:", exc)

hello
blocked after redirect: address 169.254.169.254 is not public

def make_env_app(production: bool) -> FastAPI:
    app = FastAPI(docs_url=None if production else "/docs",
                  redoc_url=None,
                  openapi_url=None if production else "/openapi.json")

    @app.exception_handler(Exception)
    async def on_error(request: Request, exc: Exception):
        detail = "Internal server error" if production else f"{type(exc).__name__}: {exc}"
        return JSONResponse({"detail": detail}, status_code=500)

    @app.get("/crash")
    def crash():
        raise RuntimeError("cannot connect to db-prod-3.internal:5432 password=hunter2")

    return app


for production in (False, True):
    client = TestClient(make_env_app(production), raise_server_exceptions=False)
    label = "production" if production else "development"
    print(f"{label:12} /docs", client.get("/docs").status_code, "| /crash", client.get("/crash").json())

development  /docs 200 | /crash {'detail': 'RuntimeError: cannot connect to db-prod-3.internal:5432 password=hunter2'}
production   /docs 404 | /crash {'detail': 'Internal server error'}

from fastapi.routing import APIRoute

inventory_app = FastAPI()


@inventory_app.get("/v2/prompts")
def prompts_v2():
    return []


@inventory_app.get("/v1/prompts")                   # old version nobody removed
def prompts_v1():
    return []


@inventory_app.get("/internal/debug/env")           # "temporary" debug route
def debug_env():
    return {}


DOCUMENTED = {("GET", "/v2/prompts")}               # in real life: loaded from openapi.json in Git
actual = {(m, r.path) for r in inventory_app.routes if isinstance(r, APIRoute) for m in r.methods}
print("not in the contract:", sorted(actual - DOCUMENTED))

not in the contract: [('GET', '/internal/debug/env'), ('GET', '/v1/prompts')]

from typing import Literal

from pydantic import ValidationError


class Translation(BaseModel):
    text: str = Field(max_length=5000)
    lang: Literal["en", "bn", "hi"]
    confidence: float = Field(ge=0, le=1)


raw_ok = '{"text": "Hello", "lang": "bn", "confidence": 0.93}'
raw_bad = '{"text": "Hello", "lang": "../../etc/passwd", "confidence": 7}'
print(Translation.model_validate_json(raw_ok))
try:
    Translation.model_validate_json(raw_bad)
except ValidationError as exc:
    print("rejected fields:", [e["loc"][0] for e in exc.errors()])

text='Hello' lang='bn' confidence=0.93
rejected fields: ['lang', 'confidence']

from fastapi.middleware.cors import CORSMiddleware

cors_app = FastAPI()
cors_app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.bitbyte.example"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
)


@cors_app.get("/ping")
def ping():
    return {"ok": True}


client = TestClient(cors_app)
for origin in ("https://app.bitbyte.example", "https://evil.example"):
    r = client.get("/ping", headers={"Origin": origin})
    print(origin, r.status_code, r.headers.get("access-control-allow-origin"))

https://app.bitbyte.example 200 https://app.bitbyte.example
https://evil.example 200 None

# scan the working tree and the full Git history (gitleaks 8.19 or newer)
gitleaks git --redact -v .
# trufflehog can also check whether a found key is still active
trufflehog git file://. --results=verified

uv add --dev pip-audit
uv run pip-audit                         # audit the current environment
uv export --no-dev --no-hashes -o requirements.txt
uv run pip-audit -r requirements.txt     # audit a lock export (no install needed)

SECURITY_HEADERS = {
    "Strict-Transport-Security": "max-age=31536000; includeSubDomains",   # browsers use HTTPS only
    "X-Content-Type-Options": "nosniff",                                   # do not guess content types
    "Content-Security-Policy": "default-src 'none'; frame-ancestors 'none'",  # an API serves no pages
    "Referrer-Policy": "no-referrer",
    "Cache-Control": "no-store",                                           # private data, no shared caches
}

headers_app = FastAPI()


@headers_app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    for name, value in SECURITY_HEADERS.items():
        response.headers.setdefault(name, value)
    return response


@headers_app.get("/me")
def me_headers():
    return {"user": "nadia"}


for name, value in TestClient(headers_app).get("/me").headers.items():
    if name.lower() not in {"content-length", "content-type"}:
        print(f"{name}: {value}")

strict-transport-security: max-age=31536000; includeSubDomains
x-content-type-options: nosniff
content-security-policy: default-src 'none'; frame-ancestors 'none'
referrer-policy: no-referrer
cache-control: no-store

import io
import logging
import re


class Redact(logging.Filter):
    PATTERNS = [
        (re.compile(r"(?i)bearer\s+[\w.\-]+"), "Bearer [redacted]"),
        (re.compile(r"[\w.+\-]+@[\w\-]+\.[\w.]+"), "[email]"),
        (re.compile(r"sk-[A-Za-z0-9_\-]{8,}"), "[api-key]"),
    ]

    def filter(self, record: logging.LogRecord) -> bool:
        message = record.getMessage()
        for pattern, replacement in self.PATTERNS:
            message = pattern.sub(replacement, message)
        record.msg, record.args = message, ()
        return True


stream = io.StringIO()
handler = logging.StreamHandler(stream)
handler.addFilter(Redact())
log = logging.getLogger("privacy-demo")
log.addHandler(handler)
log.setLevel(logging.INFO)
log.propagate = False

log.info("login ok user=%s auth=%s", "nadia@example.com", "Bearer eyJhbGciOi.abc.def")
log.info("llm call chars=%d key=%s", 412, "sk-live-12345678abcd")
print(stream.getvalue(), end="")

login ok user=[email] auth=Bearer [redacted]
llm call chars=412 key=[api-key]

import pathlib
import uuid

from fastapi import UploadFile

MAX_UPLOAD = 1_000_000
MAGIC = {".png": b"\x89PNG\r\n\x1a\n", ".pdf": b"%PDF-"}
UPLOAD_DIR = pathlib.Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

upload_app = FastAPI()


@upload_app.post("/upload")
async def upload(file: UploadFile):
    ext = pathlib.Path(file.filename or "").suffix.lower()
    magic = MAGIC.get(ext)
    if magic is None:
        raise HTTPException(415, "File type not allowed")
    data = await file.read(MAX_UPLOAD + 1)               # never read more than the limit + 1
    if len(data) > MAX_UPLOAD:
        raise HTTPException(413, "File too large")
    if not data.startswith(magic):
        raise HTTPException(415, "Content does not match the file type")
    (UPLOAD_DIR / f"{uuid.uuid4().hex}{ext}").write_bytes(data)    # our name, not theirs
    return {"type": ext, "bytes": len(data)}


client = TestClient(upload_app)
png = b"\x89PNG\r\n\x1a\n" + b"0" * 20
attempts = {
    "real png": ("photo.png", png),
    "script": ("../../evil.py", b"print(1)"),
    "png as pdf": ("report.pdf", png),
    "too large": ("big.png", b"\x89PNG\r\n\x1a\n" + b"0" * MAX_UPLOAD),
}
for label, (filename, content) in attempts.items():
    r = client.post("/upload", files={"file": (filename, content)})
    print(f"{label:11}", r.status_code, r.json())
print("stored files:", len(list(UPLOAD_DIR.iterdir())))

real png    200 {'type': '.png', 'bytes': 28}
script      415 {'detail': 'File type not allowed'}
png as pdf  415 {'detail': 'Content does not match the file type'}
too large   413 {'detail': 'File too large'}
stored files: 1

class SearchArgs(BaseModel):
    model_config = ConfigDict(extra="forbid")
    query: str = Field(max_length=200)


ALLOWED_TOOLS = {"search_docs"}                      # this endpoint only ever needs to read


def run_tool_call(user: str, call: dict) -> str:
    if call["name"] not in ALLOWED_TOOLS:
        return f"refused: tool {call['name']!r} is not allowed here"
    args = SearchArgs.model_validate(call["arguments"])
    return f"results for {user}: {args.query}"


# Suppose a poisoned document fooled the model into asking for a tool this endpoint never offered:
print(run_tool_call("nadia", {"name": "send_email", "arguments": {"to": "evil@attacker.example"}}))
print(run_tool_call("nadia", {"name": "search_docs", "arguments": {"query": "q3 plan"}}))
try:
    run_tool_call("nadia", {"name": "search_docs", "arguments": {"query": "q3", "user": "rafi"}})
except ValidationError as exc:
    print("rejected:", exc.errors()[0]["type"], exc.errors()[0]["loc"])

refused: tool 'send_email' is not allowed here
results for nadia: q3 plan
rejected: extra_forbidden ('user',)

# 1. Easy
client = TestClient(make_notes_app(True))
own = client.get("/notes/1", headers={"X-User": "nadia"})
other = client.get("/notes/2", headers={"X-User": "nadia"})
missing = client.get("/notes/99", headers={"X-User": "nadia"})
print(own.status_code, other.status_code, missing.status_code, other.json() == missing.json())


# 2. Medium
for url in ["http://100.64.0.1/", "http://0.0.0.0/", "http://224.0.0.1/"]:
    try:
        check_url(url)
    except UnsafeURL as exc:
        print(url, "->", exc)

200 404 404 True
http://100.64.0.1/ -> address 100.64.0.1 is not public
http://0.0.0.0/ -> address 0.0.0.0 is not public
http://224.0.0.1/ -> address 224.0.0.1 is not public

# 3. Hard
attacks = {
    "other user's note": lambda: TestClient(make_notes_app(True)).get("/notes/2", headers={"X-User": "nadia"}),
    "forged token": lambda: TestClient(make_auth_app(True)).get("/me", headers={"Authorization": f"Bearer {tokens['forged']}"}),
    "mass assignment": lambda: TestClient(make_profile_app(True)).patch("/profile", json={"is_admin": True}),
    "admin as user": lambda: TestClient(make_admin_app(True)).delete("/admin/users/9", headers={"X-User": "nadia"}),
}
results = {name: attack().status_code for name, attack in attacks.items()}
print(results)
assert all(code >= 400 for code in results.values())

{"other user's note": 404, 'forged token': 401, 'mass assignment': 422, 'admin as user': 403}