Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -134,3 +134,19 @@ that's what causes the version and the Docker tags to drift apart.
- `probe_iframe_embeddable` must never raise and must default to `False` (redirect) on any
ambiguity (network error, timeout, non-2xx, unparseable header) — the redirect page is the one
path that always works, so failure must never accidentally select the iframe path instead.
- `render_page` must keep escaping `link` for HTML-attribute contexts (`html.escape(..., quote=True)`
→ `link_attr`) and JSON-encoding it for the inline-JS context (`json.dumps` → `link_js`) — never
interpolate the raw string again. The admin override isn't guaranteed to match
`LIVETRACK_LINK_RE`, so this is the only thing standing between an admin-set value containing
quotes/`<>` and breaking out of an attribute or the JS string literal.
- `/admin/link` must keep rejecting (`400`) any submitted link that doesn't fullmatch
`LIVETRACK_LINK_RE` before it ever reaches `state.set_link`/`probe()` — this is what keeps the
admin override from being usable to probe/redirect to arbitrary internal URLs.
- `_check_auth` must keep using `secrets.compare_digest`, not `==`, for both the username and
password comparison.
- `__main__.py` starts the server via `waitress.create_server(...)` + `server.run()`, not the
`serve()` shortcut — the signal handler needs a `server` handle to call `.close()` on. Without
it, SIGTERM is silently swallowed by `handle_signal` and the container only stops on Docker's
SIGKILL, skipping `EmailMonitor`'s clean IMAP logout.
- The `Dockerfile`'s `HEALTHCHECK` reads `WEB_PORT` from the environment at check time (defaulting
to `8080`) rather than hard-coding a port — keep it that way if `WEB_PORT` stays configurable.
5 changes: 4 additions & 1 deletion Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,10 @@ USER waypoint
EXPOSE 8080
VOLUME ["/data"]

# Reads WEB_PORT at check time (falling back to 8080) rather than
# hard-coding it, since the CMD's JSON form doesn't get shell/env
# substitution -- only the python process it execs does.
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
CMD ["python", "-c", "import sys,urllib.request; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8080/healthz', timeout=3).status == 200 else 1)"]
CMD ["python", "-c", "import os,sys,urllib.request; port=os.environ.get('WEB_PORT','8080'); sys.exit(0 if urllib.request.urlopen(f'http://127.0.0.1:{port}/healthz', timeout=3).status == 200 else 1)"]

CMD ["python", "-m", "waypoint"]
4 changes: 3 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,9 @@ At `/admin` (Basic Auth):
- **Status dashboard** — current link, when it was last updated, IMAP connection status, last
error.
- **Manual override** — set the current link directly, or clear it (e.g. to show the "offline"
page when no activity is live).
page when no activity is live). The link must match the same
`https://livetrack.garmin.com/session/.../token/...` shape as an email-extracted one (`400` on
anything else).
- **Recent history** — the last 20 links that were published, with timestamp and source
(`email` or `admin`), including whether each was served as an iframe or a redirect.

Expand Down
39 changes: 37 additions & 2 deletions tests/test_render_page.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,19 @@
"""Tests for the pure HTML renderer in waypoint.web."""

import datetime as dt
import json

from waypoint.web import render_page
from waypoint.web import OFFLINE_REFRESH_SECONDS, render_page

LINK = "https://livetrack.garmin.com/session/xxxx/token/YYYY"
NOW = dt.datetime(2026, 7, 9, 12, 0, 0)

# Not something the email extractor or the (validating) admin route would
# ever hand render_page, but render_page itself must still neutralize it --
# defense in depth, since it's a pure function with no knowledge of the
# caller.
MALICIOUS_LINK = 'https://evil.example/"><script>alert(1)</script>'


def test_redirect_page_contains_link_meta_refresh_and_countdown():
html = render_page(LINK, use_iframe=False, countdown=15, now=NOW)
Expand All @@ -27,6 +34,34 @@ def test_offline_page_when_no_link_is_available():
html = render_page(None, use_iframe=False, countdown=10, now=NOW)

assert LINK not in html
assert "http-equiv=\"refresh\"" not in html
# Some human-readable indication that there is currently no active session.
assert "no active" in html.lower() or "offline" in html.lower()


def test_offline_page_polls_for_a_session_starting():
# The copy promises this page updates itself once a session goes live --
# it needs an actual refresh mechanism to back that up, otherwise a
# viewer who opens the link early is stuck until they manually reload.
html = render_page(None, use_iframe=False, countdown=10, now=NOW)

assert f'<meta http-equiv="refresh" content="{OFFLINE_REFRESH_SECONDS}">' in html


def test_iframe_page_escapes_link_in_html_attributes():
html = render_page(MALICIOUS_LINK, use_iframe=True, countdown=10, now=NOW)

assert "<script>alert(1)</script>" not in html
assert "&lt;script&gt;" in html


