Merge branch 'feature/security-hardening'

Author: jensens

CHANGES.md

@@ -1,6 +1,20 @@
 # Changelog
 
-## 1.0.0b2 (unreleased)
+## 1.0.0b2
+
+### Security
+
+- **SQL identifier validation**: Added `validate_identifier()` in `columns.py`
+  to reject unsafe SQL identifiers. All `idx_key` values in `IndexRegistry`
+  and `date_attr` in `DateRecurringIndexTranslator` are now validated.
+
+- **Access control declarations**: Added `declareProtected` for management
+  methods (`refreshCatalog`, `reindexIndex`, `clearFindAndRebuild`) and
+  `declarePrivate` for `unrestrictedSearchResults` on `PlonePGCatalogTool`.
+
+- **API safety**: Renamed `execute_query()` to `_execute_query()` to mark as
+  internal API. Capped path query list size to 100 (DoS prevention).
+  Documented security contract for `IPGIndexTranslator` implementations.
 
 ### Fixed
 

tests/test_catalog_plone.py

@@ -14,6 +14,43 @@ def test_implements_ipgcatalogtool(self):
         assert IPGCatalogTool.implementedBy(PlonePGCatalogTool)
 
 
+class TestSecurityDeclarations:
+    """Verify Zope security declarations on PlonePGCatalogTool."""
+
+    def test_unrestricted_search_is_private(self):
+        # declarePrivate sets MethodName__roles__ = () (empty tuple = private)
+        roles = getattr(PlonePGCatalogTool, "unrestrictedSearchResults__roles__", None)
+        assert roles == (), f"Expected () for private, got {roles!r}"
+
+    def test_refresh_catalog_is_protected(self):
+        # declareProtected sets MethodName__roles__ = PermissionRole(...)
+        roles = getattr(PlonePGCatalogTool, "refreshCatalog__roles__", None)
+        assert roles is not None and roles != ()
+
+    def test_reindex_index_is_protected(self):
+        roles = getattr(PlonePGCatalogTool, "reindexIndex__roles__", None)
+        assert roles is not None and roles != ()
+
+    def test_clear_find_and_rebuild_is_protected(self):
+        roles = getattr(PlonePGCatalogTool, "clearFindAndRebuild__roles__", None)
+        assert roles is not None and roles != ()
+
+    def test_ac_permissions_includes_manage_entries(self):
+        # __ac_permissions__ maps permission → method names
+        perms = PlonePGCatalogTool.__ac_permissions__
+        manage_entries = None
+        for perm_name, methods in perms:
+            if perm_name == "Manage ZCatalog Entries":
+                manage_entries = methods
+                break
+        assert manage_entries is not None, (
+            "Manage ZCatalog Entries not in __ac_permissions__"
+        )
+        assert "refreshCatalog" in manage_entries
+        assert "reindexIndex" in manage_entries
+        assert "clearFindAndRebuild" in manage_entries
+
+
 class TestObjToZoid:
     def test_with_p_oid(self):
         obj = mock.Mock()

tests/test_dri.py

@@ -37,6 +37,31 @@ def __init__(self, start=None, end=None, recurrence=None):
         self.recurrence = recurrence
 
 
+# ---------------------------------------------------------------------------
+# Unit tests: constructor validation
+# ---------------------------------------------------------------------------
+
+
+class TestConstructorValidation:
+    """DateRecurringIndexTranslator rejects unsafe date_attr values."""
+
+    def test_accepts_valid_date_attr(self):
+        t = DateRecurringIndexTranslator("start", "recurrence")
+        assert t.date_attr == "start"
+
+    def test_rejects_single_quote_in_date_attr(self):
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            DateRecurringIndexTranslator("start'", "recurrence")
+
+    def test_rejects_sql_injection_in_date_attr(self):
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            DateRecurringIndexTranslator("'; DROP TABLE x; --", "recurrence")
+
+    def test_rejects_dash_in_date_attr(self):
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            DateRecurringIndexTranslator("my-start", "recurrence")
+
+
 # ---------------------------------------------------------------------------
 # Unit tests: extract()
 # ---------------------------------------------------------------------------

tests/test_query.py

@@ -562,6 +562,23 @@ def test_invalid_path_type_raises(self):
         with pytest.raises(ValueError, match="must be a string"):
             _validate_path(123)
 
