Block connector limited-broadcast targets
This commit is contained in:
@@ -17,6 +17,7 @@ DEFAULT_FILE_TRANSFER_BYTES: Final = 512 * 1024 * 1024
|
|||||||
_READ_CHUNK_BYTES: Final = 64 * 1024
|
_READ_CHUNK_BYTES: Final = 64 * 1024
|
||||||
_TRUE_VALUES: Final = frozenset({"1", "true", "yes", "on"})
|
_TRUE_VALUES: Final = frozenset({"1", "true", "yes", "on"})
|
||||||
_FALSE_VALUES: Final = frozenset({"0", "false", "no", "off"})
|
_FALSE_VALUES: Final = frozenset({"0", "false", "no", "off"})
|
||||||
|
_IPV4_LIMITED_BROADCAST: Final = ipaddress.IPv4Address("255.255.255.255")
|
||||||
|
|
||||||
|
|
||||||
class OutboundHttpError(RuntimeError):
|
class OutboundHttpError(RuntimeError):
|
||||||
@@ -298,7 +299,12 @@ def _is_forbidden_special_address(value: str) -> bool:
|
|||||||
mapped = getattr(address, "ipv4_mapped", None)
|
mapped = getattr(address, "ipv4_mapped", None)
|
||||||
if mapped is not None:
|
if mapped is not None:
|
||||||
address = mapped
|
address = mapped
|
||||||
return address.is_link_local or address.is_multicast or address.is_unspecified
|
return (
|
||||||
|
address == _IPV4_LIMITED_BROADCAST
|
||||||
|
or address.is_link_local
|
||||||
|
or address.is_multicast
|
||||||
|
or address.is_unspecified
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _resolved_address_records(
|
def _resolved_address_records(
|
||||||
|
|||||||
@@ -78,7 +78,16 @@ class HttpFetchTests(unittest.TestCase):
|
|||||||
policy = outbound_http_policy(
|
policy = outbound_http_policy(
|
||||||
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"}
|
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"}
|
||||||
)
|
)
|
||||||
for address in ("0.0.0.0", "169.254.169.254", "224.0.0.1", "::", "fe80::1", "ff02::1"):
|
for address in (
|
||||||
|
"0.0.0.0",
|
||||||
|
"169.254.169.254",
|
||||||
|
"224.0.0.1",
|
||||||
|
"255.255.255.255",
|
||||||
|
"::",
|
||||||
|
"::ffff:255.255.255.255",
|
||||||
|
"fe80::1",
|
||||||
|
"ff02::1",
|
||||||
|
):
|
||||||
with self.subTest(address=address), patch(
|
with self.subTest(address=address), patch(
|
||||||
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
||||||
return_value=[(2, 1, 6, "", (address, 443))],
|
return_value=[(2, 1, 6, "", (address, 443))],
|
||||||
|
|||||||
Reference in New Issue
Block a user