def test_redirect_page_escapes_link_in_meta_and_encodes_in_js():
html = render_page(MALICIOUS_LINK, use_iframe=False, countdown=10, now=NOW)

assert "<script>alert(1)</script>" not in html
assert "&lt;script&gt;" in html
# Plain json.dumps alone would still leak the literal "</script>" into
# the page, which closes the surrounding <script> tag at the
# HTML-parser level regardless of JS string quoting -- '<'/'>' must be
# escaped to </> in the JS-string context too.
assert "</script>" in json.dumps(MALICIOUS_LINK) # sanity: the raw case would be unsafe
assert "\\u003cscript\\u003e" in html
22 changes: 22 additions & 0 deletions tests/test_web.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,28 @@ def test_admin_set_link_probes_embeddability(config, state):
assert state.iframe_ok is False


def test_admin_set_link_rejects_non_livetrack_url(client, state):
resp = client.post(
"/admin/link",
data={"link": "javascript:alert(1)"},
headers=_basic_auth_header("admin", "s3cret"),
)

assert resp.status_code == 400
assert state.current_link is None


def test_admin_set_link_rejects_non_garmin_domain(client, state):
resp = client.post(
"/admin/link",
data={"link": "https://evil.example/session/a/token/1"},
headers=_basic_auth_header("admin", "s3cret"),
)

assert resp.status_code == 400
assert state.current_link is None


def test_admin_can_clear_link(client, state):
state.set_link("https://livetrack.garmin.com/session/c/token/3", source="email", iframe_ok=True)

Expand Down
30 changes: 27 additions & 3 deletions waypoint/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import threading

from dotenv import load_dotenv
from waitress import serve
from waitress import create_server

from .config import Config
from .email_monitor import EmailMonitor
Expand Down Expand Up @@ -47,19 +47,43 @@ def main() -> None:

monitor = EmailMonitor(config, state)

app = create_app(config, state)
# create_server (rather than the serve() shortcut) gives us a handle to
# close from the signal handler -- serve() blocks forever and installs
# no signal handling of its own, so without this, SIGTERM would just get
# silently overridden by handle_signal below and the container would
# sit until Docker's SIGKILL, skipping the monitor's clean IMAP logout.
server = create_server(app, host="0.0.0.0", port=config.web_port)
shutting_down = threading.Event()

def handle_signal(signum, frame):
log.info("Shutting down Garmin LiveTrack...")
monitor.stop()
shutting_down.set()
server.close()

signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)

monitor_thread = threading.Thread(target=monitor.run, name="email-monitor", daemon=True)
monitor_thread.start()

app = create_app(config, state)
log.info(f"Starting web server on port {config.web_port}...")
serve(app, host="0.0.0.0", port=config.web_port)
try:
server.run()
except OSError:
# server.close() (above) closes the listening socket while
# server.run()'s select() loop may be blocked on it -- waitress
# surfaces that as a "Bad file descriptor" OSError rather than a
# clean return. That's expected once we've asked for a shutdown;
# anything else is a real failure and should still propagate.
if not shutting_down.is_set():
raise
log.info("Web server stopped.")
finally:
# Give the monitor thread a chance to finish process_new_emails()
# and log out of IMAP cleanly before the process exits.
monitor_thread.join(timeout=15)


if __name__ == "__main__":
Expand Down
64 changes: 56 additions & 8 deletions waypoint/web.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,23 +7,51 @@
"""

import functools
import json
import secrets
from datetime import datetime, timezone
from html import escape
from typing import Callable, Optional

from flask import Flask, Response, abort, jsonify, redirect, request, url_for

from .config import Config
from .email_monitor import LIVETRACK_LINK_RE
from .link_probe import probe_iframe_embeddable
from .state import AppState

# How often the "no active session" placeholder polls for a session having
# started, so a viewer who opens the shared link early doesn't have to
# manually reload once one goes live.
OFFLINE_REFRESH_SECONDS = 30


def _js_string_literal(value: str) -> str:
"""JSON-encode `value` for safe embedding inside an inline <script> block.

json.dumps alone is not enough: it doesn't escape '<' or '>', so a value
containing the literal text "</script>" would close the surrounding
<script> tag at the HTML-parser level -- before any JS engine ever sees
the string -- regardless of it being inside a quoted JS string. Escaping
'<', '>' and '&' to \\u escapes (the same fix Django's `json_script`
applies) neutralizes that without changing the decoded JS value.
"""
encoded = json.dumps(value)
return encoded.replace("<", "\\u003c").replace(">", "\\u003e").replace("&", "\\u0026")


def render_page(link: Optional[str], *, use_iframe: bool, countdown: int, now: datetime) -> str:
"""Render the public HTML page.

Three variants: an "offline" placeholder when no session is currently
active, a redirect page with a JS countdown (default), or an iframe
embed with a JS-based fallback link if the iframe fails to load.

