"""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.1"
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''
def footer_html(email: Optional[str]) -> str:
who = f'{html_mod.escape(email)} · Sign out · ' if email else ""
ideas = (
''
if ideas_configured() else ""
)
return (
''
)
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.",
}
def login_page(next_path: str, reason: Optional[str], configured: bool) -> str:
msg = ""
if reason:
msg = f'{html_mod.escape(SSO_REASONS.get(reason, "Sign-in failed."))}
'
if not configured:
msg = 'This 13ft isn\'t wired to OTM yet, so nobody can sign in. (Fails closed on purpose — it\'s a fetch proxy.)
'
# OTM-only app (no local accounts) — step 2's primary button is the whole
# card, with the passkey helper line under it. There is no step 3 ("Other
# ways to sign in"): there is no other way.
card = msg + (
f'Sign in with OTM'
''
'Your OTM account — Face ID, Touch ID or a passkey works there.
'
if configured else ""
)
return (
''
'13ft — sign in'
''
'13ft
Read the article without the cruft.
{card}
'
f'{tides_beacon()}'
)
def changelog_page(markdown_text: str) -> str:
return (
''
f'13ft — changelog'
f'← 13ft
Changelog
{html_mod.escape(markdown_text)}{tides_beacon()}'
)