modules/iproute2: Add filters in routes & addresses

Author: costasd

test/test_modules.py

@@ -650,6 +650,23 @@ def test_interface(host, family):
     assert default_itf.exists
 
 
+def test_iproute2_addresses(host):
+    assert host.iproute2.exists
+
+    addresses = host.iproute2.addresses()
+
+    assert len(addresses) > 0
+    assert addresses[0].get("ifname") and addresses[0].get("ifindex")
+
+    filtered_addresses = host.iproute2.addresses(ifname="lo")
+    assert filtered_addresses[0].get("ifname") == "lo" and len(filtered_addresses) == 1
+
+    filtered_addresses2 = host.iproute2.addresses(local="127.0.0.1")
+    assert (
+        filtered_addresses2[0].get("ifname") == "lo" and len(filtered_addresses2) == 1
+    )
+
+
 def test_iproute2_links(host):
     assert host.iproute2.exists
 
@@ -665,6 +682,9 @@ def test_iproute2_routes(host):
     routes = host.iproute2.routes()
     assert len(routes) > 0
 
+    filtered_routes = host.iproute2.routes(table="local", scope="host", src="127.0.0.1")
+    assert filtered_routes[0].get("protocol") == "kernel" and len(filtered_routes) > 1
+
 
 def test_iproute2_rules(host):
     assert host.iproute2.exists
@@ -675,6 +695,13 @@ def test_iproute2_rules(host):
     assert rules[0].get("src") == "all"
     assert rules[0].get("table") == "local"
 
+    cmd = host.run("ip rule add from 1.2.3.4/32 table 123")
+    assert cmd.exit_status == 0, f"{cmd.stdout}\n{cmd.stderr}"
+
+    rules_123 = host.iproute2.rules(src="1.2.3.4/32")
+    assert len(rules_123) > 0
+    assert rules_123[0].get("src") == "1.2.3.4"
+
 
 def test_iproute2_tunnels(host):
     assert host.iproute2.exists

testinfra/modules/iproute2.py

@@ -48,7 +48,7 @@ def __repr__(self):
     def _ip(self):
         ip_cmd = self.find_command("ip")
         if self.namespace is not None:
-            ip_cmd = f"{ip_cmd} netns exec {self.namespace} {ip_cmd}"
+            ip_cmd = f"{ip_cmd} -n {self.namespace}"
         if self.family is not None:
             ip_cmd = f"{ip_cmd} -f {self.family}"
         return ip_cmd
@@ -57,27 +57,107 @@ def _ip(self):
     def exists(self):
         return self.run_test("{} -V".format(self._ip)).rc == 0
 
-    def addresses(self):
+    def addresses(self, address=None, ifname=None, local=None):
         """Return the addresses associated with interfaces"""
         cmd = f"{self._ip} --json address show"
         out = self.check_output(cmd)
-        return json.loads(out)
+        j = json.loads(out)
+        o = []
+        if address is None and ifname is None and local is None:
+            # no filters, bail out early
+            return j
+        if address is not None:
+            [o.append(x) for x in j if x["address"] == address]
+        if ifname is not None:
+            [o.append(x) for x in j if x["ifname"] == ifname]
+        if local is not None:
+            for x in j:
+                for addr in x["addr_info"]:  # multiple IPs in an interface
+                    if addr["local"] == local:
+                        o.append(x)
+        return o
 
     def links(self):
         """Return links and their state"""
         cmd = f"{self._ip} --json link show"
         out = self.check_output(cmd)
         return json.loads(out)
 
-    def routes(self):
+    def routes(
+        self, table="all", device=None, scope=None, proto=None, src=None, metric=None
+    ):
         """Return the routes installed"""
-        cmd = f"{self._ip} --json route show table all"
+        cmd = f"{self._ip} --json route show "
+        options = []
+        if table is not None:
+            options += ["table", table]
+        if device is not None:
+            options += ["dev", device]
+        if scope is not None:
+            options += ["scope", scope]
+        if proto is not None:
+            options += ["proto", proto]
+        if src is not None:
+            options += ["src", src]
+        if metric is not None:
+            options += ["metric", metric]
+
+        cmd += " ".join(options)
         out = self.check_output(cmd)
         return json.loads(out)
 
-    def rules(self):
+    def rules(
+        self,
+        src=None,
+        to=None,
+        tos=None,
+        fwmark=None,
+        iif=None,
+        oif=None,
+        pref=None,
+        uidrange=None,
+        ipproto=None,
+        sport=None,
+        dport=None,
+    ):
         """Return the rules our routing policy consists of"""
-        cmd = f"{self._ip} --json rule show"
+        cmd = f"{self._ip} --json rule show "
+
+        options = []
+        if src is not None:
+            options += ["from", src]
+
+        if to is not None:
+            options += ["to", to]
+
+        if tos is not None:
+            options += ["tos", tos]
+
+        if fwmark is not None:
+            options += ["fwmark", fwmark]
+
+        if iif is not None:
+            options += ["iif", iif]
+
+        if oif is not None:
+            options += ["oif", oif]
+
+        if pref is not None:
+            options += ["pref", pref]
+
+        if uidrange is not None:
+            options += ["uidrange", uidrange]
+
+        if ipproto is not None:
+            options += ["ipproto", ipproto]
+
+        if sport is not None:
+            options += ["sport", sport]
+
+        if dport is not None:
+            options += ["dport", dport]
+
+        cmd += " ".join(options)
         out = self.check_output(cmd)
         return json.loads(out)