diff --git a/graphiti_core/edges.py b/graphiti_core/edges.py
index 28c170355e..2a983c2fff 100644
--- a/graphiti_core/edges.py
+++ b/graphiti_core/edges.py
@@ -28,7 +28,7 @@
 from graphiti_core.driver.driver import GraphDriver, GraphProvider
 from graphiti_core.embedder import EmbedderClient
 from graphiti_core.errors import EdgeNotFoundError, GroupsEdgesNotFoundError
-from graphiti_core.helpers import parse_db_date
+from graphiti_core.helpers import parse_db_date, validate_group_id
 from graphiti_core.models.edges.edge_db_queries import (
     COMMUNITY_EDGE_RETURN,
     EPISODIC_EDGE_RETURN,
@@ -53,6 +53,12 @@ class Edge(BaseModel, ABC):
     target_node_uuid: str
     created_at: datetime
 
+    def _validate_for_write(self) -> None:
+        # Validate group_id at the persistence boundary only. Hydration from the DB must
+        # stay tolerant of any stored value, so validation lives here (called by every
+        # concrete save()) rather than on the model where it would also fire on reads.
+        validate_group_id(self.group_id)
+
     @abstractmethod
     async def save(self, driver: GraphDriver): ...
 
@@ -142,6 +148,8 @@ async def get_by_uuid(cls, driver: GraphDriver, uuid: str): ...
 
 class EpisodicEdge(Edge):
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.episodic_edge_save(self, driver)
@@ -333,6 +341,8 @@ async def load_fact_embedding(self, driver: GraphDriver):
         self.fact_embedding = records[0]['fact_embedding']
 
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.edge_save(self, driver)
@@ -574,6 +584,8 @@ async def get_by_node_uuid(cls, driver: GraphDriver, node_uuid: str):
 
 class CommunityEdge(Edge):
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.community_edge_save(self, driver)
@@ -688,6 +700,8 @@ async def get_by_group_ids(
 
 class HasEpisodeEdge(Edge):
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.has_episode_edge_save(self, driver)
@@ -821,6 +835,8 @@ async def get_by_group_ids(
 
 class NextEpisodeEdge(Edge):
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.next_episode_edge_save(self, driver)
diff --git a/graphiti_core/errors.py b/graphiti_core/errors.py
index 40ae36eada..a98e952383 100644
--- a/graphiti_core/errors.py
+++ b/graphiti_core/errors.py
@@ -75,7 +75,7 @@ def __init__(self, entity_type: str, entity_type_attribute: str):
         super().__init__(self.message)
 
 
-class GroupIdValidationError(GraphitiError):
+class GroupIdValidationError(GraphitiError, ValueError):
     """Raised when a group_id contains invalid characters."""
 
     def __init__(self, group_id: str):
@@ -93,3 +93,14 @@ def __init__(self, node_labels: list[str]):
             f'alphanumeric characters or underscores: {label_list}'
         )
         super().__init__(self.message)
+
+
+class PropertyNameValidationError(GraphitiError, ValueError):
+    """Raised when a property filter name is not a safe Cypher identifier."""
+
+    def __init__(self, property_name: str):
+        self.message = (
+            f'property filter name "{property_name}" must start with a letter or underscore '
+            'and contain only alphanumeric characters or underscores'
+        )
+        super().__init__(self.message)
diff --git a/graphiti_core/helpers.py b/graphiti_core/helpers.py
index aa6c503a9d..734e1ff0bd 100644
--- a/graphiti_core/helpers.py
+++ b/graphiti_core/helpers.py
@@ -28,7 +28,11 @@
 from pydantic import BaseModel
 
 from graphiti_core.driver.driver import GraphProvider
-from graphiti_core.errors import GroupIdValidationError, NodeLabelValidationError
+from graphiti_core.errors import (
+    GroupIdValidationError,
+    NodeLabelValidationError,
+    PropertyNameValidationError,
+)
 
 load_dotenv()
 
@@ -186,6 +190,20 @@ def validate_node_labels(node_labels: list[str] | None) -> bool:
     return True
 
 
+def validate_property_name(property_name: str) -> bool:
+    """Validate that a property name is safe to interpolate into Cypher expressions.
+
+    Property keys cannot be parameterized in Cypher, so filter property names are
+    interpolated directly into the query and must be constrained to a safe identifier
+    pattern to prevent query injection.
+    """
+
+    if not SAFE_CYPHER_IDENTIFIER_PATTERN.match(property_name or ''):
+        raise PropertyNameValidationError(property_name)
+
+    return True
+
+
 def validate_excluded_entity_types(
     excluded_entity_types: list[str] | None, entity_types: dict[str, type[BaseModel]] | None = None
 ) -> bool:
diff --git a/graphiti_core/nodes.py b/graphiti_core/nodes.py
index 9644ec2b55..f5a3acd71c 100644
--- a/graphiti_core/nodes.py
+++ b/graphiti_core/nodes.py
@@ -32,7 +32,7 @@
 )
 from graphiti_core.embedder import EmbedderClient
 from graphiti_core.errors import NodeNotFoundError
-from graphiti_core.helpers import parse_db_date, validate_node_labels
+from graphiti_core.helpers import parse_db_date, validate_group_id, validate_node_labels
 from graphiti_core.models.nodes.node_db_queries import (
     COMMUNITY_NODE_RETURN,
     COMMUNITY_NODE_RETURN_NEPTUNE,
@@ -105,6 +105,12 @@ def validate_labels(cls, value: list[str]) -> list[str]:
         validate_node_labels(value)
         return value
 
+    def _validate_for_write(self) -> None:
+        # Validate group_id at the persistence boundary only. Hydration from the DB must
+        # stay tolerant of any stored value, so validation lives here (called by every
+        # concrete save()) rather than on the model where it would also fire on reads.
+        validate_group_id(self.group_id)
+
     @abstractmethod
     async def save(self, driver: GraphDriver): ...
 
@@ -332,6 +338,8 @@ class EpisodicNode(Node):
     )
 
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.episodic_node_save(self, driver)
@@ -544,6 +552,8 @@ async def load_name_embedding(self, driver: GraphDriver):
         self.name_embedding = records[0]['name_embedding']
 
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.node_save(self, driver)
@@ -689,6 +699,8 @@ class CommunityNode(Node):
     summary: str = Field(description='region summary of member nodes', default_factory=str)
 
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.community_node_save(self, driver)
@@ -876,6 +888,8 @@ class SagaNode(Node):
     last_summarized_episode_valid_at: datetime | None = None
 
     async def save(self, driver: GraphDriver):
+        self._validate_for_write()
+
         if driver.graph_operations_interface:
             try:
                 return await driver.graph_operations_interface.saga_node_save(self, driver)
diff --git a/graphiti_core/search/search_filters.py b/graphiti_core/search/search_filters.py
index 7d45f4b66c..83b4ca7e55 100644
--- a/graphiti_core/search/search_filters.py
+++ b/graphiti_core/search/search_filters.py
@@ -21,7 +21,7 @@
 from pydantic import BaseModel, Field, field_validator
 
 from graphiti_core.driver.driver import GraphProvider
-from graphiti_core.helpers import validate_node_labels
+from graphiti_core.helpers import validate_node_labels, validate_property_name
 
 
 class ComparisonOperator(Enum):
@@ -64,6 +64,9 @@ class SearchFilters(BaseModel):
     created_at: list[list[DateFilter]] | None = Field(default=None)
     expired_at: list[list[DateFilter]] | None = Field(default=None)
     edge_uuids: list[str] | None = Field(default=None)
+    # A single property_filters list is intentionally shared: it is applied to the node
+    # alias `n` by node_search_filter_query_constructor and to the edge alias `e` by
+    # edge_search_filter_query_constructor.
     property_filters: list[PropertyFilter] | None = Field(default=None)
 
     @field_validator('node_labels')
@@ -72,6 +75,16 @@ def validate_node_label_filters(cls, value: list[str] | None) -> list[str] | Non
         validate_node_labels(value)
         return value
 
+    @field_validator('property_filters')
+    @classmethod
+    def validate_property_filter_names(
+        cls, value: list[PropertyFilter] | None
+    ) -> list[PropertyFilter] | None:
+        if value is not None:
+            for property_filter in value:
+                validate_property_name(property_filter.property_name)
+        return value
+
 
 def cypher_to_opensearch_operator(op: ComparisonOperator) -> str:
     mapping = {
@@ -83,6 +96,35 @@ def cypher_to_opensearch_operator(op: ComparisonOperator) -> str:
     return mapping.get(op, op.value)
 
 
+def property_filter_query_constructor(
+    entity_alias: str,
+    property_filters: list[PropertyFilter],
+    param_prefix: str,
+) -> tuple[list[str], dict[str, Any]]:
+    """Build Cypher fragments for a list of property filters against a single entity alias.
+
+    Property keys cannot be parameterized in Cypher, so each name is validated
+    (defense-in-depth against model_construct()/other validation bypasses) before it is
+    interpolated. Property values are always passed as query parameters.
+    """
+    filter_queries: list[str] = []
+    filter_params: dict[str, Any] = {}
+
+    for i, property_filter in enumerate(property_filters):
+        validate_property_name(property_filter.property_name)
+        property_reference = f'{entity_alias}.{property_filter.property_name}'
+        operator = property_filter.comparison_operator
+
+        if operator in (ComparisonOperator.is_null, ComparisonOperator.is_not_null):
+            filter_queries.append(f'{property_reference} {operator.value}')
+        else:
+            param_name = f'{param_prefix}_{i}'
+            filter_queries.append(f'{property_reference} {operator.value} ${param_name}')
+            filter_params[param_name] = property_filter.property_value
+
+    return filter_queries, filter_params
+
+
 def node_search_filter_query_constructor(
     filters: SearchFilters,
     provider: GraphProvider,
@@ -101,6 +143,13 @@ def node_search_filter_query_constructor(
             node_label_filter = 'n:' + node_labels
         filter_queries.append(node_label_filter)
 
+    if filters.property_filters is not None:
+        property_queries, property_params = property_filter_query_constructor(
+            'n', filters.property_filters, 'node_prop'
+        )
+        filter_queries.extend(property_queries)
+        filter_params.update(property_params)
+
     return filter_queries, filter_params
 
 
@@ -133,6 +182,13 @@ def edge_search_filter_query_constructor(
         filter_queries.append('e.uuid in $edge_uuids')
         filter_params['edge_uuids'] = filters.edge_uuids
 
+    if filters.property_filters is not None:
+        property_queries, property_params = property_filter_query_constructor(
+            'e', filters.property_filters, 'edge_prop'
+        )
+        filter_queries.extend(property_queries)
+        filter_params.update(property_params)
+
     if filters.node_labels is not None:
         # Defense-in-depth for model_construct()/other validation bypasses.
         validate_node_labels(filters.node_labels)
diff --git a/graphiti_core/utils/bulk_utils.py b/graphiti_core/utils/bulk_utils.py
index 92f25582e2..08125d53cb 100644
--- a/graphiti_core/utils/bulk_utils.py
+++ b/graphiti_core/utils/bulk_utils.py
@@ -31,7 +31,7 @@
 from graphiti_core.edges import Edge, EntityEdge, EpisodicEdge, create_entity_edge_embeddings
 from graphiti_core.embedder import EmbedderClient
 from graphiti_core.graphiti_types import GraphitiClients
-from graphiti_core.helpers import normalize_l2, semaphore_gather
+from graphiti_core.helpers import normalize_l2, semaphore_gather, validate_group_id
 from graphiti_core.models.edges.edge_db_queries import (
     get_entity_edge_save_bulk_query,
     get_episodic_edge_save_bulk_query,
@@ -157,6 +157,12 @@ async def add_nodes_and_edges_bulk_tx(
     embedder: EmbedderClient,
     driver: GraphDriver,
 ):
+    # Validate group_ids on write only. Hydration from the DB stays tolerant of any
+    # stored value, so the bulk persistence boundary re-validates here (the per-object
+    # save() methods are bypassed on this path).
+    for element in (*episodic_nodes, *episodic_edges, *entity_nodes, *entity_edges):
+        validate_group_id(element.group_id)
+
     episodes = [dict(episode) for episode in episodic_nodes]
     for episode in episodes:
         episode['source'] = str(episode['source'].value)
diff --git a/server/graph_service/dto/ingest.py b/server/graph_service/dto/ingest.py
index 9b0159c859..719b5193de 100644
--- a/server/graph_service/dto/ingest.py
+++ b/server/graph_service/dto/ingest.py
@@ -1,15 +1,39 @@
-from pydantic import BaseModel, Field
+from graphiti_core.errors import GroupIdValidationError  # type: ignore
+from graphiti_core.helpers import validate_group_id  # type: ignore
+from pydantic import BaseModel, Field, field_validator
 
 from graph_service.dto.common import Message
 
 
+def _validate_request_group_id(value: str) -> str:
+    # Re-raise as ValueError so Pydantic wraps it into a ValidationError (HTTP 422)
+    # regardless of whether the installed graphiti-core makes GroupIdValidationError a
+    # ValueError subclass. Rejecting bad group_ids on write keeps records reachable and
+    # deletable via the read/delete API paths, which validate the same pattern.
+    try:
+        validate_group_id(value)
+    except GroupIdValidationError as error:
+        raise ValueError(str(error)) from error
+    return value
+
+
 class AddMessagesRequest(BaseModel):
     group_id: str = Field(..., description='The group id of the messages to add')
     messages: list[Message] = Field(..., description='The messages to add')
 
+    @field_validator('group_id')
+    @classmethod
+    def validate_group_id_field(cls, value: str) -> str:
+        return _validate_request_group_id(value)
+
 
 class AddEntityNodeRequest(BaseModel):
     uuid: str = Field(..., description='The uuid of the node to add')
     group_id: str = Field(..., description='The group id of the node to add')
     name: str = Field(..., description='The name of the node to add')
     summary: str = Field(default='', description='The summary of the node to add')
+
+    @field_validator('group_id')
+    @classmethod
+    def validate_group_id_field(cls, value: str) -> str:
+        return _validate_request_group_id(value)
diff --git a/server/tests/test_dto_validation.py b/server/tests/test_dto_validation.py
new file mode 100644
index 0000000000..12f978655c
--- /dev/null
+++ b/server/tests/test_dto_validation.py
@@ -0,0 +1,29 @@
+"""Unit tests for request DTO validation (no database or network required)."""
+
+import pytest
+from pydantic import ValidationError
+
+from graph_service.dto import AddEntityNodeRequest, AddMessagesRequest
+
+
+def test_add_entity_node_request_rejects_unsafe_group_id():
+    # The write path must reject group_ids the read/delete path would refuse to match,
+    # so a bad group_id surfaces as a 422 (ValidationError) rather than creating an
+    # unreachable record.
+    with pytest.raises(ValidationError):
+        AddEntityNodeRequest(uuid='u', group_id='bad"group', name='n')
+
+
+def test_add_entity_node_request_accepts_valid_group_id():
+    request = AddEntityNodeRequest(uuid='u', group_id='valid-group_1', name='n')
+    assert request.group_id == 'valid-group_1'
+
+
+def test_add_messages_request_rejects_unsafe_group_id():
+    with pytest.raises(ValidationError):
+        AddMessagesRequest(group_id='bad"group', messages=[])
+
+
+def test_add_messages_request_accepts_valid_group_id():
+    request = AddMessagesRequest(group_id='valid-group_1', messages=[])
+    assert request.group_id == 'valid-group_1'
diff --git a/tests/utils/search/test_search_security.py b/tests/utils/search/test_search_security.py
index 17fc419db6..205a8fb63b 100644
--- a/tests/utils/search/test_search_security.py
+++ b/tests/utils/search/test_search_security.py
@@ -7,11 +7,17 @@
 from graphiti_core.driver.driver import GraphProvider
 from graphiti_core.driver.falkordb.operations.search_ops import _build_falkor_fulltext_query
 from graphiti_core.driver.neo4j.operations.search_ops import _build_neo4j_fulltext_query
-from graphiti_core.errors import GroupIdValidationError, NodeLabelValidationError
+from graphiti_core.errors import (
+    GroupIdValidationError,
+    NodeLabelValidationError,
+    PropertyNameValidationError,
+)
 from graphiti_core.helpers import get_default_group_id, validate_group_id
 from graphiti_core.search.search import search
 from graphiti_core.search.search_config import SearchConfig
 from graphiti_core.search.search_filters import (
+    ComparisonOperator,
+    PropertyFilter,
     SearchFilters,
     edge_search_filter_query_constructor,
     node_search_filter_query_constructor,
@@ -53,6 +59,116 @@ def test_edge_search_filter_constructor_rejects_unsafe_labels_bypassing_pydantic
         edge_search_filter_query_constructor(filters, GraphProvider.NEO4J)
 
 
+def test_node_search_filter_constructor_applies_property_filters():
+    filters = SearchFilters(
+        property_filters=[
+            PropertyFilter(
+                property_name='group_id',
+                property_value='confidential-tenant',
+                comparison_operator=ComparisonOperator.equals,
+            )
+        ]
+    )
+
+    filter_queries, filter_params = node_search_filter_query_constructor(
+        filters, GraphProvider.NEO4J
+    )
+
+    assert filter_queries == ['n.group_id = $node_prop_0']
+    assert filter_params == {'node_prop_0': 'confidential-tenant'}
+
+
+def test_edge_search_filter_constructor_applies_property_filters():
+    filters = SearchFilters(
+        property_filters=[
+            PropertyFilter(
+                property_name='weight',
+                property_value=5,
+                comparison_operator=ComparisonOperator.greater_than,
+            )
+        ]
+    )
+
+    filter_queries, filter_params = edge_search_filter_query_constructor(
+        filters, GraphProvider.NEO4J
+    )
+
+    assert 'e.weight > $edge_prop_0' in filter_queries
+    assert filter_params['edge_prop_0'] == 5
+
+
+def test_property_filter_null_operators_emit_no_params():
+    filters = SearchFilters(
+        property_filters=[
+            PropertyFilter(
+                property_name='deleted_at',
+                comparison_operator=ComparisonOperator.is_null,
+            ),
+            PropertyFilter(
+                property_name='created_at',
+                comparison_operator=ComparisonOperator.is_not_null,
+            ),
+        ]
+    )
+
+    filter_queries, filter_params = node_search_filter_query_constructor(
+        filters, GraphProvider.NEO4J
+    )
+
+    assert filter_queries == ['n.deleted_at IS NULL', 'n.created_at IS NOT NULL']
+    assert filter_params == {}
+
+
+def test_property_filters_are_shared_across_node_and_edge_constructors():
+    # A single property_filters list is intentionally applied to the node alias `n` in
+    # node searches and the edge alias `e` in edge searches.
+    filters = SearchFilters(
+        property_filters=[
+            PropertyFilter(
+                property_name='status',
+                property_value='active',
+                comparison_operator=ComparisonOperator.equals,
+            )
+        ]
+    )
+
+    node_queries, node_params = node_search_filter_query_constructor(filters, GraphProvider.NEO4J)
+    edge_queries, edge_params = edge_search_filter_query_constructor(filters, GraphProvider.NEO4J)
+
+    assert node_queries == ['n.status = $node_prop_0']
+    assert edge_queries == ['e.status = $edge_prop_0']
+    assert node_params == {'node_prop_0': 'active'}
+    assert edge_params == {'edge_prop_0': 'active'}
+
+
+def test_search_filters_reject_unsafe_property_names():
+    with pytest.raises(ValidationError):
+        SearchFilters(
+            property_filters=[
+                PropertyFilter(
+                    property_name='group_id`) DETACH DELETE n //',
+                    property_value='x',
+                    comparison_operator=ComparisonOperator.equals,
+                )
+            ]
+        )
+
+
+def test_property_filter_constructor_rejects_unsafe_names_bypassing_pydantic():
+    filters = SearchFilters.model_construct(
+        property_filters=[
+            PropertyFilter.model_construct(
+                property_name='group_id`) DETACH DELETE n //',
+                property_value='x',
+                comparison_operator=ComparisonOperator.equals,
+            )
+        ]
+    )
+
+    with pytest.raises(PropertyNameValidationError):
+        node_search_filter_query_constructor(filters, GraphProvider.NEO4J)
+
+
 def test_fulltext_query_rejects_invalid_group_ids():
     driver = SimpleNamespace(provider=GraphProvider.NEO4J, fulltext_syntax='')
 
@@ -109,3 +225,67 @@ def test_falkordb_fulltext_query_escapes_default_group_id():
     built = _build_falkor_fulltext_query('hello', [default])
     assert '@group_id:"\\_"' in built
     assert '@group_id:"_"' not in built
+
+
+def test_node_construction_is_tolerant_of_any_group_id():
+    # Hydration from the DB builds models directly from stored values, so construction
+    # must NOT validate group_id — otherwise a cross-partition (group_ids=None) read of a
+    # legacy record with a non-conforming group_id would hard-fail. Validation happens on
+    # write (save()) instead.
+    from graphiti_core.nodes import EntityNode
+
+    EntityNode(name='test', group_id='bad"group')  # must not raise
+    EntityNode(name='test', group_id='')
+    EntityNode(name='test', group_id='valid-group_1')
+
+
+@pytest.mark.asyncio
+async def test_entity_node_rejects_unsafe_group_id_on_save():
+    # Regression: the write path must reject group_ids that the read/delete path would
+    # later refuse to match, otherwise records become unreachable. validate_group_id is
+    # the first statement in save(), so a MagicMock driver is never actually used.
+    from graphiti_core.nodes import EntityNode
+
+    node = EntityNode(name='test', group_id='bad"group')
+    with pytest.raises(GroupIdValidationError, match='must contain only alphanumeric'):
+        await node.save(MagicMock())
+
+
+@pytest.mark.asyncio
+async def test_entity_edge_rejects_unsafe_group_id_on_save():
+    from datetime import datetime
+
+    from graphiti_core.edges import EntityEdge
+
+    edge = EntityEdge(
+        name='RELATES_TO',
+        group_id='bad"group',
+        source_node_uuid='source',
+        target_node_uuid='target',
+        fact='fact',
+        created_at=datetime(2024, 1, 1),
+    )
+    with pytest.raises(GroupIdValidationError, match='must contain only alphanumeric'):
+        await edge.save(MagicMock())
+
+
+@pytest.mark.asyncio
+async def test_non_entity_node_also_rejects_unsafe_group_id_on_save():
+    # Every concrete save() guards group_id at the write boundary via the shared
+    # Node._validate_for_write(), not just EntityNode — a direct EpisodicNode(...).save()
+    # with a non-conforming group_id must not create an unreachable record.
+    from datetime import datetime
+
+    from graphiti_core.nodes import EpisodeType, EpisodicNode
+
+    node = EpisodicNode(
+        name='ep',
+        group_id='bad"group',
+        source=EpisodeType.text,
+        source_description='',
+        content='',
+        valid_at=datetime(2024, 1, 1),
+        entity_edges=[],
+    )
+    with pytest.raises(GroupIdValidationError, match='must contain only alphanumeric'):
+        await node.save(MagicMock())