`link` is HTML-escaped for attribute contexts (`escape(..., quote=True)`)
and encoded via `_js_string_literal` for the inline-JS context before
being interpolated -- it may come from the admin override, which (unlike
the email-extracted link) isn't guaranteed to match LIVETRACK_LINK_RE, so
it can't be trusted to be free of quotes/angle brackets.
"""
timestamp = now.strftime("%m/%d/%Y %H:%M:%S")

Expand All @@ -32,6 +60,7 @@ def render_page(link: Optional[str], *, use_iframe: bool, countdown: int, now: d
<html>
<head>
<meta charset="UTF-8">
<meta http-equiv="refresh" content="{OFFLINE_REFRESH_SECONDS}">
<title>Garmin LiveTrack</title>
<style>
body {{ font-family: Arial, sans-serif; text-align: center; padding: 50px; background-color: #f5f5f5; }}
Expand All @@ -50,6 +79,9 @@ def render_page(link: Optional[str], *, use_iframe: bool, countdown: int, now: d
</body>
</html>"""

link_attr = escape(link, quote=True)
link_js = _js_string_literal(link)

if use_iframe:
return f"""<!DOCTYPE html>
<html>
Expand Down Expand Up @@ -82,11 +114,11 @@ def render_page(link: Optional[str], *, use_iframe: bool, countdown: int, now: d
</script>
</head>
<body>
<iframe id="track" src="{link}" onerror="handleIframeError()"></iframe>
<iframe id="track" src="{link_attr}" onerror="handleIframeError()"></iframe>
<div id="fallback" class="fallback">
<h2>Garmin LiveTrack</h2>
<p>The LiveTrack session cannot be embedded directly.</p>
<p><a href="{link}" target="_blank">Open LiveTrack in a new tab</a></p>
<p><a href="{link_attr}" target="_blank">Open LiveTrack in a new tab</a></p>
<p style="color: #666; font-size: 14px;">Last updated: {timestamp}</p>
</div>
</body>
Expand All @@ -97,7 +129,7 @@ def render_page(link: Optional[str], *, use_iframe: bool, countdown: int, now: d
<head>
<meta charset="UTF-8">
<title>Garmin LiveTrack - Redirecting</title>
<meta http-equiv="refresh" content="{countdown}; url={link}">
<meta http-equiv="refresh" content="{countdown}; url={link_attr}">
<style>
body {{ font-family: Arial, sans-serif; text-align: center; padding: 50px; background-color: #f5f5f5; }}
.container {{ max-width: 600px; margin: 0 auto; background: white; padding: 30px; border-radius: 10px; box-shadow: 0 2px 10px rgba(0,0,0,0.1); }}
Expand All @@ -112,7 +144,7 @@ def render_page(link: Optional[str], *, use_iframe: bool, countdown: int, now: d
document.getElementById('countdown').textContent = countdown;
countdown--;
if (countdown < 0) {{
window.location.href = '{link}';
window.location.href = {link_js};
}} else {{
setTimeout(updateCountdown, 1000);
}}
Expand All @@ -126,7 +158,7 @@ def render_page(link: Optional[str], *, use_iframe: bool, countdown: int, now: d
<h2>Redirecting to LiveTrack...</h2>
<div class="countdown">Redirecting in <span id="countdown">{countdown}</span> seconds</div>
<p>If the automatic redirect doesn't work:</p>
<a href="{link}" class="link" target="_blank">Open LiveTrack</a>
<a href="{link_attr}" class="link" target="_blank">Open LiveTrack</a>
<div class="info"><p>Last updated: {timestamp}</p></div>
</div>
</body>
Expand Down Expand Up @@ -192,7 +224,11 @@ def create_app(
app = Flask(__name__)

def _check_auth(username: str, password: str) -> bool:
return username == config.admin_user and password == config.admin_password
# constant-time compares -- a `==` short-circuits on the first
# mismatching byte, which is timing-observable over enough requests.
return secrets.compare_digest(username, config.admin_user) and secrets.compare_digest(
password, config.admin_password
)

def require_admin_auth(view):
@functools.wraps(view)
Expand Down Expand Up @@ -243,8 +279,20 @@ def admin_dashboard():
@require_admin_auth
def admin_set_link():
link = request.form.get("link", "").strip()
if link:
state.set_link(link, source="admin", iframe_ok=probe(link))
if not link:
return redirect(url_for("admin_dashboard"))
if not LIVETRACK_LINK_RE.fullmatch(link):
# Reject rather than store: render_page HTML-escapes/JSON-encodes
# whatever ends up here regardless, but restricting the admin
# override to the same shape as an extracted email link also
# closes off using it to probe/redirect to arbitrary internal
# URLs (see probe_iframe_embeddable).
return Response(
"Invalid link: expected a "
"https://livetrack.garmin.com/session/.../token/... URL",
400,
)
state.set_link(link, source="admin", iframe_ok=probe(link))
return redirect(url_for("admin_dashboard"))

@app.route("/admin/clear", methods=["POST"])
Expand Down
Loading