diff --git a/.gitignore b/.gitignore index f5bc660..4a955d1 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,5 @@ __pycache__ .vercel .env*.local .DS_Store +.venv/ +__pycache__/ diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..c4c19d4 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,9 @@ +# Changelog — 13ft (OTM fork) + +## 0.4.0 — 2026-10-18 + +First OTM release of the fork (upstream wasi-master/13ft 0.3.4 underneath). + +- Sign in with OTM replaces Caddy basic auth: every fetch route needs an OTM session (grant on `app:13ft`); no secret configured = the app refuses everyone rather than becoming an open proxy. +- `/api/health` reports the version; `/changelog` renders this file; a footer with version, sign-out and an Ideas box that files a Gitea issue. +- tides beacon when the platform injects `TIDES_HOST`/`TIDES_SITE_ID`. diff --git a/CLAUDE.md b/CLAUDE.md index 0d59f13..d57d9a0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,7 +44,14 @@ Make it a proper **OTM first-party app** per 4. Keep the Dockerfile's non-root user (`thirteener`) and gunicorn; add a `HEALTHCHECK`. -Until step 2 lands, basic auth stays — it is the whole gate. +**Status 2026-10-18 (v0.4.0): steps 1–4 are built.** `app/otm.py` holds the ticket +verifier + session cookie + health/changelog/footer/Ideas; `portable.py` mounts the +`before_request` gate (everything but `otm.PUBLIC_PATHS` needs a session; no secret = +503, never open) and the routes. Tests: `cd app && ../.venv/bin/python -m unittest` +(pure `test_otm.py` + Flask test-client `test_app.py`). The platform catalogue entry +landed in otm-admin 0.148.0. Deploy = stop the hand stack (`cd /srv/otm/clients/13ft +&& docker compose down`), then `build_app 13ft` + `deploy_app 13ft`; grant `app:13ft` +to each reader on /users. The in-memory jti guard assumes gunicorn's single worker. ## Dev diff --git a/Dockerfile b/Dockerfile index 7471eb9..590aaa5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,12 +7,12 @@ ARG GID=1000 RUN addgroup -g $GID thirteener RUN adduser -D -g '' -u $UID -G thirteener thirteener -# Generic labels -LABEL maintainer="Arian Mollik Wasi " -LABEL version="0.3.4" -LABEL description="My own custom 12ft.io replacement" -LABEL url="https://github.com/wasi-master/13ft/" -LABEL documentation="https://github.com/wasi-master/13ft/blob/main/README.md" +# Generic labels (OTM fork of wasi-master/13ft 0.3.4 — see CLAUDE.md) +LABEL maintainer="Powered by OTM " +LABEL version="0.4.0" +LABEL description="13ft (OTM fork): read the article without the cruft, behind Sign in with OTM" +LABEL url="https://git.poweredbyotm.com/tonym/13ft" +LABEL documentation="https://git.poweredbyotm.com/tonym/13ft/src/branch/main/CLAUDE.md" # OCI compliant labels LABEL org.opencontainers.image.source="https://github.com/wasi-master/13ft" @@ -34,4 +34,7 @@ RUN --mount=type=cache,mode=0755,target=/home/thirteener/.cache/pip \ WORKDIR /app EXPOSE 5000 +# /api/health is one of the few unauthenticated routes (app/otm.py PUBLIC_PATHS). +HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \ + CMD wget -qO- http://127.0.0.1:5000/api/health || exit 1 CMD [ "python", "-m", "gunicorn", "portable:app" ] diff --git a/app/otm.py b/app/otm.py new file mode 100644 index 0000000..7ee1fbe --- /dev/null +++ b/app/otm.py @@ -0,0 +1,269 @@ +"""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'' + + +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.)

' + button = ( + f'

Sign in with OTM

' + if configured else "" + ) + return ( + '' + '13ft — sign in' + '

13ft

Read the article without the cruft. Sign-in is by OTM account — one login for every OTM tool.

' + f'{msg}{button}{tides_beacon()}' + ) + + +def changelog_page(markdown_text: str) -> str: + return ( + '' + f'13ft — changelog' + f'

← 13ft

Changelog

