"""
UI router — serves the web interface and handles login/logout.
Routes:
GET / → redirect to /{user}/{persona} if logged in, else /login
GET /login → login page
POST /login → validate credentials, set cookie, redirect
POST /logout → clear cookie, redirect to /login
GET /{user}/{persona} → serve index.html with CORTEX_CONFIG injected
GET /{user}/{persona}/ → same (trailing slash)
"""
import logging
import re
from pathlib import Path
import jwt
from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from auth_utils import COOKIE_NAME, check_credentials, create_token, decode_token
from persona import list_users, list_user_personas, validate as validate_persona, persona_path
logger = logging.getLogger(__name__)
router = APIRouter()
_STATIC = Path(__file__).parent.parent / "static"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _get_session_user(request: Request) -> str | None:
"""Return the authenticated username from the session cookie, or None."""
token = request.cookies.get(COOKIE_NAME)
if not token:
return None
try:
return decode_token(token)
except jwt.InvalidTokenError:
return None
def _set_cookie(response: Response, username: str) -> None:
from auth_utils import create_token
from config import settings
token = create_token(username)
response.set_cookie(
COOKIE_NAME,
token,
max_age=settings.jwt_expire_days * 86400,
httponly=True,
samesite="lax",
secure=False, # set True in production behind HTTPS
)
_LAST_PERSONA_COOKIE = "cx_last_persona"
def _first_persona(username: str) -> str | None:
"""Return the first available persona for a user, or None."""
names = list_user_personas(username)
return names[0] if names else None
def _preferred_persona(request: Request, username: str) -> str | None:
"""Return the last-visited persona from cookie if valid, else the first available."""
names = list_user_personas(username)
if not names:
return None
cookie_val = request.cookies.get(_LAST_PERSONA_COOKIE, "")
if cookie_val in names:
return cookie_val
return names[0]
# ---------------------------------------------------------------------------
# Favicon — default sparkle; persona pages override via JS
# ---------------------------------------------------------------------------
_FAVICON_SVG = (
""
)
@router.get("/favicon.ico", include_in_schema=False)
async def favicon():
return Response(content=_FAVICON_SVG, media_type="image/svg+xml")
@router.get("/sw.js", include_in_schema=False)
async def service_worker():
from fastapi.responses import FileResponse
return FileResponse(str(_STATIC / "sw.js"), media_type="application/javascript")
@router.get("/manifest.json", include_in_schema=False)
async def web_manifest():
from fastapi.responses import FileResponse
return FileResponse(str(_STATIC / "manifest.json"), media_type="application/manifest+json")
# ---------------------------------------------------------------------------
# Root redirect
# ---------------------------------------------------------------------------
@router.get("/", include_in_schema=False)
async def root(request: Request):
user = _get_session_user(request)
if not user:
return RedirectResponse("/login", status_code=302)
persona = _preferred_persona(request, user)
if not persona:
return HTMLResponse("
No personas configured for your account.
", status_code=500)
return RedirectResponse(f"/{user}/{persona}", status_code=302)
# ---------------------------------------------------------------------------
# Login / logout
# ---------------------------------------------------------------------------
@router.get("/login", include_in_schema=False)
async def login_page(request: Request):
user = _get_session_user(request)
if user:
# Already logged in — redirect home
persona = _preferred_persona(request, user)
if persona:
return RedirectResponse(f"/{user}/{persona}", status_code=302)
return HTMLResponse((_STATIC / "login.html").read_text())
@router.post("/login", include_in_schema=False)
async def login(
request: Request,
username: str = Form(...),
password: str = Form(...),
):
if not check_credentials(username, password):
logger.warning("failed login attempt for user: %s", username)
html = (_STATIC / "login.html").read_text().replace(
"",
'Invalid username or password.
',
)
return HTMLResponse(html, status_code=401)
persona = _first_persona(username)
if not persona:
return HTMLResponse("No personas configured for your account.
", status_code=500)
logger.info("login: %s", username)
resp = RedirectResponse(f"/{username}/{persona}", status_code=302)
_set_cookie(resp, username)
return resp
@router.post("/logout", include_in_schema=False)
async def logout():
resp = RedirectResponse("/login", status_code=302)
resp.delete_cookie(COOKIE_NAME)
return resp
# ---------------------------------------------------------------------------
# User landing — /{username} → persona picker
# ---------------------------------------------------------------------------
@router.get("/{username}", include_in_schema=False)
async def user_landing(username: str, request: Request):
session_user = _get_session_user(request)
if not session_user:
return RedirectResponse("/login", status_code=302)
if session_user != username:
return RedirectResponse(f"/{session_user}", status_code=302)
personas = list_user_personas(username)
if not personas:
return HTMLResponse("No personas configured.
", status_code=404)
cards_html = ""
for p in personas:
emoji = "✨"
identity_path = persona_path(username, p) / "IDENTITY.md"
if identity_path.exists():
m = re.search(r"\|\s*Emoji\s*\|\s*(.+?)\s*\|", identity_path.read_text())
if m:
emoji = m.group(1).strip()
cards_html += (
f''
f'{emoji}'
f'{p.capitalize()}'
f'\n'
)
html = f"""
Cortex — {username}
"""
return HTMLResponse(html)
# ---------------------------------------------------------------------------
# Main UI — /{username}/{persona}
# ---------------------------------------------------------------------------
@router.get("/api/personas", tags=["ui"])
async def api_personas(request: Request) -> dict:
"""Return the list of personas for the current session user."""
user = _get_session_user(request)
if not user:
from fastapi import HTTPException
raise HTTPException(status_code=401, detail="Not authenticated")
personas_with_emoji = []
for p in list_user_personas(user):
emoji = "✨"
identity_path = persona_path(user, p) / "IDENTITY.md"
if identity_path.exists():
m = re.search(r"\|\s*Emoji\s*\|\s*(.+?)\s*\|", identity_path.read_text())
if m:
emoji = m.group(1).strip()
personas_with_emoji.append({"name": p, "emoji": emoji})
return {"user": user, "personas": personas_with_emoji}
@router.get("/{username}/{persona}", include_in_schema=False)
@router.get("/{username}/{persona}/", include_in_schema=False)
async def serve_ui(username: str, persona: str, request: Request):
# Auth check
session_user = _get_session_user(request)
if not session_user:
return RedirectResponse("/login", status_code=302)
if session_user != username:
return RedirectResponse(f"/{session_user}/{_first_persona(session_user) or ''}", status_code=302)
# Validate persona exists
try:
validate_persona(username, persona)
except ValueError:
return RedirectResponse(f"/{username}/{_first_persona(username) or ''}", status_code=302)
# Read emoji from IDENTITY.md (| Emoji | | line)
emoji = "✨"
identity_path = persona_path(username, persona) / "IDENTITY.md"
if identity_path.exists():
m = re.search(r"\|\s*Emoji\s*\|\s*(.+?)\s*\|", identity_path.read_text())
if m:
emoji = m.group(1).strip()
# Serve index.html with user/persona/emoji injected
html = (_STATIC / "index.html").read_text()
config_tag = (
f''
)
html = html.replace("", f"{config_tag}\n", 1)
resp = HTMLResponse(html)
resp.set_cookie(_LAST_PERSONA_COOKIE, persona, max_age=365 * 86400, httponly=False, samesite="lax")
return resp