mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-10-16 21:25:14 +02:00
Merge pull request #2052 from cpuu/cpuu/fix-linux-ipv6-host-family
Linux: Fix IPv6 socket address decoding on non-Linux hosts
This commit is contained in:
@@ -0,0 +1,67 @@
|
||||
"""Linux dump address families must not depend on the analysis host OS."""
|
||||
|
||||
import ipaddress
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from volatility3.framework import contexts
|
||||
from volatility3.framework.layers import physical
|
||||
from volatility3.framework.symbols.linux.extensions import network
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host_ipv6", [10, 23, 30], ids=["linux", "windows", "macos"])
|
||||
@pytest.mark.parametrize("direction", ["src", "dst"])
|
||||
@pytest.mark.parametrize("legacy_destination", [False, True], ids=["common", "legacy"])
|
||||
@pytest.mark.parametrize(
|
||||
"dump_family, address",
|
||||
[(2, "192.0.2.123"), (10, "2001:db8:1234::5678")],
|
||||
ids=["ipv4", "ipv6"],
|
||||
)
|
||||
def test_inet_socket_address_host_family(
|
||||
monkeypatch, host_ipv6, direction, legacy_destination, dump_family, address
|
||||
):
|
||||
"""Decode Linux family 10 using the host's own IPv6 conversion constant."""
|
||||
packed = ipaddress.ip_address(address).packed
|
||||
context = contexts.Context()
|
||||
context.layers.add_layer(
|
||||
physical.BufferDataLayer(context, "memory", "memory", packed)
|
||||
)
|
||||
field = SimpleNamespace(vol=SimpleNamespace(offset=0))
|
||||
common = SimpleNamespace(
|
||||
skc_family=dump_family,
|
||||
skc_rcv_saddr=field,
|
||||
skc_daddr=field,
|
||||
skc_v6_daddr=field,
|
||||
)
|
||||
sk = SimpleNamespace()
|
||||
# StructType normally resolves Python's name-mangled kernel member name.
|
||||
setattr(sk, "_inet_sock__sk_common", common)
|
||||
inet = SimpleNamespace(
|
||||
sk=sk,
|
||||
pinet6=SimpleNamespace(saddr=field),
|
||||
_context=context,
|
||||
vol=SimpleNamespace(layer_name="memory"),
|
||||
)
|
||||
inet.get_family = lambda: network.inet_sock.get_family(inet)
|
||||
if legacy_destination:
|
||||
inet.daddr = field
|
||||
inet.rcv_saddr = field
|
||||
inet.pinet6.daddr = field
|
||||
|
||||
expected_host_family = 2 if dump_family == 2 else host_ipv6
|
||||
|
||||
def inet_ntop(family, data):
|
||||
assert family == expected_host_family
|
||||
return str(ipaddress.ip_address(data))
|
||||
|
||||
conversion = Mock(side_effect=inet_ntop)
|
||||
monkeypatch.setattr(
|
||||
network,
|
||||
"socket_module",
|
||||
SimpleNamespace(AF_INET=2, AF_INET6=host_ipv6, inet_ntop=conversion),
|
||||
)
|
||||
result = getattr(network.inet_sock, f"get_{direction}_addr")(inet)
|
||||
assert result == address
|
||||
conversion.assert_called_once_with(expected_host_family, packed)
|
||||
@@ -546,8 +546,10 @@ class inet_sock(objects.StructType):
|
||||
|
||||
def get_src_addr(self) -> Optional[str]:
|
||||
sk_common = self.sk.__sk_common
|
||||
family = sk_common.skc_family
|
||||
if family == socket_module.AF_INET:
|
||||
# Interpret the Linux family independently of the analysis host OS.
|
||||
family = self.get_family()
|
||||
if family == "AF_INET":
|
||||
host_family = socket_module.AF_INET
|
||||
addr_size = 4
|
||||
if hasattr(self, "rcv_saddr"):
|
||||
saddr = self.rcv_saddr
|
||||
@@ -555,7 +557,8 @@ class inet_sock(objects.StructType):
|
||||
saddr = self.inet_rcv_saddr
|
||||
else:
|
||||
saddr = sk_common.skc_rcv_saddr
|
||||
elif family == socket_module.AF_INET6:
|
||||
elif family == "AF_INET6":
|
||||
host_family = socket_module.AF_INET6
|
||||
addr_size = 16
|
||||
saddr = self.pinet6.saddr
|
||||
else:
|
||||
@@ -568,12 +571,14 @@ class inet_sock(objects.StructType):
|
||||
f"Unable to read socket src address from {saddr.vol.offset:#x}"
|
||||
)
|
||||
return None
|
||||
return socket_module.inet_ntop(family, addr_bytes)
|
||||
return socket_module.inet_ntop(host_family, addr_bytes)
|
||||
|
||||
def get_dst_addr(self) -> Optional[str]:
|
||||
sk_common = self.sk.__sk_common
|
||||
family = sk_common.skc_family
|
||||
if family == socket_module.AF_INET:
|
||||
# Interpret the Linux family independently of the analysis host OS.
|
||||
family = self.get_family()
|
||||
if family == "AF_INET":
|
||||
host_family = socket_module.AF_INET
|
||||
if hasattr(self, "daddr") and self.daddr:
|
||||
daddr = self.daddr
|
||||
elif hasattr(self, "inet_daddr") and self.inet_daddr:
|
||||
@@ -581,7 +586,8 @@ class inet_sock(objects.StructType):
|
||||
else:
|
||||
daddr = sk_common.skc_daddr
|
||||
addr_size = 4
|
||||
elif family == socket_module.AF_INET6:
|
||||
elif family == "AF_INET6":
|
||||
host_family = socket_module.AF_INET6
|
||||
if hasattr(self.pinet6, "daddr"):
|
||||
daddr = self.pinet6.daddr
|
||||
else:
|
||||
@@ -597,7 +603,7 @@ class inet_sock(objects.StructType):
|
||||
f"Unable to read socket dst address from {daddr.vol.offset:#x}"
|
||||
)
|
||||
return None
|
||||
return socket_module.inet_ntop(family, addr_bytes)
|
||||
return socket_module.inet_ntop(host_family, addr_bytes)
|
||||
|
||||
|
||||
class netlink_sock(objects.StructType):
|
||||
|
||||
Reference in New Issue
Block a user