ui: resolve usernames with getent for UID rules

ui/opensnitch/dialogs/ruleseditor/dialog.py

@@ -5,6 +5,7 @@
 import sys
 import os
 import pwd
+import subprocess
 import time
 import ipaddress
 
@@ -49,7 +50,7 @@ def __init__(self, parent=None, modal=True, appicon=None):
         self._rules = Rules.instance()
         self._notification_callback.connect(self.cb_notification_callback)
         self._old_rule_name = None
-        self._users_list = pwd.getpwall()
+        self._users_list = []
 
         self.setupUi(self)
         self.setModal(modal)
@@ -79,21 +80,25 @@ def init(self):
         self.ifaceCombo.clear()
         self.uidCombo.clear()
         addr = nodes.get_node_addr(self)
-        if addr is not None and self._nodes.is_local(addr):
-            self.ifaceCombo.addItems(NetworkInterfaces.list().keys())
+        if addr is not None:
+            if self._nodes.is_local(addr):
+                self.ifaceCombo.addItems(NetworkInterfaces.list().keys())
+                try:
+                    for ip in NetworkInterfaces.list().values():
+                        if self.srcIPCombo.findText(ip) == -1:
+                            self.srcIPCombo.insertItem(0, ip)
+                        if self.dstIPCombo.findText(ip) == -1:
+                            self.dstIPCombo.insertItem(0, ip)
+                except Exception as e:
+                    self.logger.warning("Error adding IPs: %s", repr(e))
+
+            self.uidCombo.blockSignals(True);
             try:
-                for ip in NetworkInterfaces.list().values():
-                    if self.srcIPCombo.findText(ip) == -1:
-                        self.srcIPCombo.insertItem(0, ip)
-                    if self.dstIPCombo.findText(ip) == -1:
-                        self.dstIPCombo.insertItem(0, ip)
-
-                self._users_list = pwd.getpwall()
-                self.uidCombo.blockSignals(True);
+                self._users_list = self._get_users(addr)
                 for user in self._users_list:
                     self.uidCombo.addItem("{0} ({1})".format(user[constants.PW_USER], user[constants.PW_UID]), user[constants.PW_UID])
             except Exception as e:
-                self.logger.warning("Error adding IPs: %s", repr(e))
+                self.logger.warning("Error loading users: %s", repr(e))
             finally:
                 self.uidCombo.blockSignals(False);
 
@@ -225,8 +230,91 @@ def cb_dstnetlists_check_toggled(self, state):
         self.dstListNetsLine.setEnabled(state)
         self.selectNetsListButton.setEnabled(state)
 
+    def _get_users(self, addr):
+        if not self._nodes.is_local(addr):
+            return []
+
+        users = []
+        result = subprocess.run(
+            ["getent", "passwd"],
+            capture_output=True,
+            text=True,
+            timeout=10
+        )
+        if result.returncode != 0:
+            return users
+
+        for line in result.stdout.splitlines():
+            parts = line.split(":")
+            if len(parts) < 7 or parts[6] in ("/sbin/nologin", "/bin/false"):
+                continue
+            try:
+                users.append((
+                    parts[0],
+                    "x",
+                    int(parts[2]),
+                    int(parts[3]),
+                    parts[4],
+                    parts[5],
+                    parts[6]
+                ))
+            except ValueError:
+                self.logger.debug("invalid passwd entry returned by getent: %s", line)
+
+        return users
+
+    def _resolve_uid(self, username):
+        try:
+            return pwd.getpwnam(username)[constants.PW_UID]
+        except KeyError:
+            pass
+
+        try:
+            result = subprocess.run(
+                ["getent", "passwd", username],
+                capture_output=True,
+                text=True,
+                timeout=5
+            )
+            if result.returncode == 0 and result.stdout.strip():
+                return int(result.stdout.split(":", 3)[2])
+        except (IndexError, OSError, ValueError, subprocess.SubprocessError):
+            pass
+
+        addr = nodes.get_node_addr(self)
+        if addr is None or self._nodes.is_local(addr):
+            raise KeyError(username)
+
+        _, _, host = addr.partition(":")
+        if host == "":
+            raise KeyError(username)
+
+        try:
+            result = subprocess.run(
+                [
+                    "ssh",
+                    "-o", "ConnectTimeout=5",
+                    "-o", "StrictHostKeyChecking=no",
+                    "-o", "BatchMode=yes",
+                    "root@%s" % host,
+                    "getent",
+                    "passwd",
+                    username
+                ],
+                capture_output=True,
+                text=True,
+                timeout=10
+            )
+            if result.returncode == 0 and result.stdout.strip():
+                return int(result.stdout.split(":", 3)[2])
+        except (IndexError, OSError, ValueError, subprocess.SubprocessError):
+            pass
+
+        raise KeyError(username)
+
     def cb_uid_combo_changed(self, index):
