HTTP গভীরে: ইডেমপোটেন্সি, ক্যাশিং ও অথ

client                                      server
  | 1. DNS lookup, TCP connect, TLS handshake  |
  |------------------------------------------->|
  | 2. METHOD /path?query HTTP/1.1             |   method: what to do
  |    Host, Accept, Authorization, If-None-Match   headers: who, which format, which version
  |    body (JSON)                             |
  |------------------------------------------->|
  |                  3. authenticate (who?)    |   401 if unknown
  |                  4. authorise (allowed?)   |   403 if not allowed
  |                  5. validate + run         |   400/404/409/422
  | 6. status, headers (ETag, Cache-Control),  |
  |    body                                    |
  |<-------------------------------------------|
  | 7. client caches, parses, retries or fails |

mkdir -p ~/projects/http-lab
cd ~/projects/http-lab
uv init --bare -q
uv add -q "fastapi[standard]"
# uv-এর বদলে pip দিয়ে:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -q "fastapi[standard]"
ls

pyproject.toml
uv.lock

# file: main.py
from typing import Annotated

from fastapi import FastAPI, Header, HTTPException, Response
from pydantic import BaseModel

import auth
import caching
import oauth
from caching import MODELS, etag_of, model_body

app = FastAPI(title="HTTP lab")
app.include_router(caching.router)
app.include_router(auth.router)
app.include_router(oauth.router)

JOBS = {"count": 0}


class ModelIn(BaseModel):
    dim: int
    status: str


class ModelPatch(BaseModel):
    dim: int | None = None
    status: str | None = None


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/jobs", status_code=201)
def create_job(response: Response):
    JOBS["count"] += 1  # every call creates something new
    response.headers["Location"] = f"/jobs/{JOBS['count']}"
    return {"job": JOBS["count"]}


@app.put("/models/{name}")
def put_model(
    name: str,
    item: ModelIn,
    response: Response,
    if_match: Annotated[str | None, Header()] = None,
):
    created = name not in MODELS
    if if_match is not None and (created or if_match != etag_of(model_body(name), "json")):
        raise HTTPException(412, "the model changed since you read it")
    MODELS[name] = item.model_dump()  # replace the whole resource
    response.status_code = 201 if created else 200
    return model_body(name)


@app.patch("/models/{name}")
def patch_model(name: str, item: ModelPatch):
    if name not in MODELS:
        raise HTTPException(404, "model not found")
    MODELS[name].update(item.model_dump(exclude_unset=True))  # change only what was sent
    return model_body(name)


@app.delete("/models/{name}", status_code=204)
def delete_model(name: str):
    if MODELS.pop(name, None) is None:
        raise HTTPException(404, "model not found")

# file: caching.py
import hashlib
import json
from typing import Annotated

from fastapi import APIRouter, Header, HTTPException, Request, Response
from fastapi.responses import JSONResponse, PlainTextResponse

router = APIRouter()

MODELS = {"embed-small": {"dim": 384, "status": "active"}}


def model_body(name: str) -> dict:
    return {"name": name, **MODELS[name]}


def etag_of(body: dict, fmt: str) -> str:
    """A strong validator: changes whenever the representation changes."""
    digest = hashlib.sha256(json.dumps([fmt, body], sort_keys=True).encode()).hexdigest()
    return f'"{digest[:12]}"'


@router.get("/models/{name}")
def get_model(name: str, request: Request, accept: Annotated[str, Header()] = "*/*"):
    if name not in MODELS:
        raise HTTPException(404, "model not found")
    if "application/json" in accept or "*/*" in accept:
        fmt = "json"
    elif "text/plain" in accept:
        fmt = "text"
    else:
        raise HTTPException(406, "available: application/json, text/plain")
    body = model_body(name)
    headers = {"ETag": etag_of(body, fmt), "Cache-Control": "no-cache", "Vary": "Accept"}
    if request.headers.get("if-none-match") == headers["ETag"]:
        return Response(status_code=304, headers=headers)
    if fmt == "text":
        return PlainTextResponse("\n".join(f"{k}={v}" for k, v in body.items()), headers=headers)
    return JSONResponse(body, headers=headers)


@router.get("/catalog")
def catalog(response: Response):
    response.headers["Cache-Control"] = "public, max-age=300"
    return {"models": sorted(MODELS)}

# file: auth.py
from typing import Annotated

from fastapi import APIRouter, Depends, Header, HTTPException

from caching import MODELS, model_body

router = APIRouter(prefix="/secure")

