security: HTTP hardening (headers, CSV injection, webhook leak)

CHANGELOG.md

@@ -14,6 +14,25 @@ All notable changes to Plutus are documented here.
   short-lived `HttpOnly; SameSite=Lax` `plutus_oauth_state` cookie and
   `handle_callback` requires the callback `state` to match it (constant-time),
   in addition to the existing `_pending` nonce check. (2026-07-05 security review)
+- **Security headers on every response.** `_send` now emits `X-Frame-Options: DENY`
+  (+ CSP `frame-ancestors 'none'`) so the dashboard can't be framed/clickjacked,
+  `X-Content-Type-Options: nosniff`, `Referrer-Policy: same-origin`, and a CSP that
+  locks `default-src`/`object-src`/`base-uri`/`form-action` to self. Script/style
+  keep `'unsafe-inline'` (the dashboard is inline-only); a nonce-based `script-src`
+  is a follow-up. (2026-07-05 security review)
+- **CSV export formula-injection neutralized.** Tenant-controlled
+  `provider`/`model`/`workspace`/`task_type` cells beginning with `= + - @` (or a
+  leading tab/CR) are now quote-prefixed in `export.csv`, so a crafted value like
+  `=HYPERLINK(...)` can't execute when a teammate opens the export. (2026-07-05)
+- **Webhook error/log hardening.** `/webhook/stripe` no longer echoes the raw
+  exception text to the caller on a bad signature (returns a generic message, logs
+  detail server-side), and the success path logs only the event id/type instead of
+  the applied result (org_id/amount/balance). (2026-07-05 security review)
+
+### Changed
+- `views.simple_page`'s body parameter renamed to `body_html` with a docstring
+  making explicit that it is inserted as raw HTML and callers must pre-escape
+  untrusted data (removes a latent XSS foot-gun; no live path today). (2026-07-05)
 
 ## [1.0.0] — 2026-06-27
 

plutus_agent/server/api.py

@@ -66,14 +66,28 @@ def events_json(conn, org_id: str, limit: int = 50, before=None) -> dict:
                    "reasoning_tokens", "cost_usd", "estimated", "source"]
 
 
+def _csv_safe(value):
+    """Neutralize spreadsheet formula injection: a cell whose text begins with
+    ``= + - @`` (or a leading tab/CR that a spreadsheet strips to reach them) is
+    prefixed with a single quote so Excel/Sheets treat it as text, not a formula."""
+    if isinstance(value, str) and value[:1] in ("=", "+", "-", "@", "\t", "\r"):
+        return "'" + value
+    return value
+
+
 def export_csv(conn, org_id: str, since=None, until=None) -> str:
-    """Org-scoped usage events as CSV text (fix #66)."""
+    """Org-scoped usage events as CSV text (fix #66).
+
+    Cells are formula-injection neutralized (2026-07-05 security review):
+    ``provider``/``model``/``workspace``/``task_type`` are tenant-controlled at
+    ingest, so a crafted value like ``=HYPERLINK(...)`` would otherwise execute
+    when a teammate opens the export in a spreadsheet."""
     rows = db.export_events(conn, org_id, since=since, until=until)
     buf = io.StringIO()
     w = csv.DictWriter(buf, fieldnames=_EXPORT_COLUMNS, extrasaction="ignore")
     w.writeheader()
     for r in rows:
-        w.writerow(r)
+        w.writerow({k: _csv_safe(v) for k, v in r.items()})
     return buf.getvalue()
 
 

plutus_agent/server/app.py

@@ -160,6 +160,18 @@ def _send(self, code, body, ctype="text/html; charset=utf-8", headers=None):
         self.send_response(code)
         self.send_header("Content-Type", ctype)
         self.send_header("Content-Length", str(len(data)))
+        # Security headers on every response: block clickjacking + MIME-sniffing
+        # and lock the resource origin. The dashboard is inline-only and loads no
+        # external resources (favicon is a data: URI), so script/style keep
+        # 'unsafe-inline'; moving to a nonce-based script-src is a follow-up.
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.send_header("X-Frame-Options", "DENY")
+        self.send_header("Referrer-Policy", "same-origin")
+        self.send_header("Content-Security-Policy",
+                         "default-src 'self'; script-src 'self' 'unsafe-inline'; "
+                         "style-src 'self' 'unsafe-inline'; img-src 'self' data:; "
+                         "frame-ancestors 'none'; base-uri 'none'; object-src 'none'; "
+                         "form-action 'self'")
         for k, v in (headers or {}).items():
             self.send_header(k, v)
         self.end_headers()
@@ -834,13 +846,13 @@ def _webhook(self, conn):
         sig = self.headers.get("Stripe-Signature", "")
         try:
             event = self.ctx.stripe.construct_event(payload, sig)
