fix: add security hardening and documentation for deployment

- Add document validation to prevent NoneType crash when document not configured
- Add SQL query validation (SELECT only, no multi-statement)
- Add 30-second HTTP request timeout
- Fix filter parameter JSON encoding for get_records
- Add return type annotation to get_document
- Add tests for document lookup and SQL validation
- Add comprehensive README with usage instructions
This commit is contained in:
2025-12-29 18:42:36 -05:00
parent f716e5d37e
commit ed612694fe
5 changed files with 247 additions and 7 deletions

View File

@@ -67,6 +67,14 @@ class Authenticator:
for scope in agent._token_obj.scope
]
def get_document(self, document_name: str):
"""Get document config by name."""
return self._config.documents.get(document_name)
def get_document(self, document_name: str) -> "Document":
"""Get document config by name.
Raises:
AuthError: If document is not configured.
"""
from grist_mcp.config import Document
doc = self._config.documents.get(document_name)
if doc is None:
raise AuthError(f"Document '{document_name}' not configured")
return doc

View File

@@ -1,21 +1,27 @@
"""Grist API client."""
import json
import httpx
from grist_mcp.config import Document
# Default timeout for HTTP requests (30 seconds)
DEFAULT_TIMEOUT = 30.0
class GristClient:
"""Async client for Grist API operations."""
def __init__(self, document: Document):
def __init__(self, document: Document, timeout: float = DEFAULT_TIMEOUT):
self._doc = document
self._base_url = f"{document.url.rstrip('/')}/api/docs/{document.doc_id}"
self._headers = {"Authorization": f"Bearer {document.api_key}"}
self._timeout = timeout
async def _request(self, method: str, path: str, **kwargs) -> dict:
"""Make an authenticated request to Grist API."""
async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.request(
method,
f"{self._base_url}{path}",
@@ -54,7 +60,7 @@ class GristClient:
"""Fetch records from a table."""
params = {}
if filter:
params["filter"] = filter
params["filter"] = json.dumps(filter)
if sort:
params["sort"] = sort
if limit:
@@ -68,10 +74,27 @@ class GristClient:
]
async def sql_query(self, sql: str) -> list[dict]:
"""Run a read-only SQL query."""
"""Run a read-only SQL query.
Raises:
ValueError: If query is not a SELECT statement or contains multiple statements.
"""
self._validate_sql_query(sql)
data = await self._request("GET", "/sql", params={"q": sql})
return [r["fields"] for r in data.get("records", [])]
@staticmethod
def _validate_sql_query(sql: str) -> None:
"""Validate SQL query for safety.
Only allows SELECT statements and rejects multiple statements.
"""
sql_stripped = sql.strip()
if not sql_stripped.upper().startswith("SELECT"):
raise ValueError("Only SELECT queries are allowed")
if ";" in sql_stripped[:-1]: # Allow trailing semicolon
raise ValueError("Multiple statements not allowed")
# Write operations
async def add_records(self, table: str, records: list[dict]) -> list[int]: