Merge pull request #2 from bluedynamics/feature/security-hardening

Security hardening: SQL identifier validation, access control, API safety

Author: jensens

src/plone/pgcatalog/catalog.py

@@ -222,6 +222,11 @@ class PlonePGCatalogTool(CatalogTool):
     meta_type = "PG Catalog Tool"
     security = ClassSecurityInfo()
 
+    security.declarePrivate("unrestrictedSearchResults")
+    security.declareProtected("Manage ZCatalog Entries", "refreshCatalog")
+    security.declareProtected("Manage ZCatalog Entries", "reindexIndex")
+    security.declareProtected("Manage ZCatalog Entries", "clearFindAndRebuild")
+
     @contextmanager
     def _pg_connection(self):
         """Get a connection, preferring request-scoped reuse.

src/plone/pgcatalog/columns.py

@@ -15,10 +15,27 @@
 from enum import Enum
 
 import logging
+import re
 
 
 log = logging.getLogger(__name__)
 
+# Safe SQL identifier pattern: letters, digits, underscores only.
+# Prevents SQL injection when idx_key values are interpolated into queries.
+_SAFE_IDENTIFIER_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
+
+
+def validate_identifier(name):
+    """Validate that a name is safe for use as a SQL identifier.
+
+    Raises ValueError if the name contains characters that could
+    enable SQL injection when interpolated into query strings.
+    """
+    if not _SAFE_IDENTIFIER_RE.match(name):
+        raise ValueError(
+            f"Invalid identifier {name!r}: must match [a-zA-Z_][a-zA-Z0-9_]*"
+        )
+
 
 class IndexType(Enum):
     """ZCatalog index types we handle natively."""
@@ -104,6 +121,16 @@ def sync_from_catalog(self, catalog):
                 )
                 continue
 
+            # Validate index name as safe SQL identifier
+            if name not in SPECIAL_INDEXES:
+                try:
+                    validate_identifier(name)
+                except ValueError:
+                    log.warning(
+                        "Skipping index %r: name is not a safe SQL identifier", name
+                    )
+                    continue
+
             # Read source attributes from index object
             source_attrs = None
             if hasattr(index_obj, "getIndexSourceNames"):
@@ -136,7 +163,12 @@ def register(self, name, idx_type, idx_key, source_attrs=None):
             idx_key: JSONB key (usually same as name)
             source_attrs: list of attribute names for extraction
                           (defaults to [idx_key])
+
+        Raises:
+            ValueError: if idx_key is not a safe SQL identifier
         """
+        if idx_key is not None:
+            validate_identifier(idx_key)
         if source_attrs is None:
             source_attrs = [idx_key] if idx_key is not None else [name]
         self._indexes[name] = (idx_type, idx_key, source_attrs)

src/plone/pgcatalog/config.py

@@ -179,7 +179,7 @@ def release_request_connection(event=None):
         try:
             pool.putconn(conn)
         except Exception:
-            pass  # pool closed or conn already returned
+            log.warning("Failed to return connection to pool", exc_info=True)
     _local.pgcat_conn = None
     _local.pgcat_pool = None
 

src/plone/pgcatalog/dri.py

@@ -67,6 +67,9 @@ class DateRecurringIndexTranslator:
     """
 
     def __init__(self, date_attr, recurdef_attr, until_attr=""):
+        from plone.pgcatalog.columns import validate_identifier
+
+        validate_identifier(date_attr)
         self.date_attr = date_attr
         self.recurdef_attr = recurdef_attr
         self.until_attr = until_attr

src/plone/pgcatalog/interfaces.py

@@ -20,6 +20,16 @@ class IPGIndexTranslator(Interface):
       a fallback when the index is not in the ``IndexRegistry``.
     - ``catalog.py``: ``_extract_from_translators()`` calls ``extract()``
       for all registered translators during indexing.
+
+    **Security contract:**
+
+    Implementations MUST use psycopg parameterized queries for all
+    user-supplied values.  The ``query()`` method returns a raw SQL
+    fragment that is appended directly to the WHERE clause — never
+    interpolate user input into this fragment.  Only use ``%(name)s``
+    placeholders with corresponding entries in the returned params dict.
+    Index/column identifiers in the fragment should be hardcoded constants
+    or validated against ``columns.validate_identifier()``.
     """
 
     def extract(obj, index_name):
