diff --git a/ui/opensnitch/dialogs/ruleseditor/dialog.py b/ui/opensnitch/dialogs/ruleseditor/dialog.py index 160d366a27..97e3648e69 100644 --- a/ui/opensnitch/dialogs/ruleseditor/dialog.py +++ b/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()): diff --git a/ui/opensnitch/nodes.py b/ui/opensnitch/nodes.py index a6b126576f..883303b679 100644 --- a/ui/opensnitch/nodes.py +++ b/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 diff --git a/ui/tests/dialogs/test_ruleseditor.py b/ui/tests/dialogs/test_ruleseditor.py index 68dac1198c..83c7a0f882 100644 --- a/ui/tests/dialogs/test_ruleseditor.py +++ b/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) diff --git a/ui/tests/test_nodes.py b/ui/tests/test_nodes.py index 60b90c9dc6..3fed8f800e 100644 --- a/ui/tests/test_nodes.py +++ b/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)