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
17 changes: 16 additions & 1 deletion src/opentab/tui/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -1872,6 +1872,21 @@ def _tools_dataset(self, session: Workflow) -> tuple[str, list[str], list[list]]
)
return "tools", header, rows

@staticmethod
def _csv_safe(value):
# Neutralize spreadsheet formula injection: a cell starting with =, +, -,
# @, tab, or CR is executed as a formula by Excel/LibreOffice/Sheets, and
# titles/dirs/models are attacker-influenced text. Only strings need the
# guard, and a string that is itself a plain number (a negative cost)
# passes through -- only would-be formulas get the leading apostrophe.
if not isinstance(value, str) or not value or value[0] not in "=+-@\t\r":
return value
try:
float(value)
return value
except ValueError:
return "'" + value

def export_current(self) -> None:
if self.store.demo:
self.notify("export disabled in demo mode", "error")
Expand All @@ -1886,7 +1901,7 @@ def export_current(self) -> None:
with open(path, "w", newline="") as fh:
writer = csv.writer(fh)
writer.writerow(header)
writer.writerows(rows)
writer.writerows([[self._csv_safe(cell) for cell in row] for row in rows])
except OSError as exc:
self.notify(f"export failed: {exc}", "error")
return
Expand Down
40 changes: 38 additions & 2 deletions src/opentab/web.py
Original file line number Diff line number Diff line change
Expand Up @@ -279,26 +279,62 @@ def _send(self, status: int, ctype: str, body: bytes) -> None:
self.send_response(status)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
# The page is fully self-contained -- inline JS/CSS, a data: favicon, and
# fetches only back to this server -- so everything else can be denied.
self.send_header(
"Content-Security-Policy",
"default-src 'none'; script-src 'unsafe-inline'; style-src 'unsafe-inline'; "
"img-src data:; connect-src 'self'",
)
self.send_header("X-Content-Type-Options", "nosniff")
self.end_headers()
self.wfile.write(body)

def _send_json(self, data: dict) -> None:
self._send(200, "application/json; charset=utf-8", json.dumps(data).encode("utf-8"))

def _check_host(self) -> bool:
# DNS-rebinding defense: an attacker's domain can be pointed at 127.0.0.1
# and read this server cross-origin, but the Host header still names that
# domain -- so on a loopback bind only local names pass. An explicit
# --bind beyond loopback opted out (and got the startup warning).
if self.server.server_address[0] not in ("127.0.0.1", "::1"):
return True
raw = (self.headers.get("Host") or "").strip().lower()
host = raw[1:].partition("]")[0] if raw.startswith("[") else raw.partition(":")[0]
if host in ("localhost", "127.0.0.1", "::1") or raw == "::1":
return True
self._send(403, "text/plain; charset=utf-8", b"forbidden host")
return False

def do_GET(self):
if not self._check_host():
return
path = self.path.split("?", 1)[0]
server: ReportServer = self.server # type: ignore[assignment]
if path == "/":
self._send(200, "text/html; charset=utf-8", server.page().encode("utf-8"))
elif path == "/api/reload":
server.reload()
self._send_json({"ok": True})
# State-changing, so POST-only: a GET can be fired cross-origin by any
# webpage (CSRF), and each reload re-parses the sources off disk.
self._send(405, "text/plain; charset=utf-8", b"reload is POST-only")
elif path.startswith("/api/session/"):
workflow_id = unquote(path[len("/api/session/") :])
self._send_json(session_extras(server.app, workflow_id))
else:
self._send(404, "text/plain; charset=utf-8", b"not found")

def do_POST(self):
if not self._check_host():
return
path = self.path.split("?", 1)[0]
server: ReportServer = self.server # type: ignore[assignment]
if path == "/api/reload":
server.reload()
self._send_json({"ok": True})
else:
self._send(404, "text/plain; charset=utf-8", b"not found")