-        self.uidCombo.setCurrentText(str(self._users_list[index][constants.PW_UID]))
+        if 0 <= index < len(self._users_list):
+            self.uidCombo.setCurrentText(str(self._users_list[index][constants.PW_UID]))
 
     def cb_nodes_combo_changed(self, index):
         addr = self.nodesCombo.itemData(index)
@@ -724,15 +812,15 @@ def save_rule(self):
                 return False, QC.translate("rules", "User ID can not be empty")
 
             try:
-                # sometimes when loading a rule, instead of the UID, the format
-                # "user (uid)" is set. So try to parse it, in order not to save
-                # a wrong uid.
                 uidtmp = uid.split(" ")
                 if len(uidtmp) == 1:
-                    int(uidtmp[0])
+                    try:
+                        int(uidtmp[0])
+                    except ValueError:
+                        uid = str(self._resolve_uid(uidtmp[0]))
                 else:
-                    uid = str(pwd.getpwnam(uidtmp[0])[constants.PW_UID])
-            except:
+                    uid = str(self._resolve_uid(uidtmp[0]))
+            except (KeyError, ValueError):
                 # if it's not a digit and nor a system user (user (id)), see if
                 # it's a regexp.
                 if utils.is_regex(self, self.uidCombo.currentText()):

ui/opensnitch/nodes.py

@@ -1,6 +1,7 @@
 from PyQt6.QtCore import QObject, pyqtSignal
 from queue import Queue
 from datetime import datetime
+import ipaddress
 import time
 import json
 
@@ -255,9 +256,17 @@ def is_local(self, addr):
             return True
 
         if addr.startswith("ipv4") or addr.startswith("ipv6"):
+            _, _, host = addr.partition(":")
+            host = host.strip("[]")
+            try:
+                if ipaddress.ip_address(host).is_loopback:
+                    return True
+            except ValueError:
+                pass
+
             ifaces = self._interfaces.list()
             for name in ifaces:
-                if ifaces[name] in addr:
+                if ifaces[name] == host:
                     return True
 
         return False

ui/tests/dialogs/test_ruleseditor.py

@@ -12,6 +12,7 @@
 from opensnitch.config import Config
 from opensnitch.dialogs.ruleseditor import RulesEditorDialog
 from opensnitch.dialogs.ruleseditor import constants as re_constants
+from opensnitch.dialogs.ruleseditor import dialog as re_dialog
 from opensnitch.dialogs.ruleseditor import rules as re_rules
 from opensnitch.dialogs.ruleseditor import utils as re_utils
 from opensnitch.dialogs.ruleseditor import nodes as re_nodes
@@ -443,6 +444,88 @@ def test_rule_with_uid(self, qtbot):
         assert self.rd._db.get_rule("test-uid", self.rd.nodesCombo.currentText()).next() == True
         assert self.rd.rule.operator.operand == Config.OPERAND_USER_ID
 