{html_mod.escape(markdown_text)}
{tides_beacon()}' + ) diff --git a/app/portable.py b/app/portable.py index 202ab5c..176169a 100644 --- a/app/portable.py +++ b/app/portable.py @@ -11,8 +11,110 @@ from urllib.parse import urljoin, urlparse, quote, unquote from flask import request, Response from bs4 import BeautifulSoup +import otm + app = flask.Flask(__name__) +# ── OTM first-party plumbing (see otm.py) ───────────────────────────────────── +# Every route except otm.PUBLIC_PATHS needs a Sign-in-with-OTM session. With no +# secret configured the app fails CLOSED — it is a fetch proxy, so an open +# deployment would be a free proxy for anyone. + +_jti_guard = otm.JtiGuard() + + +@app.before_request +def _otm_gate(): + flask.g.otm_session = None + path = request.path + if path in otm.PUBLIC_PATHS: + return None + secret = otm.session_secret() + if not secret: + return Response(otm.login_page("/", None, configured=False), status=503, mimetype="text/html") + sess = otm.decode_session(secret, request.cookies.get(otm.SESSION_COOKIE)) + if sess: + flask.g.otm_session = sess + return None + if request.method == "GET": + nxt = request.full_path if request.query_string else request.path + return flask.redirect("/login?" + urlencode_qs({"next": nxt}), code=302) + return Response(json.dumps({"error": "sign in required"}), status=401, mimetype="application/json") + + +def urlencode_qs(d): + from urllib.parse import urlencode + return urlencode(d) + + +def _esc(text): + import html as _h + return _h.escape(str(text)) + + +@app.route("/api/health") +def otm_health(): + return flask.jsonify({"status": "ok", "version": otm.APP_VERSION}) + + +@app.route("/login") +def otm_login(): + nxt = otm.safe_next(request.args.get("next"), "/") + return Response(otm.login_page(nxt, request.args.get("reason"), configured=bool(otm.session_secret())), mimetype="text/html") + + +@app.route("/logout") +def otm_logout(): + resp = flask.redirect("/login", code=302) + resp.delete_cookie(otm.SESSION_COOKIE, path="/") + return resp + + +@app.route("/api/auth/otm-sso") +def otm_sso_callback(): + def fail(reason): + return flask.redirect(f"/login?reason={reason}", code=302) + secret = (os.environ.get("OTM_SSO_SECRET") or "").strip() + sess_secret = otm.session_secret() + if not secret or not sess_secret: + return fail("not_enabled") + payload = otm.verify_sso_ticket(secret, request.args.get("token"), expected_aud=otm.OTM_AUDIENCE) + if not payload: + return fail("invalid_token") + # Claim the jti BEFORE minting any session so a replay never produces a cookie. + if not _jti_guard.claim(payload["jti"]): + return fail("link_already_used") + now = int(time.time()) + sess = {"sub": payload["sub"], "email": payload["email"], "name": payload.get("name"), "exp": now + otm.SESSION_TTL_SEC} + resp = flask.redirect(otm.safe_next(payload.get("next"), "/"), code=302) + secure = request.headers.get("X-Forwarded-Proto", request.scheme).split(",")[0].strip() == "https" + resp.set_cookie(otm.SESSION_COOKIE, otm.encode_session(sess_secret, sess), max_age=otm.SESSION_TTL_SEC, + httponly=True, secure=secure, samesite="Lax", path="/") + return resp + + +@app.route("/changelog") +def otm_changelog(): + path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "CHANGELOG.md") + try: + with open(path, "r", encoding="utf-8") as f: + text = f.read() + except OSError: + text = f"v{otm.APP_VERSION}" + return Response(otm.changelog_page(text), mimetype="text/html") + + +@app.route("/api/ideas", methods=["POST"]) +def otm_ideas(): + who = (flask.g.otm_session or {}).get("email", "") + ok, msg = otm.create_idea(request.form.get("title", ""), who, request.headers.get("Referer", "/")) + if request.accept_mimetypes.best == "application/json": + return flask.jsonify({"ok": ok, "result": msg}), (200 if ok else 400) + note = "Thanks — filed." if ok else f"Couldn't file that: {msg}" + return Response( + f'

{_esc(note)}' + f' Back

