add invite_amt_account_link view

1 files changed, 41 insertions, 1 deletions

diff --git a/jb/views/auth.py b/jb/views/auth.py
index f1c2081..1b4c364 100644
--- a/jb/views/auth.py
+++ b/jb/views/auth.py
@@ -1,7 +1,8 @@
+import secrets
 from typing import Annotated
 from urllib.parse import urlencode
 
-from fastapi import APIRouter, Depends, HTTPException, Response, status
+from fastapi import APIRouter, Depends, Header, HTTPException, Response, status
 from fastapi.responses import HTMLResponse, RedirectResponse
 
 from jb.api.auth import (
@@ -10,6 +11,7 @@ from jb.api.auth import (
     get_authenticated_user,
 )
 from jb.api.magic_token import (
+    INVITE_AMT_LINK_TOKEN_TTL,
     consume_amt_account_link_token,
     consume_magic_token,
     create_amt_account_link_token,
@@ -35,6 +37,29 @@ from jb.settings import BASE_HTML
 auth_router = APIRouter(prefix="/auth", tags=["Auth"])
 
 
+def authenticate_invite_amt_account_link(
+    authorization: Annotated[str | None, Header()] = None,
+) -> None:
+    expected_token = settings.invite_amt_account_token
+    if expected_token is None:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="not configured",
+        )
+
+    scheme, _, supplied_token = (authorization or "").partition(" ")
+    authenticated = scheme.lower() == "bearer" and secrets.compare_digest(
+        supplied_token,
+        expected_token.get_secret_value(),
+    )
+    if not authenticated:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid scheduler authentication token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
 @auth_router.post("/magic-link/request/")
 def request_magic_link(body: AccountLogin) -> dict[str, str]:
     """Create a magic link."""
@@ -119,6 +144,21 @@ def link_amt_account(body: AmtAccountLink) -> dict[str, str]:
     return {}
 
 
+@auth_router.post("/link-amt/invite/")
+def invite_amt_account_link(
+    body: AmtAccountLink,
+    _authenticated: Annotated[None, Depends(authenticate_invite_amt_account_link)],
+) -> dict[str, str]:
+    """Create a long-lived AMT account-link URL without sending an email."""
+    token = create_amt_account_link_token(
+        email=str(body.email),
+        amt_worker_id=body.amt_worker_id,
+        ttl_seconds=INVITE_AMT_LINK_TOKEN_TTL,
+    )
+    query = urlencode({"token": token})
+    return {"magic_link": f"{settings.base_url}auth/link-amt/?{query}"}
+
+
 @auth_router.get("/debug/", response_class=HTMLResponse, include_in_schema=False)
 def link_amt_account_landing_page(
     gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],