13ft/app/otm.py

"""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.0"
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.",
}


def login_page(next_path: str, reason: Optional[str], configured: bool) -> str:
    msg = ""
    if reason:
        msg = f'<p style="color:#b3556b">{html_mod.escape(SSO_REASONS.get(reason, "Sign-in failed."))}</p>'
    if not configured:
        msg = '<p style="color:#b3556b">This 13ft isn\'t wired to OTM yet, so nobody can sign in. (Fails closed on purpose — it\'s a fetch proxy.)</p>'
    button = (
        f'<p><a href="{html_mod.escape(otm_sign_in_url(next_path))}" style="display:inline-block;padding:.7em 1.2em;border:1px solid currentColor;border-radius:8px;text-decoration:none;color:inherit;font-weight:600">Sign in with OTM</a></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">'
        '<title>13ft — sign in</title><style>:root{color-scheme:light dark}body{font-family:system-ui,sans-serif;max-width:28rem;margin:12vh auto;padding:0 1rem;line-height:1.5}</style></head>'
        '<body><h1>13ft</h1><p>Read the article without the cruft. Sign-in is by OTM account — one login for every OTM tool.</p>'
        f'{msg}{button}{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>'
    )