Spaces:

sitcod3
/

modul-x

Running

FortyDN Dev

fix(auth): exact-match open paths for engine/status, metrics/inference, smtp/probe

5979e23 1 day ago

6.5 kB

"""Supabase JWT verification + license gating for the paid API.
The central Supabase project issues the session JWT; this backend verifies
it against the project's PUBLIC JWKS endpoint (no shared database needed).
License rows are read from the central project via its service-role REST
API (server-side only).
Enforcement is OFF by default (LICENSE_ENFORCEMENT != "true") so existing
open/demo deployments keep working. Turn it on at go-live:
    SUPABASE_URL=https://ciytvulujiqemqljablq.supabase.co
    SUPABASE_SERVICE_ROLE_KEY=<service role key, server only>
    LICENSE_ENFORCEMENT=true
Frontend equivalent lives in the FortyDN website repo (src/lib/productApi.ts).
Follows docs/AUTH_SETUP.md section 3 (Python snippet).
"""

import logging
from typing import Optional

import httpx
from fastapi import Request
from fastapi.responses import JSONResponse
from jose import jwt as jose_jwt
from jose import jwk as jose_jwk
from jose.exceptions import JWTError

from .config import settings

logger = logging.getLogger("emailvalidator.auth")

CENTRAL_SUPABASE_URL = (
    settings.supabase_url or "https://ciytvulujiqemqljablq.supabase.co"
)
JWKS_URL = f"{CENTRAL_SUPABASE_URL}/auth/v1/.well-known/jwks.json"

# Modul-X is unlocked directly, via Validate+Send, or via Full Bridge.
MODULX_GRANTS = frozenset(
    {
        "modul-x",
        "combo-validate-send",
        "combo-full-bridge",
    }
)

_jwks_cache: Optional[dict] = None


def enforcement_on() -> bool:
    return (settings.license_enforcement or "").lower() == "true"


async def _get_jwks() -> dict:
    """Fetch (and cache) the Supabase JWKS document."""
    global _jwks_cache
    if _jwks_cache is None:
        async with httpx.AsyncClient(timeout=10.0) as client:
            res = await client.get(JWKS_URL)
            res.raise_for_status()
            _jwks_cache = res.json()
    return _jwks_cache  # type: ignore[return-value]


def _bearer_token(request: Request) -> Optional[str]:
    header = request.headers.get("authorization", "")
    if not header.startswith("Bearer "):
        return None
    token = header[7:].strip()
    return token if token else None


async def verify_supabase_jwt(token: str) -> dict:
    """Verify the Supabase session JWT via JWKS. Returns claims (sub = user_id)."""
    jwks = await _get_jwks()
    # Supabase rotates keys; try each until one verifies.
    last_error: Optional[Exception] = None
    for key_data in jwks.get("keys", []):
        try:
            key = jose_jwk.construct(key_data)
            claims = jose_jwt.decode(
                token,
                key,
                algorithms=["ES256", "RS256"],
                audience="authenticated",
            )
            if not claims.get("sub"):
                raise JWTError("JWT has no sub claim")
            return claims
        except JWTError as exc:
            last_error = exc
            continue
    raise JWTError(f"JWT verification failed: {last_error}")


async def has_modulx_license(user_id: str) -> bool:
    """Check for an active Modul-X license via the central service-role API."""
    service_key = settings.supabase_service_role_key
    if not service_key:
        logger.warning("[auth] SUPABASE_SERVICE_ROLE_KEY not set - cannot check licenses.")
        return False
    params = {
        "user_id": f"eq.{user_id}",
        "status": "eq.active",
        "select": "product",
    }
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            res = await client.get(
                f"{CENTRAL_SUPABASE_URL}/rest/v1/licenses",
                params=params,
                headers={
                    "apikey": service_key,
                    "Authorization": f"Bearer {service_key}",
                },
            )
        if res.status_code != 200:
            # Missing table (migrations not applied yet) => fail closed with a hint.
            logger.warning(
                "[auth] license lookup HTTP %s: %s",
                res.status_code,
                res.text[:200],
            )
            return False
        rows = res.json()
        return isinstance(rows, list) and any(
            isinstance(r, dict) and r.get("product") in MODULX_GRANTS for r in rows
        )
    except Exception as exc:
        logger.warning("[auth] license lookup error: %s", exc)
        return False


