diff --git a/goodnews/api.py b/goodnews/api.py index e6af66b..9302904 100644 --- a/goodnews/api.py +++ b/goodnews/api.py @@ -1014,9 +1014,21 @@ def create_app() -> FastAPI: # --- CSV exports (admin-gated, for inspection / archiving) --------------- + def _csv_cell(v): + # Defuse CSV formula injection: a cell a spreadsheet might evaluate (=,+,-,@) + # gets a leading apostrophe. Numbers pass through untouched (no risk, and + # this avoids mangling any legitimate negative value). + if v is None: + return "" + if isinstance(v, (int, float)): + return v + s = str(v) + return "'" + s if s[:1] in ("=", "+", "-", "@") else s + def _csv_response(filename: str, write) -> Response: buf = io.StringIO() - write(csv.writer(buf)) + writer = csv.writer(buf) + write(lambda values: writer.writerow([_csv_cell(v) for v in values])) return Response( content=buf.getvalue(), media_type="text/csv", @@ -1030,14 +1042,14 @@ def create_app() -> FastAPI: _require_admin(conn, request) rows = queries.source_health(conn) - def write(w): - w.writerow([ + def write(row): + row([ "name", "feed_url", "homepage", "status", "visible", "served", "accepted_total", "total_articles", "acceptance_pct", "duplicate_pct", "accepted_dup_pct", "image_coverage_pct", "last_success", "last_error", "retry_after", "review_flag", "review_reason", ]) for s in rows: - w.writerow([ + row([ s["name"], s["feed_url"], s.get("homepage_url") or "", s.get("status") or "", "yes" if s.get("content_visible") else "no", s["served"], s["accepted_total"], s["total_articles"], @@ -1055,9 +1067,9 @@ def create_app() -> FastAPI: _require_admin(conn, request) st = queries.admin_stats(conn, days=days) - def write(w): + def write(row): v, a = st["visitors"], st["accounts"] - w.writerow(["metric", "value"]) + row(["metric", "value"]) for label, value in [ ("window_days", st["days"]), ("visitors_today", v["today"]), ("visitors_7d", v["d7"]), ("visitors_30d", v["d30"]), @@ -1065,13 +1077,13 @@ def create_app() -> FastAPI: ("accounts_total", a["total"]), ("accounts_new_7d", a["new_7d"]), ("accounts_active_7d", a["active_7d"]), ("feedback_7d", st.get("feedback_7d", 0)), ("feedback_unread", st.get("feedback_unread", 0)), ]: - w.writerow([label, value]) + row([label, value]) for kind, n in (st.get("shares") or {}).items(): - w.writerow([f"share_{kind}", n]) - w.writerow([]) # blank line, then the daily time series - w.writerow(["date", "visitors", "opens"]) + row([f"share_{kind}", n]) + row([]) # blank line, then the daily time series + row(["date", "visitors", "opens"]) for d in st.get("daily", []): - w.writerow([d["day"], d["visits"], d["opens"]]) + row([d["day"], d["visits"], d["opens"]]) return _csv_response("audience.csv", write) diff --git a/tests/test_admin.py b/tests/test_admin.py index 3f0273d..3af7ef4 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -184,3 +184,17 @@ def test_export_audience_csv(tmp_path, monkeypatch): assert "metric,value" in body and "window_days,7" in body assert "date,visitors,opens" in body # daily time-series section assert TestClient(app).get("/api/admin/export/audience.csv").status_code == 401 # gated + + +def test_export_sources_csv_escapes_formula_injection(tmp_path, monkeypatch): + import os, sqlite3 + app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com") + c = sqlite3.connect(os.environ["GOODNEWS_DB"]) + c.execute("UPDATE sources SET name = ?, review_flag = 1, review_reason = ? WHERE id = 1", + ('=HYPERLINK("http://bad")', '+danger')) + c.commit(); c.close() + tc = _signin(app, api, "boss@x.com") + body = tc.get("/api/admin/export/sources.csv").text + assert "'=HYPERLINK" in body # leading apostrophe defuses the formula (CSV may quote the cell) + assert "'+danger" in body + assert ",=HYPERLINK" not in body # never written as a bare, evaluable formula cell