""" auth.py - the HTTP surface over identity.py: login, invite acceptance, account. Deliberately thin. Every rule (single-use invites, throttling, session bounds, capabilities) lives in identity.py so the leader console and any future API client inherit them rather than reimplementing them. This module only turns those rules into pages and cookies. It does not import app.py. The page shell is injected at include time (`auth.PAGE = page`), because app.py imports this module and the reverse would be circular. If PAGE is unset the pages still render, just unstyled - an identity layer that cannot be signed into because a renderer is missing would be a worse failure than a plain page. """ import html import os from fastapi import APIRouter, Form, Request from fastapi.responses import HTMLResponse, RedirectResponse import identity router = APIRouter(tags=["auth"]) COOKIE = "s73_session" COOKIE_SECURE = identity.SITE_BASE_URL.startswith("https://") # Set by app.py after page() is defined. PAGE = None def _esc(s): return html.escape(str(s)) if s else "" def _render(title, body): if PAGE: return PAGE(title, body) return "%s%s" % (_esc(title), body) def _client_ip(request): """Real client address. The app sits behind NPM, so request.client.host is the proxy on every hit and would throttle the whole site as one address. The LAST hop, not the first. NPM sets the header with $proxy_add_x_forwarded_for, which APPENDS the connecting address to whatever the client sent, so the first hop is client-controlled and the last is the one NPM vouches for. With exactly one trusted proxy in front (the app is published on 127.0.0.1 and the docker network only), the rightmost address is the real client. Reading the first hop would let an outsider claim a LAN address with one header - and that is exactly what admin_api.client_is_lan decides on.""" fwd = request.headers.get("x-forwarded-for", "") if fwd: return fwd.split(",")[-1].strip() return request.client.host if request.client else None def current_person(request): """The signed-in person, or None. This is the single seam other modules use; documents.visible() attaches here in P1.""" return identity.session_person(request.cookies.get(COOKIE)) FORM_CSS = """""" def _shell(heading, intro, inner, error=None): err = '
%s
' % _esc(error) if error else "" return f"""{FORM_CSS}

{heading}

{intro}

{err}{inner}
""" # --------------------------------------------------------------------------- # Login # --------------------------------------------------------------------------- def _login_form(email="", error=None, next_url=""): return _shell( "Sign in", "For Pack 73 and Troop 73 leaders and families.", f"""

Accounts are created by invitation. If you need one, ask a leader.

""", error) @router.get("/login", response_class=HTMLResponse) def login_form(request: Request): # `next` arrives from the leader console's 401 redirect (next=/leaders/). # identity.safe_next() keeps it same-origin; anything odd lands on /account. nxt = identity.safe_next(request.query_params.get("next")) if current_person(request): return RedirectResponse(url=nxt, status_code=303) return HTMLResponse(_render("Sign in", _login_form(next_url=nxt))) @router.post("/login") def login(request: Request, email: str = Form(""), password: str = Form(""), next: str = Form("")): nxt = identity.safe_next(next) try: person, token = identity.authenticate( email, password, ip=_client_ip(request), user_agent=request.headers.get("user-agent")) except identity.IdentityError as e: return HTMLResponse(_render("Sign in", _login_form(email, e.detail, nxt)), status_code=e.status) resp = RedirectResponse(url=nxt, status_code=303) resp.set_cookie(COOKIE, token, max_age=identity.SESSION_ABSOLUTE_DAYS * 86400, httponly=True, secure=COOKIE_SECURE, samesite="lax", path="/") return resp @router.post("/logout") def logout(request: Request): tok = request.cookies.get(COOKIE) if tok: person = identity.session_person(tok) identity.end_session(tok) identity.log_event("logout", person_id=person["id"] if person else None, ip=_client_ip(request)) resp = RedirectResponse(url="/", status_code=303) resp.delete_cookie(COOKIE, path="/") return resp # --------------------------------------------------------------------------- # Invitations # --------------------------------------------------------------------------- # # Expired, revoked, consumed and never-existed all render the same page. The # difference is not the visitor's business, and telling them would confirm # which addresses belong to real families. DEAD_INVITE = ("This invitation link is no longer valid. It may have been used " "already, replaced by a newer one, or expired. Ask whoever invited " "you to send a fresh link.") def _invite_form(token, invite, values=None, error=None): v = values or {} return _shell( "Finish setting up your account", "Invitation for %s." % _esc(invite["email"]), f"""

At least 12 characters. A short phrase you will remember beats a short password you will not.

""", error) @router.get("/invite/{token}", response_class=HTMLResponse) def invite_form(request: Request, token: str): invite = identity.peek_invite(token) if not invite: return HTMLResponse(_render("Invitation", _shell("Invitation", "", "", DEAD_INVITE)), status_code=410) return HTMLResponse(_render("Finish setting up your account", _invite_form(token, invite))) @router.post("/invite/{token}") def invite_accept(request: Request, token: str, full_name: str = Form(""), preferred_name: str = Form(""), phone: str = Form(""), password: str = Form(""), confirm: str = Form("")): invite = identity.peek_invite(token) if not invite: return HTMLResponse(_render("Invitation", _shell("Invitation", "", "", DEAD_INVITE)), status_code=410) vals = dict(full_name=full_name, preferred_name=preferred_name, phone=phone) if password != confirm: return HTMLResponse(_render("Finish setting up your account", _invite_form(token, invite, vals, "Those two passwords do not match.")), status_code=422) try: person = identity.consume_invite(token, full_name, password, preferred_name=preferred_name, phone=phone, ip=_client_ip(request)) except identity.IdentityError as e: if e.status == 410: return HTMLResponse(_render("Invitation", _shell("Invitation", "", "", DEAD_INVITE)), status_code=410) return HTMLResponse(_render("Finish setting up your account", _invite_form(token, invite, vals, e.detail)), status_code=e.status) tok = identity.start_session(person["id"], ip=_client_ip(request), user_agent=request.headers.get("user-agent")) resp = RedirectResponse(url="/account", status_code=303) resp.set_cookie(COOKIE, tok, max_age=identity.SESSION_ABSOLUTE_DAYS * 86400, httponly=True, secure=COOKIE_SECURE, samesite="lax", path="/") return resp # --------------------------------------------------------------------------- # Account # --------------------------------------------------------------------------- ROLE_LABEL = {"owner": "Site owner", "admin": "Administrator", "leader": "Leader", "member": "Member"} @router.get("/account", response_class=HTMLResponse) def account(request: Request): person = current_person(request) if not person: return RedirectResponse(url="/login", status_code=303) rows = [] if person.get("global_role"): rows.append('
Site-wide' '%s
' % _esc(ROLE_LABEL.get(person["global_role"], person["global_role"]))) for m in person["memberships"]: title = " ยท %s" % _esc(m["title"]) if m["title"] else "" rows.append('
%s' '%s%s
' % (_esc(m["short_name"]), _esc(ROLE_LABEL.get(m["role"], m["role"])), title)) if not rows: rows.append('
No roles assigned yet.
') name = person.get("preferred_name") or person.get("full_name") or person["email"] body = _shell( "Your account", _esc(person["email"]), f"""
{''.join(rows)}

Changing your own details is not built yet. Ask an administrator.

""") return HTMLResponse(_render("Your account", body.replace( "

Your account

", "

Hello, %s

" % _esc(name))))