import base64
import secrets
import time
from typing import Any
import jwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa as _rsa
from starlette.requests import Request
from starlette.responses import JSONResponse, PlainTextResponse, RedirectResponse
from starlette.status import HTTP_201_CREATED
from signnow_client import SignNowAPIClient
from signnow_client.config import load_signnow_config
from .config import load_settings
# ============= CONFIG =============
settings = load_settings()
signnow_config = load_signnow_config()
# ============= KEYGEN (RS256) =============
private_key = settings.get_rsa_private_key()
public_key = private_key.public_key()
public_pem = public_key.public_bytes(encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo)
# JWKS
numbers: _rsa.RSAPublicNumbers = public_key.public_numbers()
e_b = numbers.e.to_bytes((numbers.e.bit_length() + 7) // 8, "big")
n_b = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big")
def b64url(b: bytes) -> str:
return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
JWKS = {
"keys": [
{
"kty": "RSA",
"use": "sig",
"kid": settings.oauth_jwk_kid,
"alg": "RS256",
"n": b64url(n_b),
"e": b64url(e_b),
}
]
}
# ============= Helpers =============
# Initialize SignNow API client
signnow_client = SignNowAPIClient(signnow_config)
def _verify_jwt(token: str) -> dict[str, Any] | None:
try:
return jwt.decode(
token,
key=public_key,
algorithms=["RS256"],
audience=[settings.effective_resource_http_url, settings.effective_resource_sse_url], # both resources are considered valid audiences
issuer=str(settings.oauth_issuer),
options={"require": ["exp", "iat", "iss", "aud"]},
)
except Exception:
return None
# ============= OAuth endpoints =============
async def openid_config(_: Request) -> JSONResponse:
return JSONResponse({
"issuer": str(settings.oauth_issuer),
"authorization_endpoint": f"{str(settings.oauth_issuer)}authorize",
"token_endpoint": f"{str(settings.oauth_issuer)}oauth2/token",
"jwks_uri": f"{str(settings.oauth_issuer)}.well-known/jwks.json",
"registration_endpoint": f"{str(settings.oauth_issuer)}oauth2/register",
"scopes_supported": ["openid", "profile", "offline_access", "*"],
"response_types_supported": ["code"],
"grant_types_supported": ["authorization_code", "refresh_token"],
"code_challenge_methods_supported": ["S256"],
"token_endpoint_auth_methods_supported": ["none", "client_secret_post"],
})
async def oauth_as_meta(_: Request) -> JSONResponse:
# can return the same object as openid-configuration
return await openid_config(_)
async def jwks(_: Request) -> JSONResponse:
return JSONResponse(JWKS)
async def authorize(req: Request) -> RedirectResponse:
q = req.query_params
redirect_uri = q.get("redirect_uri")
state = q.get("state", "")
# Build redirect URL with proper query parameters
base_url = str(signnow_config.app_base) + "authorize"
params = {"response_type": "code", "client_id": signnow_config.client_id, "redirect_uri": redirect_uri}
# Add state only if it exists
if state:
params["state"] = state
# Build query string
query_string = "&".join(f"{key}={value}" for key, value in params.items())
redirect_url = f"{base_url}?{query_string}"
return RedirectResponse(redirect_url, status_code=302)
async def token(req: Request) -> JSONResponse:
form = await req.form()
grant_type = form.get("grant_type")
if grant_type == "authorization_code":
code = form.get("code")
# Get tokens from SignNow API
if not code:
return JSONResponse({"error": "invalid_request", "error_description": "code parameter required"}, status_code=400)
if isinstance(code, str):
signnow_response = signnow_client.get_tokens(code=code)
else:
return JSONResponse({"error": "invalid_request", "error_description": "code must be a string"}, status_code=400)
if not signnow_response:
return JSONResponse({"error": "external_token_error"}, status_code=500)
# Return tokens from SignNow API
return JSONResponse({
"token_type": signnow_response.get("token_type", "Bearer"),
"access_token": signnow_response.get("access_token"),
"expires_in": signnow_response.get("expires_in", settings.access_ttl),
"refresh_token": signnow_response.get("refresh_token"),
"scope": "*",
})
elif grant_type == "refresh_token":
refresh = form.get("refresh_token")
if not refresh:
return JSONResponse({"error": "invalid_grant"}, status_code=400)
# Get new tokens from SignNow API using refresh token
if not refresh:
return JSONResponse({"error": "invalid_request", "error_description": "refresh_token parameter required"}, status_code=400)
if isinstance(refresh, str):
signnow_response = signnow_client.refresh_tokens(refresh_token=refresh)
else:
return JSONResponse({"error": "invalid_request", "error_description": "refresh_token must be a string"}, status_code=400)
if signnow_response:
return JSONResponse({
"token_type": signnow_response.get("token_type", "Bearer"),
"access_token": signnow_response.get("access_token"),
"expires_in": signnow_response.get("expires_in", settings.access_ttl),
"refresh_token": signnow_response.get("refresh_token"),
"scope": signnow_response.get("scope", "*"),
})
else:
return JSONResponse({"error": "invalid_grant"}, status_code=400)
else:
return JSONResponse({"error": "unsupported_grant_type"}, status_code=400)
async def introspect(req: Request) -> JSONResponse:
form = await req.form()
token = form.get("token", "")
if isinstance(token, str):
claims = _verify_jwt(token)
else:
claims = None
active = claims is not None
resp: dict[str, Any] = {"active": bool(active)}
if active and claims:
resp.update({
"iss": claims["iss"],
"sub": claims["sub"],
"aud": claims["aud"],
"client_id": claims.get("client_id"),
"scope": claims.get("scope", ""),
"exp": claims["exp"],
"iat": claims["iat"],
})
return JSONResponse(resp)
async def revoke(req: Request) -> PlainTextResponse | JSONResponse:
form = await req.form()
token = form.get("token")
if not token:
return JSONResponse({"error": "invalid_request", "error_description": "token parameter required"}, status_code=400)
# Send revoke request to SignNow API
if not token:
return JSONResponse({"error": "invalid_request", "error_description": "token parameter required"}, status_code=400)
if isinstance(token, str):
if signnow_client.revoke_token(token):
return PlainTextResponse("", status_code=200)
else:
return JSONResponse({"error": "external_revoke_error"}, status_code=500)
else:
return JSONResponse({"error": "invalid_request", "error_description": "token must be a string"}, status_code=400)
# ============= PRM (Protected Resource Metadata) =============
def prm_for_resource(resource_url: str) -> JSONResponse:
return JSONResponse({
"resource": resource_url,
"authorization_servers": [str(settings.oauth_issuer)],
"bearer_methods_supported": ["header"],
"scopes_supported": ["openid", "profile", "offline_access", "*"],
})
async def prm_root(_: Request) -> JSONResponse:
return prm_for_resource(settings.effective_resource_http_url)
async def prm_mcp(_: Request) -> JSONResponse:
return prm_for_resource(settings.effective_resource_http_url)
async def prm_sse(_: Request) -> JSONResponse:
return prm_for_resource(settings.effective_resource_sse_url)
REGISTERED_CLIENTS: dict[str, dict[str, Any]] = {}
async def register(req: Request) -> JSONResponse:
data = await req.json()
redirect_uris = data.get("redirect_uris") or []
token_method = (data.get("token_endpoint_auth_method") or "none").lower()
client_id = secrets.token_urlsafe(24)
client_secret = None
if token_method == "client_secret_post":
client_secret = secrets.token_urlsafe(32)
resp: dict[str, Any] = {
"client_id": client_id,
"client_id_issued_at": int(time.time()),
"redirect_uris": redirect_uris,
"token_endpoint_auth_method": token_method,
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"registration_client_uri": f"{str(settings.oauth_issuer)}/oauth2/register/{client_id}",
"client_secret_expires_at": 0,
}
if client_secret:
resp["client_secret"] = client_secret
resp["registration_access_token"] = secrets.token_urlsafe(24)
return JSONResponse(resp, status_code=HTTP_201_CREATED)
# ============= Middleware =============
class TrailingSlashCompatMiddleware:
"""
Makes /mcp and /sse equivalent to /mcp/ and /sse/ without redirect.
"""
def __init__(self, app: Any, accept_exact: tuple[str, ...] = ("/mcp", "/sse")) -> None:
self.app = app
self.accept_exact = set(accept_exact)
async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None:
if scope["type"] == "http":
path = scope.get("path", "")
# Rewrite only exact matches to avoid touching sub-routes
if path in self.accept_exact:
scope = dict(scope)
scope["path"] = path + "/"
await self.app(scope, receive, send)
class BearerJWTASGIMiddleware:
def __init__(self, app: Any, protect_prefixes: tuple[str, ...] = ("/mcp", "/sse", "/messages")) -> None:
self.app = app
self._paths = tuple(protect_prefixes)
from .token_provider import TokenProvider
self.token_provider = TokenProvider()
async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
request = Request(scope, receive=receive)
path = request.url.path
# Let CORS outside handle preflight
if request.method == "OPTIONS":
await self.app(scope, receive, send)
if any(path.startswith(x) for x in self._paths):
if not self.token_provider.has_config_credentials():
token = self.token_provider.get_access_token(dict(request.headers))
if not token:
await send({
"type": "http.response.start",
"status": 401,
"headers": [
(b"www-authenticate", f'Bearer resource_metadata="{str(settings.oauth_issuer)}/.well-known/oauth-protected-resource"'.encode()),
(b"content-type", b"text/plain; charset=utf-8"),
],
})
await send({"type": "http.response.body", "body": b"Unauthorized"})
return
await self.app(scope, receive, send)
# ============= Routes =============
def get_auth_routes() -> list[tuple[str, Any, list[str]]]:
"""Returns all OAuth-related routes"""
return [
# OAuth discovery
("/.well-known/openid-configuration", openid_config, ["GET", "HEAD", "OPTIONS"]),
("/.well-known/oauth-authorization-server", oauth_as_meta, ["GET", "HEAD", "OPTIONS"]),
("/.well-known/jwks.json", jwks, ["GET", "HEAD", "OPTIONS"]),
# OAuth core
("/authorize", authorize, ["GET"]),
("/oauth2/token", token, ["POST"]),
("/oauth2/introspect", introspect, ["POST"]),
("/oauth2/revoke", revoke, ["POST"]),
# PRM (general and per-component)
("/.well-known/oauth-protected-resource", prm_root, ["GET", "HEAD", "OPTIONS"]),
("/.well-known/oauth-protected-resource/mcp", prm_mcp, ["GET", "HEAD", "OPTIONS"]),
("/.well-known/oauth-protected-resource/sse", prm_sse, ["GET", "HEAD", "OPTIONS"]),
# OAuth registration
("/oauth2/register", register, ["POST", "OPTIONS"]),
]