storm/fenrir
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix to socket routing for self-voicing.

Browse Source
This commit is contained in:
Storm Dragon
2026-07-30 04:52:52 -04:00
parent 122a6a38b3
commit 7593c8d0b3
3 changed files with 153 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fenrirscreenreader/fenrirVersion.py
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
version = "2026.07.26"
version = "2026.07.30"
code_name = "testing"
src/fenrirscreenreader/remoteDriver/unixDriver.py
+39 -2
View File
@@ -161,6 +161,23 @@ class driver(remoteDriver):
except OSError:
return 0
def _get_peer_uid(self, client_sock):
so_peercred = getattr(socket, "SO_PEERCRED", 17)
try:
creds = client_sock.getsockopt(
socket.SOL_SOCKET, so_peercred, struct.calcsize("3i")
)
_pid, uid, _gid = struct.unpack("3i", creds)
return uid
except OSError:
return -1
def _get_socket_owner_uid(self, socket_file):
try:
return os.stat(socket_file).st_uid
except OSError:
return -1
def _get_parent_pid(self, pid):
try:
with open(f"/proc/{pid}/stat", "r", encoding="utf-8") as proc_file:
@@ -206,13 +223,20 @@ class driver(remoteDriver):
socket_files.append(socket_file)
return socket_files
def _find_available_private_socket(self, preferred_socket=""):
def _find_available_private_socket(
self, preferred_socket="", owner_uid=None
):
socket_files = self._get_registered_private_sockets()
if preferred_socket and preferred_socket in socket_files:
socket_files.remove(preferred_socket)
socket_files.insert(0, preferred_socket)
for socket_file in socket_files:
if (
owner_uid is not None
and self._get_socket_owner_uid(socket_file) != owner_uid
):
continue
if self._is_own_socket_file(socket_file):
return socket_file
if self._is_socket_active(socket_file):
@@ -253,14 +277,27 @@ class driver(remoteDriver):
def _route_main_socket_command(self, data, client_sock, socket_file):
if socket_file != MAIN_SOCKET_FILE:
return False
owner_uid = None
if not self._has_own_private_socket():
peer_uid = self._get_peer_uid(client_sock)
main_socket_uid = self._get_socket_owner_uid(MAIN_SOCKET_FILE)
if peer_uid < 0 or peer_uid == main_socket_uid:
return False
owner_uid = peer_uid
peer_pid = self._get_peer_pid(client_sock)
ancestor_socket = ""
if peer_pid > 1:
ancestor_socket = self._find_ancestor_private_socket(peer_pid)
target_socket = self._find_available_private_socket(ancestor_socket)
if owner_uid is None:
target_socket = self._find_available_private_socket(
ancestor_socket
)
else:
target_socket = self._find_available_private_socket(
ancestor_socket, owner_uid=owner_uid
)
if not target_socket:
return False
tests/unit/test_unix_remote_driver.py
+112 -1
View File
@@ -1,3 +1,5 @@
import socket
import struct
from unittest.mock import Mock, mock_open, patch
import pytest
@@ -75,6 +77,19 @@ class FakeClientSocket:
self.sent += data
def test_get_peer_uid_from_unix_socket():
driver = unixDriver.driver()
client_sock = Mock()
client_sock.getsockopt.return_value = struct.pack("3i", 4321, 1000, 100)
assert driver._get_peer_uid(client_sock) == 1000
client_sock.getsockopt.assert_called_once_with(
socket.SOL_SOCKET,
socket.SO_PEERCRED,
struct.calcsize("3i"),
)
def test_main_socket_routes_to_ancestor_private_socket(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
@@ -148,6 +163,10 @@ def test_main_socket_handles_command_locally_without_available_target(
event_queue = Mock()
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
driver, "_get_peer_uid", return_value=1000
), patch.object(
driver, "_get_socket_owner_uid", return_value=65534
), patch.object(
driver,
"_find_available_private_socket",
return_value="",
@@ -171,7 +190,11 @@ def test_vcsa_main_socket_owner_handles_command_locally(mock_environment):
client_sock = FakeClientSocket(b"command say root")
event_queue = Mock()
with patch.object(driver, "_find_available_private_socket") as find_available:
with patch.object(driver, "_get_peer_uid", return_value=1000), patch.object(
driver, "_get_socket_owner_uid", return_value=1000
), patch.object(
driver, "_find_available_private_socket"
) as find_available:
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
@@ -185,6 +208,64 @@ def test_vcsa_main_socket_owner_handles_command_locally(mock_environment):
)
def test_vcsa_main_socket_routes_to_same_user_private_socket(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
driver.bound_sockets = [(Mock(), unixDriver.MAIN_SOCKET_FILE)]
client_sock = FakeClientSocket(b"command say x session")
event_queue = Mock()
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
driver, "_get_peer_uid", return_value=1000
), patch.object(
driver, "_get_socket_owner_uid", return_value=65534
), patch.object(
driver, "_find_ancestor_private_socket", return_value=""
), patch.object(
driver,
"_find_available_private_socket",
return_value="/tmp/fenrirscreenreader-222.sock",
) as find_available, patch.object(
driver, "_forward_remote_to_socket", return_value=True
) as forward:
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
find_available.assert_called_once_with("", owner_uid=1000)
forward.assert_called_once_with(
"command say x session", "/tmp/fenrirscreenreader-222.sock"
)
event_queue.put.assert_not_called()
def test_vcsa_main_socket_stays_local_without_peer_credentials(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
driver.bound_sockets = [(Mock(), unixDriver.MAIN_SOCKET_FILE)]
client_sock = FakeClientSocket(b"command say unknown")
event_queue = Mock()
with patch.object(driver, "_get_peer_uid", return_value=-1), patch.object(
driver, "_find_available_private_socket"
) as find_available:
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
find_available.assert_not_called()
event_queue.put.assert_called_once_with(
{
"Type": unixDriver.FenrirEventType.remote_incomming,
"data": "command say unknown",
}
)
def test_private_socket_handles_command_locally(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
@@ -281,3 +362,33 @@ def test_find_available_private_socket_skips_main_socket(
driver._find_available_private_socket()
== "/tmp/fenrirscreenreader-111.sock"
)
def test_find_available_private_socket_filters_by_owner(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
with patch(
"fenrirscreenreader.remoteDriver.unixDriver.remoteInstanceRegistry.list_instances",
return_value=[
{
"pid": 111,
"socket_files": ["/tmp/fenrirscreenreader-111.sock"],
},
{
"pid": 222,
"socket_files": ["/tmp/fenrirscreenreader-222.sock"],
},
],
), patch.object(
driver,
"_get_socket_owner_uid",
side_effect=lambda socket_file: {
"/tmp/fenrirscreenreader-111.sock": 1001,
"/tmp/fenrirscreenreader-222.sock": 1000,
}[socket_file],
), patch.object(driver, "_is_socket_active", return_value=True):
assert (
driver._find_available_private_socket(owner_uid=1000)
== "/tmp/fenrirscreenreader-222.sock"
)