fix(api): bound attack paths normalized-list child IDs (#11960)

This commit is contained in:
Josema Camacho
2026-07-14 09:48:12 +02:00
committed by GitHub
parent 8debf70d5c
commit d37d5058bb
3 changed files with 55 additions and 4 deletions
@@ -0,0 +1 @@
Attack Paths scans now use bounded child node identifiers for normalized list values in Neo4j and Neptune, preventing Neo4j RANGE index key size failures
@@ -19,6 +19,7 @@ import json
import time
from collections import defaultdict
from collections.abc import Iterator
from hashlib import sha256
from typing import Any
import neo4j
@@ -392,11 +393,11 @@ def _build_child_props(
def _build_child_id(provider_id: str, child_label: str, value_key: str) -> str:
"""Deterministic `_provider_element_id` for a list-item child node.
Dedupes within (tenant, provider): multiple parents referencing the same
value share one child node via the existing MERGE-on-_provider_element_id
index in both sinks.
Hashing the value keeps the ID bounded while preserving deduplication within
each provider and child label.
"""
return f"{provider_id}::{child_label}::{value_key}"
value_digest = sha256(value_key.encode("utf-8")).hexdigest()
return f"{provider_id}::{child_label}::{value_digest}"
def _build_catalog_index(
@@ -1828,6 +1828,55 @@ def _make_session_ctx(session, call_order=None, name=None):
return ctx
class TestBuildChildId:
def test_large_value_is_hashed_and_preserved_as_child_data(self):
value = "x" * 22_796
spec = sync_module.NormalizedList(
"SomeLabel",
"values",
"SomeLabelValuesItem",
"HAS_VALUES",
)
record = {
"element_id": "elem-1",
"labels": ["SomeLabel"],
"props": {"values": [value]},
}
_, parent, children, relationships = sync_module._node_to_sync_dict(
record,
"prov-1",
sync_module._build_catalog_index([spec]),
)
child = children[0]["row"]
child_id = child["provider_element_id"]
prefix = "prov-1::SomeLabelValuesItem::"
assert parent["provider_element_id"] == "prov-1:elem-1"
assert child["props"]["value"] == value
assert len(child_id) == len(prefix) + 64
assert value not in child_id
assert relationships[0]["row"]["end_element_id"] == child_id
@pytest.mark.parametrize(
("provider_id", "child_label", "value_key"),
[
("prov-2", "ChildLabel", "value"),
("prov-1", "OtherChildLabel", "value"),
("prov-1", "ChildLabel", "other-value"),
],
)
def test_each_identity_component_changes_id(
self, provider_id, child_label, value_key
):
child_id = sync_module._build_child_id("prov-1", "ChildLabel", "value")
assert sync_module._build_child_id("prov-1", "ChildLabel", "value") == child_id
assert (
sync_module._build_child_id(provider_id, child_label, value_key) != child_id
)
class TestSyncNodes:
def test_iter_sink_batches_rejects_zero_batch_size(self):
with pytest.raises(