# Bearer tokens (the OAuth server in oauth.py adds more) and API keys for services.
TOKENS = {"tok-read": {"sub": "nadia", "scopes": ["models:read"]}}
API_KEYS = {"key-ingest-1": "ingest-service"}


def bearer(authorization: Annotated[str | None, Header()] = None) -> dict:
    """Authentication: who is calling? Failure is 401 and says how to authenticate."""
    if authorization is None or not authorization.startswith("Bearer "):
        raise HTTPException(401, "missing bearer token", headers={"WWW-Authenticate": "Bearer"})
    token = TOKENS.get(authorization.removeprefix("Bearer "))
    if token is None:
        raise HTTPException(401, "invalid token", headers={"WWW-Authenticate": 'Bearer error="invalid_token"'})
    return token


def need(scope: str):
    """Authorisation: is this caller allowed to do this? Failure is 403."""

    def check(token: Annotated[dict, Depends(bearer)]) -> dict:
        if scope not in token["scopes"]:
            raise HTTPException(403, f"requires scope {scope}")
        return token

    return check


@router.get("/models")
def list_models(token: Annotated[dict, Depends(need("models:read"))]):
    return {"user": token["sub"], "models": sorted(MODELS)}


@router.post("/models/{name}/retire")
def retire(name: str, token: Annotated[dict, Depends(need("models:write"))]):
    if name not in MODELS:
        raise HTTPException(404, "model not found")
    MODELS[name]["status"] = "retired"
    return {**model_body(name), "by": token["sub"]}


@router.post("/ingest")
def ingest(x_api_key: Annotated[str | None, Header()] = None):
    service = API_KEYS.get(x_api_key or "")
    if service is None:
        raise HTTPException(401, "invalid or missing API key")
    return {"accepted": True, "service": service}

# file: oauth.py
import hashlib
import itertools
from base64 import urlsafe_b64encode
from typing import Annotated

from fastapi import APIRouter, Form, HTTPException
from fastapi.responses import JSONResponse, RedirectResponse

from auth import TOKENS

router = APIRouter(prefix="/oauth")

CLIENTS = {
    # a confidential client (a backend job) can keep a secret
    "batch-job": {"secret": "batch-secret", "scopes": ["models:read", "models:write"]},
    # a public client (a browser or mobile app) cannot: it relies on PKCE instead
    "web-app": {"secret": None, "redirect_uri": "http://localhost:3000/callback", "scopes": ["models:read"]},
}
CODES: dict[str, dict] = {}
code_numbers = itertools.count(1)
token_numbers = itertools.count(2)  # tok-read is token 1


def s256(verifier: str) -> str:
    digest = hashlib.sha256(verifier.encode()).digest()
    return urlsafe_b64encode(digest).rstrip(b"=").decode()


def issue(sub: str, scopes: list[str]) -> JSONResponse:
    token = f"tok-{next(token_numbers)}"
    TOKENS[token] = {"sub": sub, "scopes": scopes}
    body = {"access_token": token, "token_type": "Bearer", "expires_in": 600, "scope": " ".join(scopes)}
    return JSONResponse(body, headers={"Cache-Control": "no-store"})


def error(code: str, status: int = 400) -> JSONResponse:
    return JSONResponse({"error": code}, status_code=status, headers={"Cache-Control": "no-store"})


@router.get("/authorize")
def authorize(
    response_type: str, client_id: str, redirect_uri: str, scope: str, state: str,
    code_challenge: str, code_challenge_method: str,
):
    """Normally a login + consent page. Here Nadia has already said yes."""
    client = CLIENTS.get(client_id)
    if client is None or client.get("redirect_uri") != redirect_uri:
        raise HTTPException(400, "unknown client or unregistered redirect_uri")  # never redirect there
    if response_type != "code" or code_challenge_method != "S256":
        return RedirectResponse(f"{redirect_uri}?error=invalid_request&state={state}", status_code=302)
    code = f"code-{next(code_numbers)}"
    CODES[code] = {"client_id": client_id, "challenge": code_challenge, "scopes": scope.split(), "sub": "nadia"}
    return RedirectResponse(f"{redirect_uri}?code={code}&state={state}", status_code=302)


@router.post("/token")
def token(
    grant_type: Annotated[str, Form()],
    client_id: Annotated[str, Form()],
    client_secret: Annotated[str | None, Form()] = None,
    scope: Annotated[str, Form()] = "",
    code: Annotated[str | None, Form()] = None,
    code_verifier: Annotated[str | None, Form()] = None,
):
    client = CLIENTS.get(client_id)
    if client is None:
        return error("invalid_client", 401)
    if grant_type == "client_credentials":
        if client["secret"] is None or client["secret"] != client_secret:
            return error("invalid_client", 401)
        wanted = scope.split() or client["scopes"]
        if not set(wanted) <= set(client["scopes"]):
            return error("invalid_scope")
        return issue(client_id, wanted)
    if grant_type == "authorization_code":
        grant = CODES.pop(code or "", None)  # a code works once
        if grant is None or grant["client_id"] != client_id or s256(code_verifier or "") != grant["challenge"]:
            return error("invalid_grant")
        return issue(grant["sub"], grant["scopes"])
    return error("unsupported_grant_type")

