"""Access gate — MEMBERS mode.
Only members of the workspace that owns this app can use it. A visitor who
isn't signed in is sent to the workspace sign-in page; once they sign in, a
signed browser cookie remembers them for 24 hours. Membership is managed from
the app console — there is nothing to configure in this repo.
The platform sets ACCESS_GATE_SECRET on the deployed service (locally put one
in backend/.env.local, any value works for dev). The signed-in member's email
is exposed to route handlers as `request.state.member_email`.
"""
import base64
import hashlib
import hmac
import json
import os
import time
from datetime import datetime, timezone
from urllib.parse import quote
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
COOKIE_NAME = "app_member"
COOKIE_MAX_AGE_S = 24 * 60 * 60
APP_SLUG = "{{GHOSTY_SLUG}}"
SIGN_IN_BASE = (os.environ.get("GHOSTY_ADMIN_URL") or "{{GHOSTY_ADMIN_URL}}").rstrip("/")
# The sign-in handshake lands back here. Never mount app routes under /__ghosty/.
CALLBACK_PATH = "/__ghosty/access"
ASSERTION_AUDIENCE = "ghosty-app-access"
_NOT_CONFIGURED_HTML = """
Access not configured
This app isn’t accepting visitors right now
Its access settings aren’t configured. The app’s owner can fix this from the app console.
"""
_EXPIRED_HTML = """
Sign-in link expired
That sign‑in link has expired
Sign in again to open this app.
"""
def _b64url_encode(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
def _b64url_decode(segment: str) -> bytes:
return base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4))
def _b64url_json(segment: str):
try:
parsed = json.loads(_b64url_decode(segment))
return parsed if isinstance(parsed, dict) else None
except (ValueError, TypeError):
return None
def _hmac(secret: str, data: str) -> str:
return _b64url_encode(hmac.new(secret.encode(), data.encode(), hashlib.sha256).digest())
def _read_member_cookie(secret: str, value: str) -> str | None:
"""Cookie = base64url({email, exp}) + '.' + HMAC. Returns the email or None."""
if not value or "." not in value:
return None
payload_b64, _, sig = value.rpartition(".")
if not hmac.compare_digest(sig, _hmac(secret, f"member-cookie-v1:{payload_b64}")):
return None
payload = _b64url_json(payload_b64)
if not payload or not isinstance(payload.get("email"), str) or not isinstance(payload.get("exp"), (int, float)):
return None
if payload["exp"] <= time.time():
return None
return payload["email"]
def _mint_member_cookie(secret: str, email: str) -> str:
payload_b64 = _b64url_encode(
json.dumps({"email": email, "exp": int(time.time()) + COOKIE_MAX_AGE_S}).encode()
)
return f"{payload_b64}.{_hmac(secret, f'member-cookie-v1:{payload_b64}')}"
def _verify_sign_in_assertion(secret: str, token: str) -> str | None:
"""Verify the short-lived signed sign-in assertion the workspace hands
back on the callback (HS256; audience + app + expiry checked). Returns
the member's email, or None when anything about it is off."""
parts = token.split(".")
if len(parts) != 3:
return None
header_b64, payload_b64, sig = parts
if not hmac.compare_digest(sig, _hmac(secret, f"{header_b64}.{payload_b64}")):
return None
header = _b64url_json(header_b64)
if not header or header.get("alg") != "HS256":
return None
payload = _b64url_json(payload_b64)
if not payload or not isinstance(payload.get("email"), str):
return None
if payload.get("aud") != ASSERTION_AUDIENCE or payload.get("app") != APP_SLUG:
return None
exp = payload.get("exp")
if not isinstance(exp, (int, float)) or exp <= time.time():
return None
return payload["email"]
def _sign_in_url(request: Request) -> str:
proto = request.headers.get("x-forwarded-proto") or request.url.scheme or "https"
host = request.headers.get("host", "")
return_to = f"{proto}://{host}{request.url.path}" + (f"?{request.url.query}" if request.url.query else "")
return f"{SIGN_IN_BASE}/access/authorize?app={quote(APP_SLUG)}&return_to={quote(return_to, safe='')}"
def _is_production() -> bool:
return os.environ.get("NODE_ENV", "") == "production" or os.environ.get("K_SERVICE", "") != ""
def install_access_gate(app: FastAPI) -> None:
warned = {"missing_secret": False}
@app.middleware("http")
async def access_gate(request: Request, call_next):
if request.url.path == "/health":
return await call_next(request)
secret = os.environ.get("ACCESS_GATE_SECRET", "")
# Not configured: in local dev stay open (a fresh checkout has no
# .env.local yet); in production FAIL CLOSED — a members-only app must
# never quietly become public because its secret went missing.
if not secret:
if not _is_production():
if not warned["missing_secret"]:
warned["missing_secret"] = True
print(
json.dumps(
{
"level": "warn",
"message": "access_gate: ACCESS_GATE_SECRET is not set — open for local dev only",
"timestamp": datetime.now(timezone.utc).isoformat(),
}
),
flush=True,
)
return await call_next(request)
if not warned["missing_secret"]:
warned["missing_secret"] = True
print(
json.dumps(
{
"level": "error",
"message": "access_gate: ACCESS_GATE_SECRET is not set — refusing all access",
"timestamp": datetime.now(timezone.utc).isoformat(),
}
),
flush=True,
)
if request.url.path.startswith("/api"):
return JSONResponse({"error": "Access is not configured"}, status_code=503)
return HTMLResponse(_NOT_CONFIGURED_HTML, status_code=503)
# Sign-in callback: the workspace hands back a short-lived assertion;
# a valid one becomes the member cookie. Sits above the gate by design.
if request.url.path == CALLBACK_PATH:
token = request.query_params.get("token", "")
email = _verify_sign_in_assertion(secret, token) if token else None
if not email:
return HTMLResponse(
_EXPIRED_HTML.replace("__SIGN_IN_URL__", _sign_in_url(request)),
status_code=401,
)
next_param = request.query_params.get("next", "/")
# Same-app paths only: one leading '/', second char neither '/' nor
# '\' (browsers treat '/\host' and '//host' as protocol-relative).
safe_next = next_param if next_param.startswith("/") and next_param[1:2] not in ("/", "\\") else "/"
response = RedirectResponse(safe_next, status_code=302)
response.set_cookie(
COOKIE_NAME,
_mint_member_cookie(secret, email),
max_age=COOKIE_MAX_AGE_S,
httponly=True,
secure=_is_production(),
samesite="lax",
)
return response
# Returning member with a valid cookie.
email = _read_member_cookie(secret, request.cookies.get(COOKIE_NAME, ""))
if email:
request.state.member_email = email
return await call_next(request)
# Not signed in. APIs get JSON; pages go to the workspace sign-in.
if request.url.path.startswith("/api"):
return JSONResponse(
{"error": "Sign-in required", "signInUrl": _sign_in_url(request)},
status_code=401,
)
return RedirectResponse(_sign_in_url(request), status_code=302)