tonymandClaude Fable 5.1
59ed6a5951
Build / Build and release (push) Failing after 1s
Port OTM_LOGIN_CSS + the otm-login markup structure from @otm/account-panel otm-login-core.ts 0.44.0 verbatim into login_page() (brand "13ft" / "Read the article without the cruft.", accent --otm-accent:#b5562a copper, otherwise the module's neutral ink defaults). Still OTM-only (no "Other ways to sign in"), fail-closed 503 copy + reasons + tides beacon unchanged. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014LCQmFvKshyAJsE8C3B5ET
309 lines
15 KiB
Python
309 lines
15 KiB
Python
"""OTM first-party plumbing for 13ft: Sign in with OTM, the session cookie,
|
|
health/version, the tides beacon, and the Ideas (Gitea issue) endpoint.
|
|
|
|
Sign in with OTM is the consumer half of the platform's universal login, a
|
|
straight port of pickup/src/sso.js (the control plane's lib/sso-core.ts is the
|
|
canonical format):
|
|
|
|
browser -> https://admin.poweredbyotm.com/sso/authorize?aud=app:13ft&next=/
|
|
OTM checks the account's grant on app:13ft, mints a 60s single-use HS256
|
|
ticket signed with HMAC(master, "otm-sso:app:13ft") — our OTM_SSO_SECRET —
|
|
and 302s to https://13ft.poweredbyotm.com/api/auth/otm-sso?token=...
|
|
|
|
exp is mandatory, aud is checked, the alg header is only ever read to REJECT,
|
|
and verify never raises (a raised parse error on a redirect path would be a
|
|
500, and a distinguishing error would be an oracle).
|
|
|
|
13ft is a fetch proxy, so it is grant-only: no publicAccess, and with no
|
|
secret configured every route fails CLOSED (503), never open.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import html as html_mod
|
|
import json
|
|
import os
|
|
import secrets as _secrets
|
|
import time
|
|
from typing import Any, Optional
|
|
|
|
import requests
|
|
|
|
APP_VERSION = "0.4.2"
|
|
OTM_AUDIENCE = "app:13ft"
|
|
OTM_BASE = os.environ.get("OTM_SSO_AUTHORIZE_BASE", "https://admin.poweredbyotm.com")
|
|
SESSION_COOKIE = "thirteen_session"
|
|
SESSION_TTL_SEC = 7 * 24 * 3600
|
|
|
|
# Paths that never need a session. Everything else — including the catch-all
|
|
# article fetcher — does.
|
|
PUBLIC_PATHS = frozenset({"/api/health", "/login", "/api/auth/otm-sso", "/favicon.ico"})
|
|
|
|
|
|
# ── base64url + HMAC ──────────────────────────────────────────────────────────
|
|
|
|
def _b64url(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
|
|
|
|
|
|
def _b64url_decode(s: str) -> bytes:
|
|
s = s + "=" * (-len(s) % 4)
|
|
return base64.urlsafe_b64decode(s.encode("ascii"))
|
|
|
|
|
|
def _sign(secret: str, signing_input: str) -> str:
|
|
return _b64url(hmac.new(secret.encode("utf-8"), signing_input.encode("utf-8"), hashlib.sha256).digest())
|
|
|
|
|
|
# ── the OTM ticket ────────────────────────────────────────────────────────────
|
|
|
|
def verify_sso_ticket(secret: str, token: Any, expected_aud: Optional[str] = None, now_ms: Optional[int] = None) -> Optional[dict]:
|
|
"""The ticket's payload, or None for anything that isn't a valid, unexpired
|
|
HS256 ticket for `expected_aud`. Never raises."""
|
|
try:
|
|
parts = str(token).split(".")
|
|
if len(parts) != 3 or not all(parts):
|
|
return None
|
|
header_b64, payload_b64, sig_b64 = parts
|
|
header = json.loads(_b64url_decode(header_b64))
|
|
if not isinstance(header, dict) or header.get("alg") != "HS256":
|
|
return None
|
|
expected = _sign(secret, f"{header_b64}.{payload_b64}")
|
|
if not hmac.compare_digest(expected.encode("utf-8"), sig_b64.encode("utf-8")):
|
|
return None
|
|
payload = json.loads(_b64url_decode(payload_b64))
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
exp = payload.get("exp")
|
|
if not isinstance(exp, (int, float)) or isinstance(exp, bool):
|
|
return None
|
|
now_sec = (now_ms if now_ms is not None else int(time.time() * 1000)) // 1000
|
|
if now_sec >= exp:
|
|
return None
|
|
for k in ("jti", "sub", "aud", "role", "email"):
|
|
if not isinstance(payload.get(k), str) or not payload[k]:
|
|
return None
|
|
if expected_aud is not None and payload["aud"] != expected_aud:
|
|
return None
|
|
return payload
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def mint_sso_ticket(secret: str, claims: dict, ttl_sec: int = 60, now_ms: Optional[int] = None) -> str:
|
|
"""Test/dev helper — mint a ticket the way the control plane does."""
|
|
now = (now_ms if now_ms is not None else int(time.time() * 1000)) // 1000
|
|
header = _b64url(json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode())
|
|
body = dict(claims)
|
|
body.setdefault("jti", _secrets.token_urlsafe(12))
|
|
body["iat"] = now
|
|
body["exp"] = now + ttl_sec
|
|
payload = _b64url(json.dumps(body, separators=(",", ":")).encode())
|
|
return f"{header}.{payload}.{_sign(secret, f'{header}.{payload}')}"
|
|
|
|
|
|
class JtiGuard:
|
|
"""Single-use jti guard. Tickets live 60s and gunicorn runs ONE sync worker
|
|
here (gunicorn.conf.py), so an in-memory map with a short TTL is a complete
|
|
replay guard. If workers are ever raised above one, move this to a shared
|
|
store — per-worker memory would let a ticket replay against another worker."""
|
|
|
|
def __init__(self, ttl_sec: int = 600):
|
|
self.ttl = ttl_sec
|
|
self.seen: dict[str, float] = {}
|
|
|
|
def claim(self, jti: str, now: Optional[float] = None) -> bool:
|
|
now = time.time() if now is None else now
|
|
if len(self.seen) > 5000:
|
|
for k, at in list(self.seen.items()):
|
|
if now - at > self.ttl:
|
|
del self.seen[k]
|
|
at = self.seen.get(jti)
|
|
if at is not None and now - at < self.ttl:
|
|
return False
|
|
self.seen[jti] = now
|
|
return True
|
|
|
|
|
|
# ── the session cookie ────────────────────────────────────────────────────────
|
|
# {sub, email, name?, exp} as base64url JSON + HMAC under the session secret.
|
|
|
|
def session_secret() -> Optional[str]:
|
|
"""AUTH_SECRET (injected by the platform for needsAuthSecret apps), else the
|
|
SSO secret itself. None = not configured → the app fails closed."""
|
|
for k in ("AUTH_SECRET", "OTM_SSO_SECRET"):
|
|
v = (os.environ.get(k) or "").strip()
|
|
if v:
|
|
return v
|
|
return None
|
|
|
|
|
|
def encode_session(secret: str, s: dict) -> str:
|
|
body = _b64url(json.dumps(s, separators=(",", ":")).encode())
|
|
return f"{body}.{_sign(secret, body)}"
|
|
|
|
|
|
def decode_session(secret: str, value: Any, now_ms: Optional[int] = None) -> Optional[dict]:
|
|
try:
|
|
body, sig = str(value or "").split(".", 1)
|
|
if not body or not sig:
|
|
return None
|
|
if not hmac.compare_digest(_sign(secret, body).encode(), sig.encode()):
|
|
return None
|
|
s = json.loads(_b64url_decode(body))
|
|
if not isinstance(s, dict):
|
|
return None
|
|
exp = s.get("exp")
|
|
now_sec = (now_ms if now_ms is not None else int(time.time() * 1000)) // 1000
|
|
if not isinstance(exp, (int, float)) or isinstance(exp, bool) or now_sec >= exp:
|
|
return None
|
|
if not isinstance(s.get("email"), str):
|
|
return None
|
|
return s
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def safe_next(next_path: Any, default: str = "/") -> str:
|
|
"""Path-only, single leading slash — never an open redirect."""
|
|
n = str(next_path or "")
|
|
if n.startswith("/") and not n.startswith("//") and not n.startswith("/\\"):
|
|
return n
|
|
return default
|
|
|
|
|
|
def otm_sign_in_url(next_path: str = "/") -> str:
|
|
from urllib.parse import urlencode
|
|
return f"{OTM_BASE.rstrip('/')}/sso/authorize?" + urlencode({"aud": OTM_AUDIENCE, "next": safe_next(next_path)})
|
|
|
|
|
|
# ── tides + footer + ideas ────────────────────────────────────────────────────
|
|
|
|
def tides_beacon() -> str:
|
|
host = (os.environ.get("TIDES_HOST") or "").strip().rstrip("/")
|
|
site = (os.environ.get("TIDES_SITE_ID") or "").strip()
|
|
if not host or not site:
|
|
return ""
|
|
return f'<script defer src="{html_mod.escape(host)}/t.js" data-site="{html_mod.escape(site)}"></script>'
|
|
|
|
|
|
def footer_html(email: Optional[str]) -> str:
|
|
who = f'<span>{html_mod.escape(email)}</span> · <a href="/logout">Sign out</a> · ' if email else ""
|
|
ideas = (
|
|
'<form id="otm-idea" method="post" action="/api/ideas" style="display:inline">'
|
|
'<input name="title" maxlength="140" placeholder="💡 an idea for 13ft…" '
|
|
'style="font:inherit;padding:.2em .5em;border:1px solid #8884;border-radius:6px;background:transparent;color:inherit;width:14em">'
|
|
' <button type="submit" style="font:inherit;padding:.2em .6em;border:1px solid #8884;border-radius:6px;background:transparent;color:inherit;cursor:pointer">Send</button>'
|
|
'</form>'
|
|
if ideas_configured() else ""
|
|
)
|
|
return (
|
|
'<footer style="margin:2.5rem auto 1rem;max-width:60rem;padding:0 1rem;font-size:.8rem;opacity:.75;'
|
|
'display:flex;flex-wrap:wrap;gap:.6em;align-items:center;justify-content:space-between">'
|
|
f'<div>{who}<a href="/changelog">v{APP_VERSION}</a> · Powered by OTM</div>'
|
|
f'<div>{ideas}</div></footer>'
|
|
)
|
|
|
|
|
|
def ideas_configured() -> bool:
|
|
return bool(os.environ.get("GITEA_URL") and os.environ.get("GITEA_TOKEN") and os.environ.get("GITEA_REPO"))
|
|
|
|
|
|
def create_idea(title: str, who: str, page: str = "/") -> tuple[bool, str]:
|
|
"""File an idea as a Gitea issue on this app's repo (GITEA_URL/TOKEN/REPO
|
|
injected by the platform's needsSuggestions). Returns (ok, message/url)."""
|
|
t = (title or "").strip()[:140]
|
|
if not t:
|
|
return False, "A title is required."
|
|
if not ideas_configured():
|
|
return False, "Ideas isn't configured."
|
|
base = os.environ["GITEA_URL"].rstrip("/")
|
|
repo = os.environ["GITEA_REPO"]
|
|
headers = {"Authorization": f"token {os.environ['GITEA_TOKEN']}", "Content-Type": "application/json", "Accept": "application/json"}
|
|
body = f"{t}\n\n---\nFiled from 13ft's Ideas box by {who or 'someone signed in'} on `{page}` (v{APP_VERSION})."
|
|
try:
|
|
r = requests.post(f"{base}/api/v1/repos/{repo}/issues", headers=headers, json={"title": t, "body": body}, timeout=10)
|
|
if r.status_code >= 300:
|
|
return False, f"Gitea said {r.status_code}."
|
|
return True, r.json().get("html_url", "")
|
|
except Exception:
|
|
return False, "Couldn't reach Gitea."
|
|
|
|
|
|
# ── small pages ───────────────────────────────────────────────────────────────
|
|
|
|
SSO_REASONS = {
|
|
"not_enabled": "Sign in with OTM isn't configured on this deployment.",
|
|
"invalid_token": "That OTM sign-in link is no longer valid. Try again.",
|
|
"link_already_used": "That OTM sign-in link was already used. Try again.",
|
|
"server_error": "Something went wrong completing OTM sign-in. Try again.",
|
|
}
|
|
|
|
|
|
# Ported from @otm/account-panel otm-login-core.ts 0.44.0 (OTM_LOGIN_CSS) —
|
|
# keep in sync, verbatim. 13ft is OTM-only (no local accounts), so only the
|
|
# .otm-login rules are needed — no .otm-menu/.otm-badge here.
|
|
OTM_LOGIN_CSS = """
|
|
.otm-login{--otm-accent:#b5562a;--otm-accent-ink:#fff;--otm-bg:#faf8f5;--otm-card:#fff;--otm-fg:#1f1b18;--otm-muted:#6b6560;--otm-line:rgba(31,27,24,.12);--otm-radius:20px;--otm-font-display:inherit;
|
|
min-height:100vh;display:flex;align-items:center;justify-content:center;padding:24px 16px;background:var(--otm-bg);color:var(--otm-fg);box-sizing:border-box}
|
|
@media (prefers-color-scheme:dark){.otm-login{--otm-bg:#161412;--otm-card:#211e1b;--otm-fg:#f3efe9;--otm-muted:#a39b92;--otm-line:rgba(243,239,233,.12)}}
|
|
.otm-login *{box-sizing:border-box}
|
|
.otm-login__col{width:100%;max-width:26rem}
|
|
.otm-login__brand{text-align:center;margin-bottom:24px}
|
|
.otm-login__mark{width:56px;height:56px;border-radius:999px;background:var(--otm-accent);color:var(--otm-accent-ink);display:inline-flex;align-items:center;justify-content:center;font-weight:700;font-size:20px;letter-spacing:.02em;margin-bottom:12px}
|
|
.otm-login__mark svg,.otm-login__mark img{width:28px;height:28px}
|
|
.otm-login__name{font-family:var(--otm-font-display);font-size:30px;line-height:1.1;letter-spacing:-.01em;font-weight:600;margin:0}
|
|
.otm-login__tagline{color:var(--otm-muted);font-size:14px;margin:6px 0 0}
|
|
.otm-login__card{background:var(--otm-card);border:1px solid var(--otm-line);border-radius:var(--otm-radius);padding:24px;box-shadow:0 4px 24px rgba(0,0,0,.06)}
|
|
.otm-login__btn{display:flex;align-items:center;justify-content:center;width:100%;height:44px;border-radius:12px;border:0;background:var(--otm-accent);color:var(--otm-accent-ink);font:inherit;font-weight:600;font-size:15px;text-decoration:none;cursor:pointer}
|
|
.otm-login__btn:hover{filter:brightness(1.06)}
|
|
.otm-login__btn--outline{background:transparent;color:var(--otm-fg);border:1px solid var(--otm-line)}
|
|
.otm-login__btn--outline:hover{background:rgba(127,127,127,.06);filter:none}
|
|
.otm-login__btn[disabled]{opacity:.55;cursor:default}
|
|
.otm-login__helper{color:var(--otm-muted);font-size:14px;text-align:center;margin:12px 0 0;line-height:1.45}
|
|
.otm-login__error{color:#b3556b;font-size:14px;margin:0 0 12px;text-align:center}
|
|
.otm-login__link{display:inline-block;background:none;border:0;padding:0;font:inherit;font-size:14px;color:var(--otm-muted);text-decoration:underline;text-underline-offset:3px;cursor:pointer}
|
|
.otm-login__link:hover{color:var(--otm-fg)}
|
|
.otm-login__foot{text-align:center;margin-top:16px}
|
|
.otm-login__stack>*+*{margin-top:10px}
|
|
"""
|
|
|
|
|
|
def login_page(next_path: str, reason: Optional[str], configured: bool) -> str:
|
|
msg = ""
|
|
if reason:
|
|
msg = f'<p class="otm-login__error" role="alert">{html_mod.escape(SSO_REASONS.get(reason, "Sign-in failed."))}</p>'
|
|
if not configured:
|
|
msg = '<p class="otm-login__error" role="alert">This 13ft isn\'t wired to OTM yet, so nobody can sign in. (Fails closed on purpose — it\'s a fetch proxy.)</p>'
|
|
# OTM-only app (no local accounts) — the "Sign in with OTM" button is the
|
|
# whole card, with the passkey helper line under it. There is no "Other
|
|
# ways to sign in": there is no other way.
|
|
card = msg + (
|
|
f'<a class="otm-login__btn" href="{html_mod.escape(otm_sign_in_url(next_path))}">Sign in with OTM</a>'
|
|
'<p class="otm-login__helper">Your OTM account — Face ID, Touch ID or a passkey works there.</p>'
|
|
if configured else ""
|
|
)
|
|
return (
|
|
'<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">'
|
|
f'<title>13ft — sign in</title><style>{OTM_LOGIN_CSS}</style></head>'
|
|
'<body>'
|
|
'<div class="otm-login" style="--otm-accent:#b5562a;--otm-accent-ink:#fff">'
|
|
'<div class="otm-login__col">'
|
|
'<div class="otm-login__brand"><div class="otm-login__mark">13</div>'
|
|
'<h1 class="otm-login__name">13ft</h1><p class="otm-login__tagline">Read the article without the cruft.</p></div>'
|
|
f'<div class="otm-login__card">{card}</div>'
|
|
'</div></div>'
|
|
f'{tides_beacon()}</body></html>'
|
|
)
|
|
|
|
|
|
def changelog_page(markdown_text: str) -> str:
|
|
return (
|
|
'<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">'
|
|
f'<title>13ft — changelog</title><style>:root{{color-scheme:light dark}}body{{font-family:system-ui,sans-serif;max-width:40rem;margin:6vh auto;padding:0 1rem;line-height:1.5}}pre{{white-space:pre-wrap}}</style></head>'
|
|
f'<body><p><a href="/">← 13ft</a></p><h1>Changelog</h1><pre>{html_mod.escape(markdown_text)}</pre>{tides_beacon()}</body></html>'
|
|
)
|