Ensure the forward rule is added after connection tracking.

Also improve tests and add Tox for mypy and pytest.

.gitignore

@@ -10,3 +10,4 @@ build
 *.egg-info
 src/*.service
 .mypy_cache
+.tox

Makefile

@@ -11,7 +11,7 @@ src/qubes-routing-manager.service: src/qubes-routing-manager.service.in
 
 ROOT_DIR := $(shell dirname $(realpath $(firstword $(MAKEFILE_LIST))))
 
-.PHONY: clean dist rpm srpm install-template install-dom0
+.PHONY: clean dist rpm srpm install-template install-dom0 test
 
 clean:
 	cd $(ROOT_DIR) || exit $$? ; find -name '*.pyc' -o -name '*~' -print0 | xargs -0 rm -f
@@ -42,3 +42,6 @@ install-dom0:
 	PYTHONDONTWRITEBYTECODE=1 python3 networkserversetup.py install $(PYTHON_PREFIX_ARG) -O0 --root $(DESTDIR)
 
 install: install-dom0 install-template
+
+test:
+	tox --current-env

qubes-network-server.spec

@@ -3,7 +3,7 @@
 %define mybuildnumber %{?build_number}%{?!build_number:1}
 
 Name:           qubes-network-server
-Version:        0.1.2
+Version:        0.1.3
 Release:        %{mybuildnumber}%{?dist}
 Summary:        Turn your Qubes OS into a network server
 BuildArch:      noarch
@@ -19,6 +19,9 @@ BuildRequires:  findutils
 BuildRequires:  python3
 BuildRequires:  python3-rpm-macros
 BuildRequires:  systemd-rpm-macros
+BuildRequires:  python3-tox-current-env
+BuildRequires:  python3-mypy
+BuildRequires:  python3-pytest
 
 Requires:       qubes-core-agent-networking >= 4.2
 Conflicts:      qubes-core-agent < 4.2
@@ -71,6 +74,9 @@ make install DESTDIR=$RPM_BUILD_ROOT SBINDIR=%{_sbindir} UNITDIR=%{_unitdir} PYT
 mkdir -p "$RPM_BUILD_ROOT"/%{_presetdir}
 echo 'enable qubes-routing-manager.service' > "$RPM_BUILD_ROOT"/%{_presetdir}/75-%{name}.preset
 
+%check
+tox --current-env
+
 %files
 %attr(0755, root, root) %{_sbindir}/qubes-routing-manager
 %attr(0644, root, root) %{python3_sitelib}/qubesroutingmanager/*

qubesroutingmanager/__init__.py

@@ -4,10 +4,10 @@ import json
 import logging
 import subprocess
 
-from typing import TypedDict, Any, cast, Literal
+from typing import TypedDict, Any, cast, Literal, Union
 
 
-ADDRESS_FAMILIES = Literal["ip"] | Literal["ip6"]
+ADDRESS_FAMILIES = Union[Literal["ip"], Literal["ip6"]]
 
 
 class Chain(TypedDict):
@@ -69,7 +69,6 @@ POSTROUTING_CHAIN_NAME = "postrouting"
 ROUTING_MANAGER_CHAIN_NAME = "qubes-routing-manager"
 ROUTING_MANAGER_POSTROUTING_CHAIN_NAME = "qubes-routing-manager-postrouting"
 NFTABLES_CMD = "nft"
-ADD_FORWARD_RULE_AFTER_THIS_RULE = "custom-forward"
 
 
 def get_table(address_family: ADDRESS_FAMILIES, table: str) -> NFTablesOutput:
@@ -254,12 +253,21 @@ def setup_plain_forwarding_for_address(source: str, enable: bool, family: int) -
                 chain_name,
             )
 
-    def is_forward_jump_to_custom_forward(rule):
+    def is_oifgroup_2(rule):
         return (
             rule["chain"] == forward_chain["name"]
-            and len(rule["expr"]) == 1
-            and rule["expr"][0].get("jump", {}).get("target")
-            == ADD_FORWARD_RULE_AFTER_THIS_RULE
+            and len(rule["expr"]) == 3
+            and (
+                rule["expr"][0].get("match", {}).get("op") == "=="
+                and rule["expr"][0]
+                .get("match", {})
+                .get("left", {})
+                .get("meta", {})
+                .get("key")
+                == "oifgroup"
+                and rule["expr"][0].get("match", {}).get("right") == 2
+            )
+            and (rule["expr"][-1].get("drop", "not none") is None)
         )
 
     def is_postrouting_masquerade(rule):
@@ -273,8 +281,8 @@ def setup_plain_forwarding_for_address(source: str, enable: bool, family: int) -
         (
             forward_chain,
             ROUTING_MANAGER_CHAIN_NAME,
-            is_forward_jump_to_custom_forward,
-            append_rule_after,
+            is_oifgroup_2,
+            insert_rule_before,
         ),
         (
             postrouting_chain,

qubesroutingmanager/test_firewalling.py

@@ -50,13 +50,13 @@ def test_partial_add_completes_the_add():
             "counter",
         ],
         [
-            "add",
+            "insert",
             "rule",
             "ip",
             "qubes",
             "postrouting",
             "position",
-            "66",
+            "67",
             "jump",
             "qubes-routing-manager-postrouting",
         ],
@@ -78,6 +78,79 @@ def test_partial_add_completes_the_add():
     assert got == expected
 
 
+def test_forward_rule_added_before_oifgroup_2():
+    got, MockedPopen = mock_collector(get_fixture("no_routing_manager.json"))
+    expected = [
+        ["list", "table", "ip", "qubes"],
+        ["add", "chain", "ip", "qubes", "qubes-routing-manager"],
+        [
+            "add",
+            "rule",
+            "ip",
+            "qubes",
+            "qubes-routing-manager",
+            "counter",
+        ],
+        ["add", "chain", "ip", "qubes", "qubes-routing-manager-postrouting"],
+        [
+            "add",
+            "rule",
+            "ip",
+            "qubes",
+            "qubes-routing-manager-postrouting",
+            "counter",
+        ],
+        [
+            "insert",
+            "rule",
+            "ip",
+            "qubes",
+            "forward",
+            "position",
+            "79",
+            "jump",
+            "qubes-routing-manager",
+        ],
+        [
+            "insert",
+            "rule",
+            "ip",
+            "qubes",
+            "postrouting",
+            "position",
+            "67",
+            "jump",
+            "qubes-routing-manager-postrouting",
+        ],
+        [
+            "add",
+            "rule",
+            "ip",
+            "qubes",
+            "qubes-routing-manager",
+            "ip",
+            "daddr",
+            "10.250.4.13",
+            "accept",
+        ],
+        [
+            "add",
+            "rule",
+            "ip",
+            "qubes",
+            "qubes-routing-manager-postrouting",
+            "ip",
+            "saddr",
+            "10.250.4.13",
+            "accept",
+        ],
+    ]
+    with mock.patch("subprocess.Popen", MockedPopen):
+        setup_plain_forwarding_for_address("10.250.4.13", True, 4)
+
+    assert got == expected
+
+
 def test_forwarding_does_not_add_twice():
     got, MockedPopen = mock_collector(get_fixture("fully_added.json"))
     expected = [

qubesroutingmanager/worker.py

@@ -14,7 +14,7 @@ import logging
 import os
 import socket
 
-import qubesdb
+import qubesdb  # type: ignore
 
 from qubesroutingmanager import setup_plain_forwarding_for_address
 

tox.ini

@@ -0,0 +1,10 @@
+[tox]
+envlist = basepython
+
+[testenv]
+deps =
+    pytest
+    mypy
+commands =
+    pytest -vv
+    mypy -p qubesroutingmanager