Add firewall state tests

Signed-off-by: bartoszWojciechO <[email protected]>

test/qa/lib/firewall.py

@@ -78,6 +78,19 @@
 # -A OUTPUT -o {iface} -m mark --mark 0xe1f1 -m comment --comment nordvpn -j ACCEPT
 # -A OUTPUT -o {iface} -m comment --comment nordvpn -j DROP
 
+inputLanDiscoveryRules = [
+    "-A INPUT -s 169.254.0.0/16 -i eth0 -m comment --comment nordvpn -j ACCEPT",
+    "-A INPUT -s 192.168.0.0/16 -i eth0 -m comment --comment nordvpn -j ACCEPT",
+    "-A INPUT -s 172.16.0.0/12 -i eth0 -m comment --comment nordvpn -j ACCEPT",
+    "-A INPUT -s 10.0.0.0/8 -i eth0 -m comment --comment nordvpn -j ACCEPT",
+]
+
+outputLanDiscoveryRules = [
+    "-A OUTPUT -d 169.254.0.0/16 -o eth0 -m comment --comment nordvpn -j ACCEPT",
+    "-A OUTPUT -d 192.168.0.0/16 -o eth0 -m comment --comment nordvpn -j ACCEPT",
+    "-A OUTPUT -d 172.16.0.0/12 -o eth0 -m comment --comment nordvpn -j ACCEPT",
+    "-A OUTPUT -d 10.0.0.0/8 -o eth0 -m comment --comment nordvpn -j ACCEPT",
+]
 
 # ToDo: Add missing IPv6 rules (icmp6 & dhcp6)
 def _get_firewall_rules(killswitch, server_ip, iface, port="", protocol="", subnet=""):

test/qa/lib/meshnet.py

@@ -8,11 +8,19 @@
 
 PEER_USERNAME = os.environ.get("QA_PEER_USERNAME")
 
+LANS = [
+        "169.254.0.0/16",
+        "192.168.0.0/16",
+        "172.16.0.0/12",
+        "10.0.0.0/8",
+]
+
 class PeerName(Enum):
     Hostname = 0
     Ip = 1
     Pubkey = 2
 
+
 def get_peer_name(output: str, name_type: PeerName) -> str:
     match name_type:
         case PeerName.Hostname:
@@ -22,16 +30,37 @@ def get_peer_name(output: str, name_type: PeerName) -> str:
         case PeerName.Pubkey:
             return get_this_device_pubkey(output)
 
-def add_peer(ssh_client: ssh.Ssh, tester_allow_fileshare: bool = True, peer_allow_fileshare: bool = True):
+
+def add_peer(ssh_client: ssh.Ssh,
+             tester_allow_fileshare: bool = True,
+             tester_allow_routing: bool = True,
+             tester_allow_local: bool = True,
+             tester_allow_incoming: bool = True,
+             peer_allow_fileshare: bool = True,
+             peer_allow_routing: bool = True,
+             peer_allow_local: bool = True,
+             peer_allow_incoming: bool = True):
     """
     adds QA peer to meshnet
     try to minimize usage of this, because there's a weekly invite limit
     """
-    tester_allow_fileshare_arg = f"-allow-peer-send-files={str(tester_allow_fileshare).lower()}"
-    peer_allow_fileshare_arg = f"-allow-peer-send-files={str(peer_allow_fileshare).lower()}"
-    sh.nordvpn.mesh.inv.send("--allow-incoming-traffic=true", "--allow-traffic-routing=true", tester_allow_fileshare_arg, PEER_USERNAME)
+    tester_allow_fileshare_arg = f"--allow-peer-send-files={str(tester_allow_fileshare).lower()}"
+    tester_allow_routing_arg = f"--allow-traffic-routing={str(tester_allow_routing).lower()}"
+    tester_allow_local_arg = f"--allow-local-network-access={str(tester_allow_local).lower()}"
+    tester_allow_incoming_arg = f"--allow-incoming-traffic={str(tester_allow_incoming).lower()}"
+
+
+    peer_allow_fileshare_arg = f"--allow-peer-send-files={str(peer_allow_fileshare).lower()}"
+    peer_allow_routing_arg = f"--allow-traffic-routing={str(peer_allow_routing).lower()}"
+    peer_allow_local_arg = f"--allow-local-network-access={str(peer_allow_local).lower()}"
+    peer_allow_incoming_arg = f"--allow-incoming-traffic={str(peer_allow_incoming).lower()}"
+
+    sh.nordvpn.mesh.inv.send(tester_allow_incoming_arg, tester_allow_local_arg, tester_allow_routing_arg, tester_allow_fileshare_arg, PEER_USERNAME)
     local_user, _ = login.get_default_credentials()
-    ssh_client.exec_command(f"yes | nordvpn mesh inv accept --allow-incoming-traffic=true --allow-traffic-routing=true {peer_allow_fileshare_arg} {local_user}")
+    ssh_client.exec_command(f"yes | nordvpn mesh inv accept {peer_allow_local_arg} {peer_allow_incoming_arg} {peer_allow_routing_arg} {peer_allow_fileshare_arg} {local_user}")
+
+    sh.nordvpn.mesh.peer.refresh()
+
 
 def get_peers(output: str) -> list:
     """parses list of peer names from 'nordvpn meshnet peer list' output"""
@@ -42,20 +71,23 @@ def get_peers(output: str) -> list:
             peers.append(line.split(" ")[1])
     return peers
 
+
 def get_this_device(output: str):
     """parses current device hostname from 'nordvpn meshnet peer list' output"""
     output_lines = output.split("\n")
     for i, line in enumerate(output_lines):
         if line.find("This device:") != -1:
             return output_lines[i+1].split(" ")[1]
 
