"""
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}
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.
')
return HTMLResponse(_render("Your account", _account_page(person, request.query_params.get("done"))))
def _account_page(person, done=None, error=None, error_form=None):
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"]
flash = {"details": "Details saved.", "password": "Password changed. Your other devices were signed out."}.get(done or "")
err_d = '%s
' % _esc(error) if error and error_form == "details" else ""
err_p = '%s
' % _esc(error) if error and error_form == "password" else ""
body = _shell(
"Your account", _esc(person["email"]),
f"""{'%s
' % _esc(flash) if flash else ''}
{''.join(rows)}
Your email address and roles are set by an administrator.
""")
return body.replace(
"Your account
",
"Hello, %s
" % _esc(name))
@router.post("/account/details")
def account_details(request: Request, full_name: str = Form(""), preferred_name: str = Form(""),
phone: str = Form("")):
person = current_person(request)
if not person:
return RedirectResponse(url="/login", status_code=303)
try:
identity.update_own_details(person["id"], full_name, preferred_name, phone)
except identity.IdentityError as e:
return HTMLResponse(_render("Your account", _account_page(person, error=e.detail, error_form="details")),
status_code=e.status)
return RedirectResponse(url="/account?done=details", status_code=303)
@router.post("/account/password")
def account_password(request: Request, current: str = Form(""), new: str = Form(""), confirm: str = Form("")):
person = current_person(request)
if not person:
return RedirectResponse(url="/login", status_code=303)
if new != confirm:
return HTMLResponse(_render("Your account", _account_page(person, error="Those two passwords do not match.",
error_form="password")), status_code=422)
try:
identity.change_password(person["id"], current, new, ip=_client_ip(request))
except identity.IdentityError as e:
return HTMLResponse(_render("Your account", _account_page(person, error=e.detail, error_form="password")),
status_code=e.status)
# End the other sessions, keep this one: the person is still here.
tok = request.cookies.get(COOKIE)
con = identity.connect()
try:
con.execute("UPDATE sessions SET revoked_at=? WHERE person_id=? AND revoked_at IS NULL AND token_hash<>?",
(identity._now(), person["id"], identity._hash_token(tok or "")))
con.commit()
finally:
con.close()
return RedirectResponse(url="/account?done=password", status_code=303)
# ---------------------------------------------------------------------------
# Password reset by link. The link is minted by an admin on the console and
# handed over out of band; there is no outbound mail. Same posture as
# invites: expired, used, revoked and never-existed all read the same.
# ---------------------------------------------------------------------------
DEAD_RESET = ("This reset link is no longer valid. It may have been used already, replaced by "
"a newer one, or expired. Ask an administrator for a fresh link.")
def _reset_form(token, person, error=None):
return _shell(
"Choose a new password", "For %s." % _esc(person["email"]),
f"""
At least 12 characters. Every device signed in as you will be signed out.
""",
error)
@router.get("/reset/{token}", response_class=HTMLResponse)
def reset_form(request: Request, token: str):
person = identity.peek_reset(token)
if not person:
return HTMLResponse(_render("Reset", _shell("Reset", "", "", DEAD_RESET)), status_code=410)
return HTMLResponse(_render("Choose a new password", _reset_form(token, person)))
@router.post("/reset/{token}")
def reset_accept(request: Request, token: str, password: str = Form(""), confirm: str = Form("")):
person = identity.peek_reset(token)
if not person:
return HTMLResponse(_render("Reset", _shell("Reset", "", "", DEAD_RESET)), status_code=410)
if password != confirm:
return HTMLResponse(_render("Choose a new password",
_reset_form(token, person, "Those two passwords do not match.")), status_code=422)
try:
person = identity.consume_reset(token, password, ip=_client_ip(request))
except identity.IdentityError as e:
if e.status == 410:
return HTMLResponse(_render("Reset", _shell("Reset", "", "", DEAD_RESET)), status_code=410)
return HTMLResponse(_render("Choose a new password", _reset_form(token, person, 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