gerhards
50ab46dbe1
App layer (R1): bind frps NewProxy to token-owned subdomain (anti-hijack), default-deny unknown webhook ops, HMAC-verify stripe-stub billing webhook, enforce bandwidth quota kill-switch (Ping op), least-privilege table grants (migrations 0002/0003), GOTRUE_PASSWORD_MIN_LENGTH=12. Infra/net (R2): unpublish internal host ports (kong/pooler/analytics/frps-dash), read-only docker-socket-proxy for vector (no host breakout), on-demand-TLS allow-list authorizer, edge-block machine-only webhooks, no-new-privileges on custom containers. Secrets (R3): rotate Postgres password (all roles) + frps dashboard; replace predictable supavisor defaults; secrets externalized to gitignored .env. Med/Low (R4): security response headers (HSTS/XCTO/XFO/Referrer/Permissions/COOP), restrict frp proxy_type to http (no open relay), disable destructive redis commands, tighten frps.toml perms. No secrets committed; rotated values live only in gitignored .env files.
64 lines
1.6 KiB
Python
64 lines
1.6 KiB
Python
"""stripe-stub: pretends every checkout succeeds and forwards test webhooks.
|
|
|
|
Forwarded webhooks are signed with an HMAC-SHA256 signature so the edge
|
|
function can reject unsigned/forged/replayed events.
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import time
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import FastAPI, Request
|
|
|
|
DOMAIN = os.environ.get("DOMAIN", "linumiq.net")
|
|
EDGE_URL = os.environ.get(
|
|
"EDGE_STRIPE_WEBHOOK_URL",
|
|
"http://supabase-edge-functions:9000/stripe-webhook",
|
|
)
|
|
WEBHOOK_SECRET = os.environ.get("STRIPE_STUB_WEBHOOK_SECRET", "")
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
def sign(body: bytes) -> str:
|
|
ts = int(time.time())
|
|
mac = hmac.new(
|
|
WEBHOOK_SECRET.encode(),
|
|
f"{ts}.".encode() + body,
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
return f"t={ts},v1={mac}"
|
|
|
|
|
|
@app.post("/v1/checkout/sessions")
|
|
async def create_session(_request: Request) -> dict[str, str]:
|
|
session_id = f"stub_{uuid.uuid4().hex}"
|
|
return {
|
|
"id": session_id,
|
|
"url": f"https://app.{DOMAIN}/billing/success?session={session_id}",
|
|
}
|
|
|
|
|
|
@app.post("/v1/webhooks/test")
|
|
async def forward_test(request: Request) -> Any:
|
|
body = await request.body()
|
|
headers = {
|
|
"content-type": request.headers.get("content-type", "application/json"),
|
|
"x-stub-signature": sign(body),
|
|
}
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
resp = await client.post(EDGE_URL, content=body, headers=headers)
|
|
try:
|
|
return resp.json()
|
|
except Exception:
|
|
return {"upstream_status": resp.status_code, "body": resp.text}
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|