uv run fastapi run main.py --port 8000 > server.log 2>&1 &
export SERVER_PID=$!
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health
cat > call <<'EOF'
#!/bin/sh
exec curl -s -w ' -> HTTP %{http_code}\n' "$@"
EOF
chmod +x call
export BASE=http://localhost:8000

{"status":"ok"}

./call -X POST $BASE/jobs
./call -X POST $BASE/jobs
./call -X POST $BASE/jobs
echo "--- PUT twice"
./call -X PUT $BASE/models/embed-large -H 'Content-Type: application/json' -d '{"dim": 1024, "status": "active"}'
./call -X PUT $BASE/models/embed-large -H 'Content-Type: application/json' -d '{"dim": 1024, "status": "active"}'

{"job":1} -> HTTP 201
{"job":2} -> HTTP 201
{"job":3} -> HTTP 201
--- PUT twice
{"name":"embed-large","dim":1024,"status":"active"} -> HTTP 201
{"name":"embed-large","dim":1024,"status":"active"} -> HTTP 200

./call -X PUT $BASE/models/embed-large -H 'Content-Type: application/json' -d '{"status": "retired"}'
./call -X PATCH $BASE/models/embed-large -H 'Content-Type: application/json' -d '{"status": "retired"}'
./call -X DELETE $BASE/models/embed-large
./call -X DELETE $BASE/models/embed-large

{"detail":[{"type":"missing","loc":["body","dim"],"msg":"Field required","input":{"status":"retired"}}]} -> HTTP 422
{"name":"embed-large","dim":1024,"status":"retired"} -> HTTP 200
 -> HTTP 204
{"detail":"model not found"} -> HTTP 404

./call $BASE/models/does-not-exist
curl -s -i -X DELETE $BASE/health | tr -d '\r' | grep -i -E '^(HTTP|allow)'
./call -X PUT $BASE/models/x -H 'Content-Type: application/json' -d '{"dim": "big", "status": "active"}'