# Paths that stay open (health probes, docs, root) plus the autonomous-engine
# ops surfaces: /engine/status + /metrics/inference are aggregate diagnostics
# (no PII), and /smtp/probe* enforce their own shared-secret guard
# (X-Worker-Token) — license JWTs do not apply to machine-to-machine probes.
_OPEN_PREFIXES = (
    "/api/v1/health",
    "/api/health",
    "/api/v1/engine/status",
    "/api/v1/metrics",
    "/api/v1/smtp/probe",
    "/docs",
    "/redoc",
    "/openapi.json",
)
_OPEN_EXACT = frozenset(
    {
        "/",
        "/api/v1/health",
        "/api/health",
        "/api/v1/engine/status",
        "/api/v1/metrics/inference",
        "/api/v1/smtp/probe",
    }
)


def _is_open_path(path: str) -> bool:
    if path in _OPEN_EXACT:
        return True
    return any(path.startswith(p) and path != p for p in _OPEN_PREFIXES)


async def license_middleware(request: Request, call_next):
    """FastAPI HTTP middleware: JWT + license gate for paid routes.
    Mirrors the Express pattern (Modul-S/E): /api/v1/health stays open,
    everything else under /api requires a valid JWT + active license.
    No-op while LICENSE_ENFORCEMENT != "true" (demo mode).
    """
    if not enforcement_on():
        return await call_next(request)
    path = request.url.path
    if _is_open_path(path) or not path.startswith("/api"):
        return await call_next(request)
    token = _bearer_token(request)
    if not token:
        return JSONResponse(
            status_code=401,
            content={"detail": "Missing bearer token - sign in first."},
        )
    try:
        claims = await verify_supabase_jwt(token)
    except JWTError:
        return JSONResponse(
            status_code=401,
            content={"detail": "Invalid or expired token."},
        )
    user_id = claims.get("sub", "")
    if not user_id or not await has_modulx_license(user_id):
        return JSONResponse(
            status_code=402,
            content={"detail": "No active Modul-X license - see /pricing."},
        )
    request.state.auth = {"user_id": user_id, "email": claims.get("email")}
    return await call_next(request)

	"""Supabase JWT verification + license gating for the paid API.

	The central Supabase project issues the session JWT; this backend verifies
	it against the project's PUBLIC JWKS endpoint (no shared database needed).
	License rows are read from the central project via its service-role REST
	API (server-side only).

	Enforcement is OFF by default (LICENSE_ENFORCEMENT != "true") so existing
	open/demo deployments keep working. Turn it on at go-live:

	SUPABASE_URL=https://ciytvulujiqemqljablq.supabase.co
	SUPABASE_SERVICE_ROLE_KEY=<service role key, server only>
	LICENSE_ENFORCEMENT=true

	Frontend equivalent lives in the FortyDN website repo (src/lib/productApi.ts).
	Follows docs/AUTH_SETUP.md section 3 (Python snippet).
	"""

	import logging
	from typing import Optional

	import httpx
	from fastapi import Request
	from fastapi.responses import JSONResponse
	from jose import jwt as jose_jwt
	from jose import jwk as jose_jwk
	from jose.exceptions import JWTError

	from .config import settings

	logger = logging.getLogger("emailvalidator.auth")

	CENTRAL_SUPABASE_URL = (
	settings.supabase_url or "https://ciytvulujiqemqljablq.supabase.co"
	)
	JWKS_URL = f"{CENTRAL_SUPABASE_URL}/auth/v1/.well-known/jwks.json"

	# Modul-X is unlocked directly, via Validate+Send, or via Full Bridge.
	MODULX_GRANTS = frozenset(
	{
	"modul-x",
	"combo-validate-send",
	"combo-full-bridge",
	}
	)

	_jwks_cache: Optional[dict] = None


	def enforcement_on() -> bool:
	return (settings.license_enforcement or "").lower() == "true"


	async def _get_jwks() -> dict:
	"""Fetch (and cache) the Supabase JWKS document."""
	global _jwks_cache
	if _jwks_cache is None:
	async with httpx.AsyncClient(timeout=10.0) as client:
	res = await client.get(JWKS_URL)
	res.raise_for_status()
	_jwks_cache = res.json()
	return _jwks_cache # type: ignore[return-value]


	def _bearer_token(request: Request) -> Optional[str]:
	header = request.headers.get("authorization", "")
	if not header.startswith("Bearer "):
	return None
	token = header[7:].strip()
	return token if token else None


	async def verify_supabase_jwt(token: str) -> dict:
	"""Verify the Supabase session JWT via JWKS. Returns claims (sub = user_id)."""
	jwks = await _get_jwks()
	# Supabase rotates keys; try each until one verifies.
	last_error: Optional[Exception] = None
	for key_data in jwks.get("keys", []):
	try:
	key = jose_jwk.construct(key_data)
	claims = jose_jwt.decode(
	token,
	key,
	algorithms=["ES256", "RS256"],
	audience="authenticated",
	)
	if not claims.get("sub"):
	raise JWTError("JWT has no sub claim")
	return claims
	except JWTError as exc:
	last_error = exc
	continue
	raise JWTError(f"JWT verification failed: {last_error}")


	async def has_modulx_license(user_id: str) -> bool:
	"""Check for an active Modul-X license via the central service-role API."""
	service_key = settings.supabase_service_role_key
	if not service_key:
	logger.warning("[auth] SUPABASE_SERVICE_ROLE_KEY not set - cannot check licenses.")
	return False
	params = {
	"user_id": f"eq.{user_id}",
	"status": "eq.active",
	"select": "product",
	}
	try:
	async with httpx.AsyncClient(timeout=10.0) as client:
	res = await client.get(
	f"{CENTRAL_SUPABASE_URL}/rest/v1/licenses",
	params=params,
	headers={
	"apikey": service_key,
	"Authorization": f"Bearer {service_key}",
	},
	)
	if res.status_code != 200:
	# Missing table (migrations not applied yet) => fail closed with a hint.
	logger.warning(
	"[auth] license lookup HTTP %s: %s",
	res.status_code,
	res.text[:200],
	)
	return False
	rows = res.json()
	return isinstance(rows, list) and any(
	isinstance(r, dict) and r.get("product") in MODULX_GRANTS for r in rows
	)
	except Exception as exc:
	logger.warning("[auth] license lookup error: %s", exc)
	return False


	# Paths that stay open (health probes, docs, root) plus the autonomous-engine
	# ops surfaces: /engine/status + /metrics/inference are aggregate diagnostics
	# (no PII), and /smtp/probe* enforce their own shared-secret guard
	# (X-Worker-Token) — license JWTs do not apply to machine-to-machine probes.
	_OPEN_PREFIXES = (
	"/api/v1/health",
	"/api/health",
	"/api/v1/engine/status",
	"/api/v1/metrics",
	"/api/v1/smtp/probe",
	"/docs",
	"/redoc",
	"/openapi.json",
	)
	_OPEN_EXACT = frozenset(
	{
	"/",
	"/api/v1/health",
	"/api/health",
	"/api/v1/engine/status",
	"/api/v1/metrics/inference",
	"/api/v1/smtp/probe",
	}
	)


	def _is_open_path(path: str) -> bool:
	if path in _OPEN_EXACT:
	return True
	return any(path.startswith(p) and path != p for p in _OPEN_PREFIXES)


	async def license_middleware(request: Request, call_next):
	"""FastAPI HTTP middleware: JWT + license gate for paid routes.

	Mirrors the Express pattern (Modul-S/E): /api/v1/health stays open,
	everything else under /api requires a valid JWT + active license.
	No-op while LICENSE_ENFORCEMENT != "true" (demo mode).
	"""
	if not enforcement_on():
	return await call_next(request)
	path = request.url.path
	if _is_open_path(path) or not path.startswith("/api"):
	return await call_next(request)
	token = _bearer_token(request)
	if not token:
	return JSONResponse(
	status_code=401,
	content={"detail": "Missing bearer token - sign in first."},
	)
	try:
	claims = await verify_supabase_jwt(token)
	except JWTError:
	return JSONResponse(
	status_code=401,
	content={"detail": "Invalid or expired token."},
	)
	user_id = claims.get("sub", "")
	if not user_id or not await has_modulx_license(user_id):
	return JSONResponse(
	status_code=402,
	content={"detail": "No active Modul-X license - see /pricing."},
	)
	request.state.auth = {"user_id": user_id, "email": claims.get("email")}
	return await call_next(request)