+    def test_rule_with_username_uid(self, qtbot, monkeypatch):
+        """Test creating a UID rule with a username resolved through NSS."""
+        qtbot.addWidget(self.rd)
+        re_constants.WORK_MODE = re_constants.ADD_RULE
+        re_utils.reset_state(self.rd)
+        self.rd.ruleNameEdit.setText("test-uid-user")
+        self.rd.uidCheck.setChecked(True)
+        self.rd.uidCombo.setCurrentText("test1@ad.zs")
+        monkeypatch.setattr(self.rd, "_resolve_uid", lambda username: 693401141)
+
+        result, error = self.rd.save_rule()
+
+        assert result is True
+        assert error == ""
+        assert self.rd.rule.operator.operand == Config.OPERAND_USER_ID
+        assert self.rd.rule.operator.data == "693401141"
+
+    def test_uid_combo_changed_ignores_invalid_index(self, qtbot):
+        """Test typing custom UID text does not reuse the last user entry."""
+        qtbot.addWidget(self.rd)
+        self.rd._users_list = [
+            ("alice", "x", 1000, 1000, "", "/home/alice", "/bin/bash"),
+            ("bob", "x", 2000, 2000, "", "/home/bob", "/bin/bash"),
+        ]
+        self.rd.uidCombo.setCurrentText("test1@ad.zs")
+
+        self.rd.cb_uid_combo_changed(-1)
+
+        assert self.rd.uidCombo.currentText() == "test1@ad.zs"
+
+    def test_get_users_uses_getent_for_local_nodes(self, qtbot, monkeypatch):
+        """Test local user list loading uses getent passwd output."""
+        qtbot.addWidget(self.rd)
+
+        class Result:
+            returncode = 0
+            stdout = (
+                "alice:x:1000:1000:Alice:/home/alice:/bin/bash\n"
+                "daemon:x:2:2:daemon:/sbin:/sbin/nologin\n"
+            )
+
+        monkeypatch.setattr(self.rd._nodes, "is_local", lambda addr: True)
+        monkeypatch.setattr(re_dialog.subprocess, "run", lambda *args, **kwargs: Result())
+
+        users = self.rd._get_users("unix:/tmp/osui.sock")
+
+        assert users == [("alice", "x", 1000, 1000, "Alice", "/home/alice", "/bin/bash")]
+
+    def test_resolve_uid_uses_remote_getent(self, qtbot, monkeypatch):
+        """Test remote usernames are resolved via non-interactive SSH getent."""
+        qtbot.addWidget(self.rd)
+        self.rd.nodesCombo.clear()
+        self.rd.nodesCombo.addItem("ipv4:192.168.122.19", "ipv4:192.168.122.19")
+        calls = []
+
+        class MissingResult:
+            returncode = 2
+            stdout = ""
+
+        class RemoteResult:
+            returncode = 0
+            stdout = "test1@ad.zs:x:693401141:693400513:Test:/home/test1:/bin/bash\n"
+
+        def fake_getpwnam(username):
+            raise KeyError(username)
+
+        def fake_run(args, **kwargs):
+            calls.append(args)
+            if args[:3] == ["getent", "passwd", "test1@ad.zs"]:
+                return MissingResult()
+            return RemoteResult()
+
+        monkeypatch.setattr(re_dialog.pwd, "getpwnam", fake_getpwnam)
+        monkeypatch.setattr(re_dialog.subprocess, "run", fake_run)
+        monkeypatch.setattr(self.rd._nodes, "is_local", lambda addr: False)
+
+        uid = self.rd._resolve_uid("test1@ad.zs")
+
+        assert uid == 693401141
+        assert calls[-1][:2] == ["ssh", "-o"]
+        assert "BatchMode=yes" in calls[-1]
+
     def test_rule_with_source_port(self, qtbot):
         """Test creating a rule with source port."""
         qtbot.addWidget(self.rd)

ui/tests/test_nodes.py

@@ -182,6 +182,13 @@ def test_is_local_remote_address(self, qtbot):
         # Result depends on local interfaces, but 8.8.8.8 should not be local
         assert result == False
 
+    def test_is_local_uses_exact_address_match(self, qtbot, monkeypatch):
+        """Test local detection does not use substring matching."""
+        monkeypatch.setattr(self.nodes._interfaces, "list", lambda: {"virbr0": "192.168.122.1"})
+
+        assert self.nodes.is_local("ipv4:192.168.122.1") == True
+        assert self.nodes.is_local("ipv4:192.168.122.19") == False
+
     def test_get_node_hostname(self, qtbot):
         """Test hostname retrieval."""
         self.nodes.add("peer:1.2.3.4", self.daemon_config)