+    def test_too_many_paths_raises(self):
+        """Path queries with more than _MAX_PATHS paths are rejected."""
+        from plone.pgcatalog.query import _MAX_PATHS
+
+        import pytest
+
+        paths = [f"/plone/folder{i}" for i in range(_MAX_PATHS + 1)]
+        with pytest.raises(ValueError, match="Too many paths"):
+            build_query({"path": {"query": paths}})
+
+    def test_exactly_max_paths_accepted(self):
+        from plone.pgcatalog.query import _MAX_PATHS
+
+        paths = [f"/plone/folder{i}" for i in range(_MAX_PATHS)]
+        qr = build_query({"path": {"query": paths}})
+        assert "idx" in qr["where"]
+
 
 class TestNavtreeEdgeCases:
     def test_navtree_breadcrumbs_empty(self):

tests/test_registry.py

@@ -594,6 +594,135 @@ def test_resync_adds_new_metadata(self):
         assert "getObjSize" in registry.metadata
 
 
+class TestValidateIdentifier:
+    """validate_identifier() rejects unsafe SQL identifiers."""
+
+    def test_accepts_simple_name(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        validate_identifier("portal_type")  # no exception
+
+    def test_accepts_underscore_prefix(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        validate_identifier("_private")
+
+    def test_accepts_uppercase(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        validate_identifier("Subject")
+
+    def test_accepts_alphanumeric(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        validate_identifier("field_2")
+
+    def test_rejects_single_quote(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            validate_identifier("foo'bar")
+
+    def test_rejects_semicolon(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            validate_identifier("foo;DROP TABLE")
+
+    def test_rejects_dash(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            validate_identifier("my-index")
+
+    def test_rejects_space(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            validate_identifier("my index")
+
+    def test_rejects_dot(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            validate_identifier("schema.table")
+
+    def test_rejects_leading_digit(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            validate_identifier("1field")
+
+    def test_rejects_sql_injection_payload(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            validate_identifier("'; DROP TABLE object_state; --")
+
+    def test_rejects_empty_string(self):
+        from plone.pgcatalog.columns import validate_identifier
+
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            validate_identifier("")
+
+
+class TestIndexRegistryRejectsUnsafeNames:
+    """register() and sync_from_catalog() reject unsafe SQL identifiers."""
+
+    def test_register_rejects_unsafe_idx_key(self):
+        from plone.pgcatalog.columns import IndexRegistry
+
+        registry = IndexRegistry()
+        with pytest.raises(ValueError, match="Invalid identifier"):
+            registry.register("my_index", IndexType.FIELD, "foo'bar")
+
+    def test_register_allows_none_idx_key(self):
+        from plone.pgcatalog.columns import IndexRegistry
+
+        registry = IndexRegistry()
+        registry.register("SearchableText", IndexType.TEXT, None)
+        assert "SearchableText" in registry
+
+    def test_sync_skips_unsafe_index_name(self):
+        from plone.pgcatalog.columns import IndexRegistry
+
+        idx = MockIndex("FieldIndex")
+        idx.id = "bad-name"
+        catalog = MockCatalog(indexes={"bad-name": idx})
+
+        registry = IndexRegistry()
+        registry.sync_from_catalog(catalog)
+
+        assert "bad-name" not in registry
+
+    def test_sync_skips_sql_injection_name(self):
+        from plone.pgcatalog.columns import IndexRegistry
+
+        idx = MockIndex("FieldIndex")
+        idx.id = "'; DROP TABLE x; --"
+        catalog = MockCatalog(indexes={"'; DROP TABLE x; --": idx})
+
+        registry = IndexRegistry()
+        registry.sync_from_catalog(catalog)
+
+        assert "'; DROP TABLE x; --" not in registry
+
+    def test_sync_accepts_safe_alongside_unsafe(self):
+        from plone.pgcatalog.columns import IndexRegistry
+
+        safe = MockIndex("FieldIndex")
+        safe.id = "portal_type"
+        unsafe = MockIndex("FieldIndex")
+        unsafe.id = "bad-name"
+        catalog = MockCatalog(indexes={"portal_type": safe, "bad-name": unsafe})
+
+        registry = IndexRegistry()
+        registry.sync_from_catalog(catalog)
+
+        assert "portal_type" in registry
+        assert "bad-name" not in registry
+
+
 class TestGetRegistry:
     """get_registry() returns module-level singleton."""