class ReportServer(HTTPServer):
"""Serves the rendered report page plus the per-session JSON extras. The page
Expand Down
2 changes: 1 addition & 1 deletion src/opentab/webpage.py
Original file line number Diff line number Diff line change
Expand Up @@ -1083,7 +1083,7 @@
right.appendChild(h('button', { class: 'hbtn', title: 'Model prices (P)', onclick: openPrices }, '$/M prices'));
right.appendChild(h('button', { class: 'hbtn', title: 'Theme', onclick: openTheme }, '◑ theme'));
if (META.serve) right.appendChild(h('button', { class: 'hbtn', title: 're-read the data sources',
onclick: () => fetch('/api/reload').then(() => location.reload()) }, '↻ refresh'));
onclick: () => fetch('/api/reload', { method: 'POST' }).then(() => location.reload()) }, '↻ refresh'));
const hints = document.getElementById('hints');
hints.textContent = '';
[['j/k', 'move'], ['Tab', 'panel'], ['h/l', 'tabs'], ['Esc', 'back'], ['$', 'what-if'], ['p/t', 'projects/time'], ['T', 'trends'], ['P', 'prices'], ['R', 'range']]
Expand Down
86 changes: 86 additions & 0 deletions test_opentab.py
Original file line number Diff line number Diff line change
Expand Up @@ -3866,6 +3866,47 @@ def test_export_sources_tab_exports_the_source_breakdown():
assert rows[0][0] == "Claude Code" and rows[0][1] == 5 # cost-sorted, priciest first


def test_export_neutralizes_formula_prefixed_cells():
# Formula injection: a cell starting with =, +, -, @, tab, or CR is executed
# by Excel/LibreOffice/Sheets on import. Would-be formulas get a leading
# apostrophe; plain numbers (negative included) and non-strings pass through.
safe = ot.App._csv_safe
assert safe("=SUM(A1:A9)") == "'=SUM(A1:A9)"
assert safe("+cmd|' /C calc'!A0") == "'+cmd|' /C calc'!A0"
assert safe("@evil") == "'@evil"
assert safe("-rm -rf notes") == "'-rm -rf notes"
assert safe("\t=1+1") == "'\t=1+1"
assert safe("\r=1+1") == "'\r=1+1"
assert safe("-1.5") == "-1.5" # a negative number string is not a formula
assert safe("+42") == "+42"
assert safe(-1.5) == -1.5 and safe(0) == 0 # non-strings untouched
assert safe("session title") == "session title"
assert safe("") == ""


def test_export_current_sanitizes_the_written_csv():
import csv
import tempfile

w = workflow("w1", "2026-06-01 12:00:00", title='=HYPERLINK("http://x","y")')
app = app_with([w])
app.view = "zoom"
app.focus = "months"
app.tab = app.month_tabs.index("Sessions")
cwd = os.getcwd()
os.chdir(tempfile.mkdtemp(prefix="ot-export-"))
try:
app.export_current()
assert "exported" in app.notice
(name,) = (f for f in os.listdir(".") if f.startswith("opentab-sessions-"))
with open(name, newline="") as fh:
header, row = list(csv.reader(fh))
finally:
os.chdir(cwd)
assert row[header.index("title")] == '\'=HYPERLINK("http://x","y")'
assert row[header.index("total_cost")] == "1.0" # numeric cells stay numbers


def test_export_session_tabs_dispatch_to_their_tables():
# A store rich enough to back the Subagents / Turns / Tools tabs.
class RichStore(FakeStore):
Expand Down Expand Up @@ -8136,6 +8177,51 @@ def test_web_report_server_serves_page_extras_and_404():
thread.join(timeout=5)


def test_web_server_is_hardened_against_csrf_and_dns_rebinding():
import threading
import urllib.error
import urllib.request

app = app_with([workflow("w1", "2026-05-01 10:00:00")])
server = ot.web.ReportServer(("127.0.0.1", 0), app)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
port = server.server_address[1]
base = f"http://127.0.0.1:{port}"
try:
# Every response carries the lockdown headers (self-contained page: inline
# JS/CSS, data: favicon, fetch back to this server only).
resp = urllib.request.urlopen(base + "/")
csp = resp.headers["Content-Security-Policy"]
assert csp.startswith("default-src 'none'")
assert "connect-src 'self'" in csp and "img-src data:" in csp
assert resp.headers["X-Content-Type-Options"] == "nosniff"
# Reload mutates state, so it is POST-only: a GET (fireable cross-origin
# by any webpage) gets a 405 and does not touch the stores.
try:
urllib.request.urlopen(base + "/api/reload")
raise AssertionError("expected a 405")
except urllib.error.HTTPError as exc:
assert exc.code == 405
req = urllib.request.Request(base + "/api/reload", data=b"", method="POST")
assert json.loads(urllib.request.urlopen(req).read().decode("utf-8")) == {"ok": True}
# DNS rebinding: a foreign Host header is rejected on a loopback bind...
req = urllib.request.Request(base + "/", headers={"Host": "evil.example.com"})
try:
urllib.request.urlopen(req)
raise AssertionError("expected a 403")
except urllib.error.HTTPError as exc:
assert exc.code == 403
# ...while every local spelling passes, with or without the port.
for host in ("localhost", f"localhost:{port}", "127.0.0.1", f"[::1]:{port}"):
req = urllib.request.Request(base + "/", headers={"Host": host})
assert urllib.request.urlopen(req).status == 200
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)


def test_cli_web_flag_is_recognized_and_is_distinct_from_serve():
# --web is its own flag; web_command/main route it through the serve path.
import sys as _sys
Expand Down
Loading