Skip to content

Commit 56378c5

Browse files
LucasG0LucasG0
committed
Minor changes
1 parent 739e211 commit 56378c5

File tree

3 files changed

+73
-75
lines changed

3 files changed

+73
-75
lines changed

‎backend/infrahub/core/utils.py‎

Lines changed: 0 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -191,68 +191,6 @@ def convert_ip_to_binary_str(
191191
return ip_bin.zfill(obj.max_prefixlen)
192192

193193

194-
def collapse_ipv6(s: str) -> str:
195-
"""Collapse an ipv6 address, ipv6 network, or a partial ipv6 address. Raises an error if input does not resemble an IPv6 address."""
196-
197-
try:
198-
return str(ipaddress.IPv6Address(s))
199-
except ipaddress.AddressValueError:
200-
pass
201-
202-
try:
203-
return ipaddress.IPv6Network(s).with_prefixlen
204-
except ipaddress.AddressValueError:
205-
pass
206-
207-
# Input string might be an incomplete address in IPv6 format,
208-
# in which case we would like the collapsed form equivalent of this incomplete address for matching purposes.
209-
# To get it, we first try to pad the incomplete address with zeros, then we retrieve the collapsed form
210-
# of the full address, and we remove extra "::" or ":0" at the end of it.
211-
212-
error_message = "Input string does not match IPv6 format"
213-
214-
if "::" in s:
215-
raise ValueError(error_message)
216-
217-
# Add padding to complete the address if needed
218-
segments = s.split(":")
219-
220-
if len(segments) == 0:
221-
raise ValueError(error_message)
222-
223-
# If any of the non-last segments has less than 4 characters it means we deal with
224-
# a IPv6 collapsed form or an invalid address
225-
for segment in segments[:-1]:
226-
if len(segment) != 4:
227-
raise ValueError(error_message)
228-
229-
# Add 0 padding to last segment
230-
if len(segments[-1]) > 4:
231-
raise ValueError(error_message)
232-
233-
segments[-1] += "0" * (4 - len(segments[-1]))
234-
235-
# Complete the address to have 8 segments by padding with zeros
236-
while len(segments) < 8:
237-
segments.append("0000")
238-
239-
# Create a full IPv6 address from the partial input
240-
full_address = ":".join(segments)
241-
242-
# Create an IPv6Address object for validation and to build IPv6 collapsed form.
243-
ipv6_address = ipaddress.IPv6Address(full_address)
244-
245-
compressed_address = ipv6_address.compressed
246-
247-
# We padded with zeros so address might endswith "::" or ":0".
248-
if compressed_address.endswith(("::", ":0")):
249-
return compressed_address[:-2]
250-
251-
# Otherwise, it means 8th segment of ipv6 address was not full and not composed of 0 only
252-
# e.g. 2001:0db8:0000:0000:0000:0000:03
253-
return compressed_address
254-
255-
256194
# --------------------------------------------------------------------------------
257195
# CODE IMPORTED FROM:
258196
# https://github.com/graphql-python/graphene/blob/9c3e4bb7da001aac48002a3b7d83dcd072087770/graphene/utils/subclass_with_meta.py#L18

‎backend/infrahub/graphql/queries/search.py‎

Lines changed: 69 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,13 @@
11
from __future__ import annotations
22

3+
import ipaddress
34
from typing import TYPE_CHECKING, Any, Optional
45

56
from graphene import Boolean, Field, Int, List, ObjectType, String
67
from infrahub_sdk.utils import extract_fields_first_node, is_valid_uuid
78

89
from infrahub.core.constants import InfrahubKind
910
from infrahub.core.manager import NodeManager
10-
from infrahub.core.utils import collapse_ipv6
1111

1212
if TYPE_CHECKING:
1313
from graphql import GraphQLResolveInfo
@@ -30,6 +30,72 @@ class NodeEdges(ObjectType):
3030
edges = Field(List(of_type=NodeEdge, required=True), required=False)
3131

3232

33+
def _collapse_ipv6(s: str) -> str:
34+
"""Collapse an ipv6 address, ipv6 network, or a partial ipv6 address in extended format, into its collapsed form.
35+
Raises an error if input does not resemble an IPv6 address in extended format. It means this function also raises
36+
an error if input string is the start of an IPv6 address in collapsed format.
37+
"""
38+
39+
try:
40+
return str(ipaddress.IPv6Address(s))
41+
except ipaddress.AddressValueError:
42+
pass
43+
44+
try:
45+
return ipaddress.IPv6Network(s).with_prefixlen
46+
except ipaddress.AddressValueError:
47+
pass
48+
49+
# Input string might be an incomplete address in IPv6 format,
50+
# in which case we would like the collapsed form equivalent of this incomplete address for matching purposes.
51+
# To get it, we first try to pad the incomplete address with zeros, then we retrieve the collapsed form
52+
# of the full address, and we remove extra "::" or ":0" at the end of it.
53+
54+
error_message = "Input string does not match IPv6 extended format"
55+
56+
# Input string cannot be an IPv6 in extended format if it contains ":"
57+
if "::" in s:
58+
raise ValueError(error_message)
59+
60+
# Add padding to complete the address if needed
61+
segments = s.split(":")
62+
63+
if len(segments) == 0:
64+
raise ValueError(error_message)
65+
66+
# If any of the non-last segments has less than 4 characters it means we deal with
67+
# a IPv6 collapsed form or an invalid address
68+
for segment in segments[:-1]:
69+
if len(segment) != 4:
70+
raise ValueError(error_message)
71+
72+
# Add 0 padding to last segment
73+
if len(segments[-1]) > 4:
74+
raise ValueError(error_message)
75+
76+
segments[-1] += "0" * (4 - len(segments[-1]))
77+
78+
# Complete the address to have 8 segments by padding with zeros
79+
while len(segments) < 8:
80+
segments.append("0000")
81+
82+
# Create a full IPv6 address from the partial input
83+
full_address = ":".join(segments)
84+
85+
# Create an IPv6Address object for validation and to build IPv6 collapsed form.
86+
ipv6_address = ipaddress.IPv6Address(full_address)
87+
88+
compressed_address = ipv6_address.compressed
89+
90+
# We padded with zeros so address might endswith "::" or ":0".
91+
if compressed_address.endswith(("::", ":0")):
92+
return compressed_address[:-2]
93+
94+
# Otherwise, it means 8th segment of ipv6 address was not full and not composed of 0 only
95+
# e.g. 2001:0db8:0000:0000:0000:0000:03
96+
return compressed_address
97+
98+
3399
async def search_resolver(
34100
root: dict, # pylint: disable=unused-argument
35101
info: GraphQLResolveInfo,
@@ -52,8 +118,8 @@ async def search_resolver(
52118
else:
53119
try:
54120
# Convert any IPv6 address, network or partial address to collapsed format as it might be stored in db.
55-
q = collapse_ipv6(q)
56-
except ValueError:
121+
q = _collapse_ipv6(q)
122+
except (ValueError, ipaddress.AddressValueError):
57123
pass
58124

59125
result.extend(

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

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,9 @@
33

44
from infrahub.core.branch import Branch
55
from infrahub.core.node import Node
6-
from infrahub.core.utils import collapse_ipv6
76
from infrahub.database import InfrahubDatabase
87
from infrahub.graphql.initialization import prepare_graphql_params
8+
from infrahub.graphql.queries.search import _collapse_ipv6
99

1010
SEARCH_QUERY = """
1111
query ($search: String!) {
@@ -272,19 +272,13 @@ async def test_search_ipv4(
272272
],
273273
)
274274
def test_collapse_ipv6_address_or_network(query, expected):
275-
assert collapse_ipv6(query) == expected
275+
assert _collapse_ipv6(query) == expected
276276

277277

278278
@pytest.mark.parametrize(
279279
"query",
280-
[
281-
"invalid",
282-
"invalid:case",
283-
"2001:invalid",
284-
"2001:0db81:0000",
285-
"10.0.0.0",
286-
],
280+
["invalid", "invalid:case", "2001:invalid", "2001:0db81:0000", "10.0.0.0", "2001:db8:1"],
287281
)
288282
def test_collapse_ipv6_address_or_network_invalid_cases(query):
289283
with pytest.raises(ValueError):
290-
collapse_ipv6(query)
284+
_collapse_ipv6(query)

0 commit comments

Comments
 (0)