@@ -32,6 +42,10 @@ def extract(obj, index_name):
     def query(index_name, query_value, query_options):
         """Translate a ZCatalog query dict into a SQL fragment + params.
 
+        The returned SQL fragment is inserted directly into a WHERE clause.
+        All user-supplied values MUST use ``%(name)s`` parameter placeholders
+        — never string-format values into the SQL.
+
         Returns (sql_fragment, params_dict), e.g.:
             ("pgcatalog_to_timestamptz(idx->>'event_start') <= %(drir_date)s",
              {"drir_date": query_value})
@@ -40,5 +54,8 @@ def query(index_name, query_value, query_options):
     def sort(index_name):
         """Return SQL expression for ORDER BY, or None if not sortable.
 
+        The returned expression is inserted directly into an ORDER BY clause.
+        Only use hardcoded column references — never interpolate user input.
+
         Returns e.g.: "pgcatalog_to_timestamptz(idx->>'event_start')"
         """

src/plone/pgcatalog/query.py

@@ -37,6 +37,9 @@
 # Path validation pattern
 _PATH_RE = re.compile(r"^/[a-zA-Z0-9._/@+\-]*$")
 
+# Maximum number of paths in a single path query (DoS prevention)
+_MAX_PATHS = 100
+
 
 def _lookup_translator(name):
     """Look up an IPGIndexTranslator utility for a given index name.
@@ -110,15 +113,16 @@ def apply_security_filters(query_dict, roles, show_inactive=False):
     return result
 
 
-def execute_query(conn, query_dict, columns="zoid, path, idx, state"):
+def _execute_query(conn, query_dict, columns="zoid, path, idx, state"):
     """Execute a catalog query and return result rows.
 
-    Convenience function that builds SQL from a query dict and executes it.
+    Internal convenience function for testing.  The ``columns`` parameter
+    is interpolated into SQL, so callers must only pass trusted constants.
 
     Args:
         conn: psycopg connection (with dict_row factory)
         query_dict: ZCatalog-style query dict
-        columns: SQL column list to SELECT
+        columns: SQL column list to SELECT (must be a trusted constant)
 
     Returns:
         list of row dicts
@@ -411,6 +415,11 @@ def _handle_path(self, name, idx_key, spec):
 
         paths = [query_val] if isinstance(query_val, str) else list(query_val)
 
+        if len(paths) > _MAX_PATHS:
+            raise ValueError(
+                f"Too many paths in query ({len(paths)}), maximum is {_MAX_PATHS}"
+            )
+
         for path in paths:
             _validate_path(path)
 

src/plone/pgcatalog/schema.py

@@ -21,6 +21,12 @@
 -- The built-in ::timestamptz cast is STABLE (depends on session timezone),
 -- but our ISO 8601 dates always include timezone info, making the result
 -- deterministic.  This wrapper lets PG accept it in index expressions.
+--
+-- SECURITY NOTE: declared IMMUTABLE despite wrapping a STABLE cast.
+-- Safe because all stored dates include explicit timezone info (ISO 8601).
+-- If a value without timezone were stored, the result would depend on the
+-- session timezone setting and PG could cache incorrect index entries.
+-- Ensure all date values written to idx JSONB include timezone offsets.
 CREATE OR REPLACE FUNCTION pgcatalog_to_timestamptz(text)
 RETURNS timestamptz AS $$
     SELECT $1::timestamptz;

tests/test_fulltext.py

@@ -4,7 +4,7 @@
 """
 
 from plone.pgcatalog.indexing import catalog_object
-from plone.pgcatalog.query import execute_query
+from plone.pgcatalog.query import _execute_query as execute_query
 from tests.conftest import insert_object
 
 

tests/test_path.py

@@ -5,7 +5,7 @@
 """
 
 from plone.pgcatalog.indexing import catalog_object
-from plone.pgcatalog.query import execute_query
+from plone.pgcatalog.query import _execute_query as execute_query
 from tests.conftest import insert_object
 
 

tests/test_query_integration.py

@@ -7,7 +7,7 @@
 from datetime import datetime
 from datetime import UTC
 from plone.pgcatalog.indexing import catalog_object
-from plone.pgcatalog.query import execute_query
+from plone.pgcatalog.query import _execute_query as execute_query
 from tests.conftest import insert_object