{"detail":"model not found"} -> HTTP 404
HTTP/1.1 405 Method Not Allowed
allow: GET
{"detail":[{"type":"int_parsing","loc":["body","dim"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"big"}]} -> HTTP 422

./call $BASE/models/embed-small
./call -H 'Accept: text/plain' $BASE/models/embed-small
./call -H 'Accept: image/png' $BASE/models/embed-small
curl -s -i -H 'Accept: text/plain' $BASE/models/embed-small | tr -d '\r' | grep -i -E '^(content-type|vary)'

{"name":"embed-small","dim":384,"status":"active"} -> HTTP 200
name=embed-small
dim=384
status=active -> HTTP 200
{"detail":"available: application/json, text/plain"} -> HTTP 406
vary: Accept
content-type: text/plain; charset=utf-8

curl -s -i $BASE/models/embed-small | tr -d '\r' | grep -i -E '^(HTTP|etag|cache-control)'
curl -s -i $BASE/catalog | tr -d '\r' | grep -i -E '^(HTTP|etag|cache-control)'

HTTP/1.1 200 OK
etag: "3c66bb0b1a1d"
cache-control: no-cache
HTTP/1.1 200 OK
cache-control: public, max-age=300

ETAG=$(curl -s -i $BASE/models/embed-small | tr -d '\r' | grep -i '^etag:' | cut -d' ' -f2)
echo "validator: $ETAG"
curl -s -o /dev/null -w 'first fetch:  HTTP %{http_code}, %{size_download} bytes\n' $BASE/models/embed-small
curl -s -o /dev/null -w 'revalidation: HTTP %{http_code}, %{size_download} bytes\n' -H "If-None-Match: $ETAG" $BASE/models/embed-small
./call -X PATCH $BASE/models/embed-small -H 'Content-Type: application/json' -d '{"dim": 768}'
curl -s -o /dev/null -w 'after change: HTTP %{http_code}, %{size_download} bytes\n' -H "If-None-Match: $ETAG" $BASE/models/embed-small

validator: "3c66bb0b1a1d"
first fetch:  HTTP 200, 50 bytes
revalidation: HTTP 304, 0 bytes
{"name":"embed-small","dim":768,"status":"active"} -> HTTP 200
after change: HTTP 200, 50 bytes

ETAG=$(curl -s -i $BASE/models/embed-small | tr -d '\r' | grep -i '^etag:' | cut -d' ' -f2)
./call -X PATCH $BASE/models/embed-small -H 'Content-Type: application/json' -d '{"status": "testing"}'
./call -X PUT $BASE/models/embed-small -H "If-Match: $ETAG" -H 'Content-Type: application/json' -d '{"dim": 384, "status": "active"}'

{"name":"embed-small","dim":768,"status":"testing"} -> HTTP 200
{"detail":"the model changed since you read it"} -> HTTP 412

curl -s -i $BASE/secure/models | tr -d '\r' | grep -i -E '^(HTTP|www-authenticate)'
./call -H 'Authorization: Bearer tok-read' $BASE/secure/models
./call -X POST -H 'Authorization: Bearer tok-read' $BASE/secure/models/embed-small/retire
./call -H 'Authorization: Bearer wrong' $BASE/secure/models

HTTP/1.1 401 Unauthorized
www-authenticate: Bearer
{"user":"nadia","models":["embed-small"]} -> HTTP 200
{"detail":"requires scope models:write"} -> HTTP 403
{"detail":"invalid token"} -> HTTP 401

./call -X POST $BASE/secure/ingest
./call -X POST -H 'X-API-Key: key-ingest-1' $BASE/secure/ingest

{"detail":"invalid or missing API key"} -> HTTP 401
{"accepted":true,"service":"ingest-service"} -> HTTP 200

Service                         Authorization server            API
  | POST /token                        |                         |
  | grant_type=client_credentials      |                         |
  | client_id + client_secret + scope  |                         |
  |----------------------------------->|                         |
  |      access_token (expires in 10 min)                        |
  |<-----------------------------------|                         |
  | GET /secure/models  Authorization: Bearer <token>            |
  |------------------------------------------------------------->|

./call -X POST $BASE/oauth/token -d grant_type=client_credentials -d client_id=batch-job -d client_secret=batch-secret -d scope=models:read
./call -X POST $BASE/oauth/token -d grant_type=client_credentials -d client_id=batch-job -d client_secret=wrong
./call -X POST $BASE/oauth/token -d grant_type=client_credentials -d client_id=batch-job -d client_secret=batch-secret -d scope=admin
./call -X POST $BASE/oauth/token -d grant_type=client_credentials -d client_id=batch-job -d client_secret=batch-secret

{"access_token":"tok-2","token_type":"Bearer","expires_in":600,"scope":"models:read"} -> HTTP 200
{"error":"invalid_client"} -> HTTP 401
{"error":"invalid_scope"} -> HTTP 400
{"access_token":"tok-3","token_type":"Bearer","expires_in":600,"scope":"models:read models:write"} -> HTTP 200

./call -X POST -H 'Authorization: Bearer tok-3' $BASE/secure/models/embed-small/retire
./call -H 'Authorization: Bearer tok-2' $BASE/secure/models

{"name":"embed-small","dim":768,"status":"retired","by":"batch-job"} -> HTTP 200
{"user":"batch-job","models":["embed-small"]} -> HTTP 200

App (browser/mobile)            Authorization server         API
 | 1. make code_verifier (secret random string)
 |    code_challenge = BASE64URL(SHA256(code_verifier))
 | 2. redirect user: /authorize?response_type=code&client_id&redirect_uri
 |       &scope&state&code_challenge&code_challenge_method=S256 ------>|
 |                          3. user logs in, approves scopes          |
 | 4. redirect to redirect_uri?code=...&state=... <-------------------|
 | 5. POST /token: grant_type=authorization_code, code, code_verifier ->|
 |                          6. server hashes the verifier, compares    |
 | 7. access_token (+ refresh_token, + id_token if scope has openid) <-|
 | 8. GET /api  Authorization: Bearer <access_token> ----------------------------->|

import base64
import hashlib


def code_challenge(verifier: str) -> str:
    digest = hashlib.sha256(verifier.encode("ascii")).digest()
    return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")


RFC_VERIFIER = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
print(code_challenge(RFC_VERIFIER))

E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM

export VERIFIER=dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk
export CHALLENGE=E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM
export AUTH="$BASE/oauth/authorize?response_type=code&client_id=web-app&redirect_uri=http://localhost:3000/callback&scope=models:read&state=xyz&code_challenge=$CHALLENGE&code_challenge_method=S256"
curl -s -o /dev/null -w '%{http_code} %{redirect_url}\n' "$AUTH"
./call -X POST $BASE/oauth/token -d grant_type=authorization_code -d client_id=web-app -d code=code-1 -d code_verifier=$VERIFIER

302 http://localhost:3000/callback?code=code-1&state=xyz
{"access_token":"tok-4","token_type":"Bearer","expires_in":600,"scope":"models:read"} -> HTTP 200

curl -s -o /dev/null -w '%{http_code} %{redirect_url}\n' "$AUTH"
./call -X POST $BASE/oauth/token -d grant_type=authorization_code -d client_id=web-app -d code=code-2 -d code_verifier=not-the-verifier
./call -X POST $BASE/oauth/token -d grant_type=authorization_code -d client_id=web-app -d code=code-2 -d code_verifier=$VERIFIER
./call -H 'Authorization: Bearer tok-4' $BASE/secure/models

302 http://localhost:3000/callback?code=code-2&state=xyz
{"error":"invalid_grant"} -> HTTP 400
{"error":"invalid_grant"} -> HTTP 400
{"user":"nadia","models":["embed-small"]} -> HTTP 200

./call -X POST $BASE/oauth/token -d grant_type=password -d client_id=web-app -d username=nadia -d password=s3cret-pass

{"error":"unsupported_grant_type"} -> HTTP 400

TEXT_ETAG=$(curl -s -i -H 'Accept: text/plain' $BASE/models/embed-small | tr -d '\r' | grep -i '^etag:' | cut -d' ' -f2)
curl -s -o /dev/null -w 'text revalidation: HTTP %{http_code}\n' -H 'Accept: text/plain' -H "If-None-Match: $TEXT_ETAG" $BASE/models/embed-small
curl -s -o /dev/null -w 'json with text etag: HTTP %{http_code}\n' -H "If-None-Match: $TEXT_ETAG" $BASE/models/embed-small

text revalidation: HTTP 304
json with text etag: HTTP 200

for method in PUT PUT DELETE DELETE; do
  curl -s -o /dev/null -w "$method %{http_code}\n" -X $method $BASE/models/embed-xl -H 'Content-Type: application/json' -d '{"dim": 2048, "status": "active"}'
done

PUT 201
PUT 200
DELETE 204
DELETE 404

ETAG=$(curl -s -i $BASE/models/embed-small | tr -d '\r' | grep -i '^etag:' | cut -d' ' -f2)
curl -s -o /dev/null -X PATCH $BASE/models/embed-small -H 'Content-Type: application/json' -d '{"dim": 512}'
curl -s -o /dev/null -w 'stale PUT: HTTP %{http_code}\n' -X PUT $BASE/models/embed-small -H "If-Match: $ETAG" -H 'Content-Type: application/json' -d '{"dim": 384, "status": "active"}'

stale PUT: HTTP 412

TOKEN=$(curl -s -X POST $BASE/oauth/token -d grant_type=client_credentials -d client_id=batch-job -d client_secret=batch-secret -d scope=models:read | python -c "import json,sys; print(json.load(sys.stdin)['access_token'])")
curl -s -o /dev/null -w 'list: HTTP %{http_code}\n' -H "Authorization: Bearer $TOKEN" $BASE/secure/models
curl -s -o /dev/null -w 'retire: HTTP %{http_code}\n' -X POST -H "Authorization: Bearer $TOKEN" $BASE/secure/models/embed-small/retire

list: HTTP 200
retire: HTTP 403

import base64
import hashlib
import secrets
from urllib.parse import parse_qs, urlparse

import requests

BASE = "http://localhost:8000"
verifier = secrets.token_urlsafe(64)
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
state = secrets.token_urlsafe(16)

params = {
    "response_type": "code", "client_id": "web-app", "redirect_uri": "http://localhost:3000/callback",
    "scope": "models:read", "state": state, "code_challenge": challenge, "code_challenge_method": "S256",
}
redirect = requests.get(f"{BASE}/oauth/authorize", params=params, allow_redirects=False, timeout=5)
query = parse_qs(urlparse(redirect.headers["Location"]).query)
assert query["state"] == [state], "state mismatch: possible forged redirect"

tokens = requests.post(f"{BASE}/oauth/token", timeout=5, data={
    "grant_type": "authorization_code", "client_id": "web-app",
    "code": query["code"][0], "code_verifier": verifier,
}).json()
api = requests.get(f"{BASE}/secure/models", headers={"Authorization": f"Bearer {tokens['access_token']}"}, timeout=5)
print(redirect.status_code, tokens["scope"], api.status_code, api.json()["user"])

302 models:read 200 nadia

kill $SERVER_PID