Search anywhere supports IPV6 extended format

Author: LucasG0

backend/infrahub/core/attribute.py

@@ -872,7 +872,7 @@ def validate_format(cls, value: Any, name: str, schema: AttributeSchema) -> None
             raise ValidationError({name: f"{value} is not a valid {schema.kind}"}) from exc
 
     def serialize_value(self) -> str:
-        """Serialize the value before storing it in the database."""
+        """Serialize the value before storing it in the database. If network is an IPv6 network, it is converted to collapsed form."""
 
         return ipaddress.ip_network(self.value).with_prefixlen
 
@@ -998,7 +998,7 @@ def validate_format(cls, value: Any, name: str, schema: AttributeSchema) -> None
             raise ValidationError({name: f"{value} is not a valid {schema.kind}"}) from exc
 
     def serialize_value(self) -> str:
-        """Serialize the value before storing it in the database."""
+        """Adds a prefix to address before storing it in the database. If address in an IPv6 address, it is converted to collapsed form."""
 
         return ipaddress.ip_interface(self.value).with_prefixlen
 

backend/infrahub/core/utils.py

@@ -191,6 +191,12 @@ def convert_ip_to_binary_str(
     return ip_bin.zfill(obj.max_prefixlen)
 
 
+def collapse_ipv6_address_or_network(address_or_network: str) -> str:
+    if "/" in address_or_network:
+        return ipaddress.IPv6Network(address_or_network).with_prefixlen
+    return str(ipaddress.IPv6Address(address_or_network))
+
+
 # --------------------------------------------------------------------------------
 # CODE IMPORTED FROM:
 #   https://github.com/graphql-python/graphene/blob/9c3e4bb7da001aac48002a3b7d83dcd072087770/graphene/utils/subclass_with_meta.py#L18

backend/infrahub/graphql/queries/search.py

@@ -7,6 +7,7 @@
 
 from infrahub.core.constants import InfrahubKind
 from infrahub.core.manager import NodeManager
+from infrahub.core.utils import collapse_ipv6_address_or_network
 
 if TYPE_CHECKING:
     from graphql import GraphQLResolveInfo
@@ -49,6 +50,12 @@ async def search_resolver(
         if matching:
             result.append(matching)
     else:
+        try:
+            # Convert any IPv6 address/network to collapsed format as it might be stored in db.
+            q = collapse_ipv6_address_or_network(q)
+        except ValueError:
+            pass
+
         result.extend(
             await NodeManager.query(
                 db=context.db,

backend/tests/unit/graphql/queries/test_search.py

@@ -93,3 +93,113 @@ async def test_search_anywhere_by_string(
 
     assert sorted(node_ids) == sorted([person_john_main.id, person_jane_main.id])
     assert sorted(node_kinds) == sorted([person_john_main.get_kind(), person_jane_main.get_kind()])
+
+
+async def test_search_ipv6_address_extended_format(
+    db: InfrahubDatabase,
+    ip_dataset_01,
+    branch: Branch,
+):
+    gql_params = prepare_graphql_params(db=db, include_subscription=False, branch=branch)
+
+    res_collapsed = await graphql(
+        schema=gql_params.schema,
+        source=SEARCH_QUERY,
+        context_value=gql_params.context,
+        root_value=None,
+        variable_values={"search": "2001:db8::"},
+    )
+
+    res_extended = await graphql(
+        schema=gql_params.schema,
+        source=SEARCH_QUERY,
+        context_value=gql_params.context,
+        root_value=None,
+        variable_values={"search": "2001:0db8:0000:0000:0000:0000:0000:0000"},
+    )
+
+    assert (
+        res_extended.data["InfrahubSearchAnywhere"]["count"]
+        == res_collapsed.data["InfrahubSearchAnywhere"]["count"]
+        == 2
+    )
+
+    assert (
+        res_extended.data["InfrahubSearchAnywhere"]["edges"][0]["node"]["id"]
+        == res_collapsed.data["InfrahubSearchAnywhere"]["edges"][0]["node"]["id"]
+    )
+
+    assert (
+        res_extended.data["InfrahubSearchAnywhere"]["edges"][1]["node"]["id"]
+        == res_collapsed.data["InfrahubSearchAnywhere"]["edges"][1]["node"]["id"]
+    )
+
+
+async def test_search_ipv6_network_extended_format(
+    db: InfrahubDatabase,
+    ip_dataset_01,
+    branch: Branch,
+):
+    gql_params = prepare_graphql_params(db=db, include_subscription=False, branch=branch)
+
+    res_collapsed = await graphql(
+        schema=gql_params.schema,
+        source=SEARCH_QUERY,
+        context_value=gql_params.context,
+        root_value=None,
+        variable_values={"search": "2001:db8::/48"},
+    )
+
+    res_extended = await graphql(
+        schema=gql_params.schema,
+        source=SEARCH_QUERY,
+        context_value=gql_params.context,
+        root_value=None,
+        variable_values={"search": "2001:0db8:0000:0000:0000:0000:0000:0000/48"},
+    )
+
+    assert (
+        res_extended.data["InfrahubSearchAnywhere"]["count"]
+        == res_collapsed.data["InfrahubSearchAnywhere"]["count"]
+        == 1
+    )
+
+    assert (
+        res_extended.data["InfrahubSearchAnywhere"]["edges"][0]["node"]["id"]
+        == res_collapsed.data["InfrahubSearchAnywhere"]["edges"][0]["node"]["id"]
+    )
+
+
+async def test_search_ipv4(
+    db: InfrahubDatabase,
+    ip_dataset_01,
+    branch: Branch,
+):
+    gql_params = prepare_graphql_params(db=db, include_subscription=False, branch=branch)
+
+    result_address = await graphql(
+        schema=gql_params.schema,
+        source=SEARCH_QUERY,
+        context_value=gql_params.context,
+        root_value=None,
+        variable_values={"search": "10.0.0.0"},
+    )
+
+    result_network = await graphql(
+        schema=gql_params.schema,
+        source=SEARCH_QUERY,
+        context_value=gql_params.context,
+        root_value=None,
+        variable_values={"search": "10.0.0.0/8"},
+    )
+
+    assert (
+        result_address.data["InfrahubSearchAnywhere"]["count"]
+        == result_network.data["InfrahubSearchAnywhere"]["count"]
+        == 1
+    )
+
+    assert (
+        result_address.data["InfrahubSearchAnywhere"]["edges"][0]["node"]["id"]
+        == result_network.data["InfrahubSearchAnywhere"]["edges"][0]["node"]["id"]
+    )

changelog/4613.fixed.md

@@ -0,0 +1 @@
+Search anywhere now supports IPv6 extended format