-        except BillingError as e:
-            return self._json(400, {"error": str(e)})
-        except Exception as e:  # signature failure etc.
-            return self._json(400, {"error": f"invalid webhook: {e}"})
-        # Stripe events are dict-like
+        except Exception as e:  # signature/parse failure — log detail, don't echo it
+            sys.stderr.write(f"plutus: webhook rejected: {e!r}\n")
+            return self._json(400, {"error": "invalid webhook signature or payload"})
         result = handle_webhook_event(conn, event)
-        sys.stderr.write(f"plutus: stripe event {result}\n")
+        # Log event id/type only — not the applied result (org_id/amount/balance).
+        sys.stderr.write(
+            f"plutus: stripe event id={event.get('id', '')} type={event.get('type', '')}\n")
         return self._json(200, {"received": True, "result": result})
 
 

plutus_agent/server/views.py

@@ -510,14 +510,17 @@ def login_page(login_href: str) -> str:
 </div></div></body></html>"""
 
 
-def simple_page(title: str, heading: str, body: str, *, ok: bool = True) -> str:
+def simple_page(title: str, heading: str, body_html: str, *, ok: bool = True) -> str:
+    """Render a minimal status page. ``title`` and ``heading`` are auto-escaped,
+    but ``body_html`` is inserted as RAW HTML — callers MUST pre-escape any
+    untrusted/user-controlled data (e.g. ``html.escape(...)``) before passing it."""
     color = "var(--green)" if ok else "var(--coral)"
     return f"""<!doctype html><html lang="en"><head><meta charset="utf-8">
 <meta name="viewport" content="width=device-width,initial-scale=1"><title>Plutus — {_e(title)}</title>{FAVICON}
 <style>{CSS}</style></head><body><div class="wrap" style="max-width:620px">
 <div class="brand" style="margin-bottom:20px"><div class="logo">◆</div><div><h1>Plutus</h1></div></div>
 <div class="panel" style="padding:26px 22px">
   <h2 style="color:{color};font-size:18px;padding:0;text-transform:none;letter-spacing:0">{_e(heading)}</h2>
-  <div class="muted" style="margin-top:8px">{body}</div>
+  <div class="muted" style="margin-top:8px">{body_html}</div>
   <p style="margin-top:20px"><a href="/">← Back to dashboard</a></p>
 </div></div></body></html>"""

tests/test_http_hardening.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""HTTP hardening (2026-07-05 security review): security headers on every
+response, and CSV export formula-injection neutralization."""
+import os
+import sys
+import tempfile
+import threading
+import unittest
+import urllib.request
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from plutus_agent import db
+from plutus_agent.config import DEFAULT_CONFIG
+from plutus_agent.server import api, app
+
+
+class TestCsvSafe(unittest.TestCase):
+    def test_neutralizes_formula_leads(self):
+        for bad in ("=HYPERLINK(1)", "+1", "-1", "@SUM(1)", "\tx", "\rx"):
+            self.assertEqual(api._csv_safe(bad), "'" + bad,
+                             f"{bad!r} should be prefixed with a quote")
+
+    def test_leaves_safe_values_untouched(self):
+        for ok in ("openai", "gpt-5", "general", "workspace-1", "", "1.5"):
+            self.assertEqual(api._csv_safe(ok), ok)
+
+    def test_non_strings_passthrough(self):
+        for v in (0, 1, 3.14, None, True):
+            self.assertEqual(api._csv_safe(v), v)
+
+
+class TestSecurityHeaders(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        fd, cls.dbpath = tempfile.mkstemp(suffix=".db")
+        os.close(fd)
+        conn = db.connect(cls.dbpath); db.init_schema(conn); conn.close()
+        ctx = app._Ctx(dict(DEFAULT_CONFIG), cls.dbpath, demo=False)
+        cls.httpd = app._Server(("127.0.0.1", 0), app.Handler, ctx)
+        cls.port = cls.httpd.server_address[1]
+        threading.Thread(target=cls.httpd.serve_forever, daemon=True).start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.httpd.shutdown(); cls.httpd.server_close()
+        for ext in ("", "-wal", "-shm"):
+            try:
+                os.remove(cls.dbpath + ext)
+            except OSError:
+                pass
+
+    def test_security_headers_present(self):
+        r = urllib.request.urlopen(f"http://127.0.0.1:{self.port}/healthz", timeout=5)
+        self.assertEqual(r.headers.get("X-Frame-Options"), "DENY")
+        self.assertEqual(r.headers.get("X-Content-Type-Options"), "nosniff")
+        self.assertEqual(r.headers.get("Referrer-Policy"), "same-origin")
+        csp = r.headers.get("Content-Security-Policy", "")
+        self.assertIn("frame-ancestors 'none'", csp)
+        self.assertIn("default-src 'self'", csp)
+
+
+if __name__ == "__main__":
+    unittest.main()