linumiq_net-user
7fe0cc3753
Near-1:1 clone of the prod remote-access stack, isolated on a new external dev_edge network and fronted by the same shared Caddy instance (dual-homed on edge + dev_edge). Dev is manual-start (not on boot). - Hostnames: app-dev / api-dev .linumiq.net, tunnels under *.dev.linumiq.net, dev tunnel ingress on port 7001. - Dev Supabase (project supabase-dev, *-dev containers), web, frps, redis, stripe-stub, bandwidth-worker with fresh independent secrets (gitignored). - Shared Caddyfile: app-dev -> web-dev, api-dev -> dev kong (+webhook block), *.dev -> frps-dev vhost. Caddy compose dual-homed on dev_edge. - On-demand-TLS authorizer (prod check-subdomain, in gitignored volumes/) extended additively: app-dev/api-dev -> 200; *.dev delegated to the dev authorizer. Prod allow-list logic unchanged. - dev.sh manual up/down/ps helper; README documents topology + secrets. Secrets, frps.toml, volumes/, web worktree and data dirs are gitignored.
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"}
|