""" 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}

Cortex

Signed in as {username} — choose a persona

{cards_html}
Account settings
""" 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