+
 def get_this_device_ipv4(output: str):
     """parses current device ip from 'nordvpn meshnet peer list' output"""
     output_lines = output.split("\n")
     for i, line in enumerate(output_lines):
         if line.find("This device:") != -1:
             return output_lines[i+2].split(" ")[1]
 
+
 def get_this_device_pubkey(output: str):
     """parses current device pubkey from 'nordvpn meshnet peer list' output"""
     output_lines = output.split("\n")
@@ -64,18 +96,21 @@ def get_this_device_pubkey(output: str):
             # example: Public Key: uAexQo2yuiVBZocvuiFPQjAujkDmQVemKaircpxDaUc=
             return output_lines[i+3].split(" ")[2]
 
+
 def remove_all_peers():
     """removes all meshnet peers from local device"""
     output = f"{sh.nordvpn.mesh.peer.list(_tty_out=False)}" # convert to string, _tty_out false disables colors
     for p in get_peers(output):
         sh.nordvpn.mesh.peer.remove(p)
 
+
 def remove_all_peers_in_peer(ssh_client: ssh.Ssh):
     """removes all meshnet peers from peer device"""
     output = ssh_client.exec_command("nordvpn mesh peer list")
     for p in get_peers(output):
         ssh_client.exec_command(f"nordvpn mesh peer remove {p}")
 
+
 def is_peer_reachable(ssh_client: ssh.Ssh, retry: int = 5) -> bool:
     """returns True when ping to peer succeeds."""
     output = ssh_client.exec_command("nordvpn mesh peer list")
@@ -94,6 +129,7 @@ def is_peer_reachable(ssh_client: ssh.Ssh, retry: int = 5) -> bool:
     print(output)
     return False
 
+
 def get_sent_invites(output: str) -> list:
     """parses list of sent invites from 'nordvpn meshnet inv list' output"""
     emails = []
@@ -104,12 +140,14 @@ def get_sent_invites(output: str) -> list:
             emails.append(line.split(" ")[1])
     return emails
 
+
 def revoke_all_invites():
     """revokes all sent meshnet invites in local device"""
     output = f"{sh.nordvpn.mesh.inv.list(_tty_out=False)}" # convert to string, _tty_out false disables colors
     for i in get_sent_invites(output):
         sh.nordvpn.mesh.inv.revoke(i)
 
+
 def revoke_all_invites_in_peer(ssh_client: ssh.Ssh):
     """revokes all sent meshnet invites in peer device"""
     output = ssh_client.exec_command("nordvpn mesh inv list")

test/qa/test_firewall.py

@@ -29,6 +29,7 @@ def setup_function(function):
 def teardown_function(function):
     logging.log(data=info.collect())
     logging.log()
+    sh.nordvpn.set("lan-discovery", "off", _ok_code=(0,1))
 
 
 @pytest.mark.flaky(reruns=2, reruns_delay=90)
@@ -267,4 +268,58 @@ def test_firewall_exitnode():
     assert lib.is_disconnect_successful(output)
     assert network.is_disconnected()
 
-    assert not firewall.is_active()
+    assert not firewall.is_active()
+
+
+@pytest.mark.parametrize("before_connect", [True, False])
+def test_firewall_lan_discovery(before_connect):
+    if before_connect:
+        sh.nordvpn.set("lan-discovery", "on")
+
+    sh.nordvpn.connect()
+
+    if not before_connect:
+        sh.nordvpn.set("lan-discovery", "on")
+
+    rules = sh.sudo.iptables("-S", "INPUT")
+    for rule in firewall.inputLanDiscoveryRules:
+        assert rule in rules, f"{rule} input rule not found in iptables."
+
+    rules = sh.sudo.iptables("-S", "OUTPUT")
+    for rule in firewall.outputLanDiscoveryRules:
+        assert rule in rules, f"{rule} output rule not found in iptables"
+
+    sh.nordvpn.set("lan-discovery", "off")
+
+    rules = sh.sudo.iptables("-S", "INPUT")
+    for rule in firewall.inputLanDiscoveryRules:
+        assert rule not in rules, f"{rule} input rule not found in iptables."
+
+    rules = sh.sudo.iptables("-S", "OUTPUT")
+    for rule in firewall.outputLanDiscoveryRules:
+        assert rule not in rules, f"{rule} output rule not found in iptables"
+
+
+def test_firewall_lan_allowlist_interaction():
+    sh.nordvpn.connect()
+
+    subnet = "192.168.0.0/18"
+
+    sh.nordvpn.allowlist.add.subnet(subnet)
+    sh.nordvpn.set("lan-discovery", "on")
+
+    rules = sh.sudo.iptables("-S", "INPUT")
+    assert f"-A INPUT -s {subnet} -i eth0 -m comment --comment nordvpn -j ACCEPT" not in rules, "Whitelist rule was not removed from the INPUT chain when LAN discovery was enabled."
+
+    rules = sh.sudo.iptables("-S", "OUTPUT")
+    assert f"-A OUTPUT -s {subnet} -o eth0 -m comment --comment nordvpn -j ACCEPT" not in rules, "Whitelist rule was not removed from the OUTPUT chain when LAN discovery was enabled."
+
+    sh.nordvpn.set("lan-discovery", "off")
+
+    rules = sh.sudo.iptables("-S", "INPUT")
+    for rule in firewall.inputLanDiscoveryRules:
+        assert rule not in rules, f"{rule} input rule not found in iptables."
+
+    rules = sh.sudo.iptables("-S", "OUTPUT")
+    for rule in firewall.outputLanDiscoveryRules:
+        assert rule not in rules, f"{rule} output rule not found in iptables"