', mimetype="text/html", status=200 if ok else 400) + _DEFAULT_STRINGS = { "heading": "Enter Website Link", "label": "Link of the website you want to remove paywall for:", @@ -596,12 +698,18 @@ def render_main_page(): "elapsed_template": strings["elapsed_template"], "connection_lost_error": strings["connection_lost_error"], } - return flask.render_template_string( + rendered = flask.render_template_string( html, steps=build_steps(strings), ui_strings=ui_strings, **strings, ) + # OTM: tides beacon in , footer (version · changelog · sign out · + # Ideas) before . Post-processed so upstream's template stays + # mergeable. + email = (getattr(flask.g, "otm_session", None) or {}).get("email") + rendered = rendered.replace("", otm.tides_beacon() + "", 1) + return rendered.replace("", otm.footer_html(email) + "", 1) def process_html_document(html_content, original_url): diff --git a/app/test_app.py b/app/test_app.py new file mode 100644 index 0000000..01faac8 --- /dev/null +++ b/app/test_app.py @@ -0,0 +1,85 @@ +"""End-to-end through Flask's test client: the gate, the SSO callback, replay, +health. Run with the venv python: `cd app && ../.venv/bin/python -m unittest`.""" + +import os +import unittest + +os.environ["OTM_SSO_SECRET"] = "sso-secret" +os.environ["AUTH_SECRET"] = "auth-secret" + +import portable # noqa: E402 (env must be set before import? no — read per request; set anyway) +import otm # noqa: E402 + + +class Gate(unittest.TestCase): + def setUp(self): + self.c = portable.app.test_client() + + def test_health_is_public(self): + r = self.c.get("/api/health") + self.assertEqual(r.status_code, 200) + self.assertEqual(r.get_json()["version"], otm.APP_VERSION) + + def test_anonymous_is_bounced_to_login(self): + r = self.c.get("/") + self.assertEqual(r.status_code, 302) + self.assertTrue(r.headers["Location"].startswith("/login?next=%2F")) + r = self.c.get("/https://example.com/article") + self.assertEqual(r.status_code, 302) + self.assertIn("/login", r.headers["Location"]) + r = self.c.post("/article", data={"link": "https://example.com"}) + self.assertEqual(r.status_code, 401) + + def test_login_page_offers_otm(self): + r = self.c.get("/login?next=/x") + self.assertEqual(r.status_code, 200) + self.assertIn(b"sso/authorize?aud=app%3A13ft&next=%2Fx", r.data) + + def test_ticket_signs_in_once(self): + t = otm.mint_sso_ticket("sso-secret", {"aud": "app:13ft", "sub": "u1", "email": "tony@example.com", "role": "admin", "next": "/"}) + r = self.c.get(f"/api/auth/otm-sso?token={t}") + self.assertEqual(r.status_code, 302) + self.assertEqual(r.headers["Location"], "/") + self.assertIn(otm.SESSION_COOKIE, r.headers.get("Set-Cookie", "")) + # Signed in: the home page renders, with the OTM footer. + r = self.c.get("/") + self.assertEqual(r.status_code, 200) + self.assertIn(b"Powered by OTM", r.data) + self.assertIn(b"tony@example.com", r.data) + # Replay of the same ticket is refused. + fresh = portable.app.test_client() + r = fresh.get(f"/api/auth/otm-sso?token={t}") + self.assertEqual(r.headers["Location"], "/login?reason=link_already_used") + + def test_bad_ticket(self): + r = self.c.get("/api/auth/otm-sso?token=nope") + self.assertEqual(r.headers["Location"], "/login?reason=invalid_token") + wrong = otm.mint_sso_ticket("sso-secret", {"aud": "app:qr", "sub": "u1", "email": "e@x", "role": "admin"}) + r = self.c.get(f"/api/auth/otm-sso?token={wrong}") + self.assertEqual(r.headers["Location"], "/login?reason=invalid_token") + + def test_logout_clears(self): + t = otm.mint_sso_ticket("sso-secret", {"aud": "app:13ft", "sub": "u1", "email": "e@x.io", "role": "admin"}) + self.c.get(f"/api/auth/otm-sso?token={t}") + self.assertEqual(self.c.get("/").status_code, 200) + self.c.get("/logout") + self.assertEqual(self.c.get("/").status_code, 302) + + +class FailClosed(unittest.TestCase): + def test_no_secret_means_503_everywhere(self): + saved = {k: os.environ.pop(k, None) for k in ("OTM_SSO_SECRET", "AUTH_SECRET")} + try: + c = portable.app.test_client() + self.assertEqual(c.get("/").status_code, 503) + self.assertEqual(c.get("/https://example.com").status_code, 503) + self.assertEqual(c.get("/api/health").status_code, 200) + self.assertIn(b"isn't wired to OTM", c.get("/login").data) + finally: + for k, v in saved.items(): + if v is not None: + os.environ[k] = v + + +if __name__ == "__main__": + unittest.main() diff --git a/app/test_otm.py b/app/test_otm.py new file mode 100644 index 0000000..bfe3234 --- /dev/null +++ b/app/test_otm.py @@ -0,0 +1,95 @@ +"""python3 -m unittest app/test_otm.py (or `cd app && python -m unittest`).""" + +import unittest + +from otm import ( + JtiGuard, + decode_session, + encode_session, + mint_sso_ticket, + otm_sign_in_url, + safe_next, + verify_sso_ticket, +) + +SECRET = "test-secret" +CLAIMS = {"aud": "app:13ft", "sub": "u1", "email": "tony@example.com", "role": "admin"} + + +class Tickets(unittest.TestCase): + def test_round_trip(self): + t = mint_sso_ticket(SECRET, CLAIMS, now_ms=1_000_000) + p = verify_sso_ticket(SECRET, t, expected_aud="app:13ft", now_ms=1_030_000) + self.assertIsNotNone(p) + self.assertEqual(p["email"], "tony@example.com") + self.assertTrue(p["jti"]) + + def test_expired(self): + t = mint_sso_ticket(SECRET, CLAIMS, ttl_sec=60, now_ms=1_000_000) + self.assertIsNone(verify_sso_ticket(SECRET, t, expected_aud="app:13ft", now_ms=1_060_000)) + + def test_wrong_secret_and_wrong_audience(self): + t = mint_sso_ticket(SECRET, CLAIMS) + self.assertIsNone(verify_sso_ticket("other", t, expected_aud="app:13ft")) + self.assertIsNone(verify_sso_ticket(SECRET, t, expected_aud="app:qr")) + + def test_tampered_payload_and_junk(self): + t = mint_sso_ticket(SECRET, CLAIMS) + h, p, s = t.split(".") + self.assertIsNone(verify_sso_ticket(SECRET, f"{h}.{p}x.{s}", expected_aud="app:13ft")) + for junk in ("", "a.b", "a.b.c", None, 42, "....."): + self.assertIsNone(verify_sso_ticket(SECRET, junk, expected_aud="app:13ft")) + + def test_missing_exp_or_claims_rejected(self): + import base64, json + def b64(o): + return base64.urlsafe_b64encode(json.dumps(o).encode()).decode().rstrip("=") + from otm import _sign + h = b64({"alg": "HS256"}) + for payload in ({**CLAIMS, "jti": "x"}, {**CLAIMS, "jti": "x", "exp": 9e12, "email": ""}): + p = b64(payload) + tok = f"{h}.{p}.{_sign(SECRET, f'{h}.{p}')}" + self.assertIsNone(verify_sso_ticket(SECRET, tok, expected_aud="app:13ft")) + + def test_alg_none_rejected(self): + import base64, json + h = base64.urlsafe_b64encode(json.dumps({"alg": "none"}).encode()).decode().rstrip("=") + p = base64.urlsafe_b64encode(json.dumps({**CLAIMS, "jti": "x", "exp": 9e12}).encode()).decode().rstrip("=") + self.assertIsNone(verify_sso_ticket(SECRET, f"{h}.{p}.", expected_aud="app:13ft")) + + +class Guard(unittest.TestCase): + def test_single_use(self): + g = JtiGuard(ttl_sec=600) + self.assertTrue(g.claim("a", now=0)) + self.assertFalse(g.claim("a", now=10)) + self.assertTrue(g.claim("a", now=700)) + + +class Sessions(unittest.TestCase): + def test_round_trip_and_expiry(self): + s = {"sub": "u1", "email": "tony@example.com", "exp": 2000} + v = encode_session(SECRET, s) + self.assertEqual(decode_session(SECRET, v, now_ms=1_000_000)["email"], "tony@example.com") + self.assertIsNone(decode_session(SECRET, v, now_ms=2_000_000)) + self.assertIsNone(decode_session("other", v, now_ms=1_000_000)) + self.assertIsNone(decode_session(SECRET, v + "x", now_ms=1_000_000)) + self.assertIsNone(decode_session(SECRET, "", now_ms=1_000_000)) + + +class Paths(unittest.TestCase): + def test_safe_next(self): + self.assertEqual(safe_next("/x?y=1"), "/x?y=1") + self.assertEqual(safe_next("//evil.com"), "/") + self.assertEqual(safe_next("/\\evil.com"), "/") + self.assertEqual(safe_next("https://evil.com"), "/") + self.assertEqual(safe_next(None), "/") + + def test_sign_in_url(self): + self.assertIn("aud=app%3A13ft", otm_sign_in_url("/foo")) + self.assertIn("next=%2Ffoo", otm_sign_in_url("/foo")) + self.assertIn("next=%2F", otm_sign_in_url("//evil")) + + +if __name__ == "__main__": + unittest.main()