154 lines
5.0 KiB
Python
154 lines
5.0 KiB
Python
"""
|
|
web/routers/auth.py
|
|
-------------------
|
|
OAuth endpoints for Adobe Sign and DocuSign.
|
|
|
|
Adobe Sign: Authorization Code flow
|
|
DocuSign: Authorization Code flow (demo sandbox)
|
|
|
|
Tokens are stored in a signed session cookie.
|
|
"""
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import JSONResponse, RedirectResponse
|
|
|
|
from web.config import settings
|
|
from web.session import get_session, save_session, clear_session
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/status")
|
|
def auth_status(request: Request):
|
|
"""Returns which platforms the current session is connected to."""
|
|
session = get_session(request)
|
|
return {
|
|
"adobe": bool(session.get("adobe_access_token")),
|
|
"docusign": bool(session.get("docusign_access_token")),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adobe Sign
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/adobe/start")
|
|
def adobe_start():
|
|
"""Redirect the browser to the Adobe Sign OAuth authorization page."""
|
|
params = (
|
|
f"?response_type=code"
|
|
f"&client_id={settings.adobe_client_id}"
|
|
f"&redirect_uri={settings.adobe_redirect_uri}"
|
|
f"&scope=library_read:self+library_write:self+user_read:self"
|
|
)
|
|
auth_url = "https://secure.eu2.adobesign.com/public/oauth/v2" + params
|
|
return RedirectResponse(auth_url)
|
|
|
|
|
|
@router.get("/adobe/callback")
|
|
async def adobe_callback(request: Request, code: str = ""):
|
|
"""Exchange authorization code for access + refresh tokens."""
|
|
if not code:
|
|
return JSONResponse({"error": "missing code"}, status_code=400)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
"https://api.eu2.adobesign.com/oauth/v2/token",
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"client_id": settings.adobe_client_id,
|
|
"client_secret": settings.adobe_client_secret,
|
|
"redirect_uri": settings.adobe_redirect_uri,
|
|
"code": code,
|
|
},
|
|
)
|
|
|
|
if not resp.is_success:
|
|
return JSONResponse({"error": "token exchange failed", "detail": resp.text}, status_code=502)
|
|
|
|
token_data = resp.json()
|
|
session = get_session(request)
|
|
session["adobe_access_token"] = token_data.get("access_token")
|
|
session["adobe_refresh_token"] = token_data.get("refresh_token")
|
|
|
|
response = RedirectResponse("/")
|
|
save_session(response, session)
|
|
return response
|
|
|
|
|
|
@router.get("/adobe/disconnect")
|
|
def adobe_disconnect(request: Request):
|
|
session = get_session(request)
|
|
session.pop("adobe_access_token", None)
|
|
session.pop("adobe_refresh_token", None)
|
|
response = JSONResponse({"disconnected": "adobe"})
|
|
save_session(response, session)
|
|
return response
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DocuSign
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/docusign/start")
|
|
def docusign_start():
|
|
"""Redirect the browser to the DocuSign OAuth authorization page."""
|
|
params = (
|
|
f"?response_type=code"
|
|
f"&scope=signature"
|
|
f"&client_id={settings.docusign_client_id}"
|
|
f"&redirect_uri={settings.docusign_redirect_uri}"
|
|
)
|
|
auth_url = f"https://{settings.docusign_auth_server}/oauth/auth" + params
|
|
return RedirectResponse(auth_url)
|
|
|
|
|
|
@router.get("/docusign/callback")
|
|
async def docusign_callback(request: Request, code: str = ""):
|
|
"""Exchange authorization code for access token."""
|
|
if not code:
|
|
return JSONResponse({"error": "missing code"}, status_code=400)
|
|
|
|
import base64
|
|
credentials = base64.b64encode(
|
|
f"{settings.docusign_client_id}:{settings.docusign_client_secret}".encode()
|
|
).decode()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"https://{settings.docusign_auth_server}/oauth/token",
|
|
headers={"Authorization": f"Basic {credentials}"},
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": settings.docusign_redirect_uri,
|
|
},
|
|
)
|
|
|
|
if not resp.is_success:
|
|
return JSONResponse({"error": "token exchange failed", "detail": resp.text}, status_code=502)
|
|
|
|
token_data = resp.json()
|
|
session = get_session(request)
|
|
session["docusign_access_token"] = token_data.get("access_token")
|
|
session["docusign_refresh_token"] = token_data.get("refresh_token")
|
|
|
|
response = RedirectResponse("/")
|
|
save_session(response, session)
|
|
return response
|
|
|
|
|
|
@router.get("/docusign/disconnect")
|
|
def docusign_disconnect(request: Request):
|
|
session = get_session(request)
|
|
session.pop("docusign_access_token", None)
|
|
session.pop("docusign_refresh_token", None)
|
|
response = JSONResponse({"disconnected": "docusign"})
|
|
save_session(response, session)
|
|
return response
|