Storm Dragon
395 lines
12 KiB
Python
395 lines
12 KiB
Python
import socket
|
|
import struct
|
|
from unittest.mock import Mock, mock_open, patch
|
|
|
|
import pytest
|
|
|
|
from fenrirscreenreader.remoteDriver import unixDriver
|
|
|
|
|
|
def test_watchdog_keeps_serving_when_instance_registration_fails(
|
|
mock_environment,
|
|
):
|
|
driver = unixDriver.driver()
|
|
driver.env = mock_environment
|
|
driver.fenrirSocks = [Mock()]
|
|
active = Mock(value=True)
|
|
event_queue = Mock()
|
|
|
|
def stop_after_select(*_args):
|
|
active.value = False
|
|
return ([], [], [])
|
|
|
|
with patch.object(driver, "_get_socket_candidates", return_value=[]), patch.object(
|
|
driver, "_register_instance", side_effect=PermissionError("denied")
|
|
), patch(
|
|
"fenrirscreenreader.remoteDriver.unixDriver.select.select",
|
|
side_effect=stop_after_select,
|
|
) as select_call:
|
|
driver.watch_dog(active, event_queue)
|
|
|
|
select_call.assert_called_once()
|
|
|
|
|
|
def test_watchdog_closes_listeners_when_unexpected_error_escapes(
|
|
mock_environment,
|
|
):
|
|
driver = unixDriver.driver()
|
|
driver.env = mock_environment
|
|
listener = Mock()
|
|
active = Mock(value=True)
|
|
|
|
with patch.object(
|
|
driver,
|
|
"_get_socket_candidates",
|
|
return_value=[("/tmp/fenrir-test.sock", False)],
|
|
), patch.object(
|
|
driver, "_bind_socket", return_value=listener
|
|
), patch.object(
|
|
driver, "_try_register_instance"
|
|
), patch(
|
|
"fenrirscreenreader.remoteDriver.unixDriver.select.select",
|
|
side_effect=RuntimeError("unexpected failure"),
|
|
), patch(
|
|
"fenrirscreenreader.remoteDriver.unixDriver.os.path.exists",
|
|
return_value=False,
|
|
), patch(
|
|
"fenrirscreenreader.remoteDriver.unixDriver.remoteInstanceRegistry.remove_instance"
|
|
) as remove_instance:
|
|
with pytest.raises(RuntimeError, match="unexpected failure"):
|
|
driver.watch_dog(active, Mock())
|
|
|
|
listener.close.assert_called_once_with()
|
|
assert driver.fenrirSocks == []
|
|
assert driver.bound_sockets == []
|
|
remove_instance.assert_called_once_with()
|
|
|
|
|
|
class FakeClientSocket:
|
|
def __init__(self, data):
|
|
self.data = data
|
|
self.sent = b""
|
|
|
|
def recv(self, _size):
|
|
return self.data
|
|
|
|
def sendall(self, data):
|
|
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
|
|
driver.bound_sockets = [
|
|
(Mock(), "/tmp/fenrirscreenreader-999.sock"),
|
|
(Mock(), unixDriver.MAIN_SOCKET_FILE),
|
|
]
|
|
client_sock = FakeClientSocket(b"command say routed")
|
|
event_queue = Mock()
|
|
|
|
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
|
|
driver,
|
|
"_find_ancestor_private_socket",
|
|
return_value="/tmp/fenrirscreenreader-222.sock",
|
|
), patch.object(
|
|
driver,
|
|
"_find_available_private_socket",
|
|
return_value="/tmp/fenrirscreenreader-222.sock",
|
|
), patch.object(
|
|
driver, "_forward_remote_to_socket", return_value=True
|
|
) as forward:
|
|
driver._handle_client(
|
|
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
|
|
)
|
|
|
|
forward.assert_called_once_with(
|
|
"command say routed", "/tmp/fenrirscreenreader-222.sock"
|
|
)
|
|
event_queue.put.assert_not_called()
|
|
|
|
|
|
def test_main_socket_routes_to_first_available_private_socket(
|
|
mock_environment,
|
|
):
|
|
driver = unixDriver.driver()
|
|
driver.env = mock_environment
|
|
driver.bound_sockets = [
|
|
(Mock(), "/tmp/fenrirscreenreader-999.sock"),
|
|
(Mock(), unixDriver.MAIN_SOCKET_FILE),
|
|
]
|
|
client_sock = FakeClientSocket(b"command say fallback")
|
|
event_queue = Mock()
|
|
|
|
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
|
|
driver,
|
|
"_find_ancestor_private_socket",
|
|
return_value="",
|
|
), patch.object(
|
|
driver,
|
|
"_find_available_private_socket",
|
|
return_value="/tmp/fenrirscreenreader-333.sock",
|
|
), patch.object(
|
|
driver, "_forward_remote_to_socket", return_value=True
|
|
) as forward:
|
|
driver._handle_client(
|
|
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
|
|
)
|
|
|
|
forward.assert_called_once_with(
|
|
"command say fallback", "/tmp/fenrirscreenreader-333.sock"
|
|
)
|
|
event_queue.put.assert_not_called()
|
|
|
|
|
|
def test_main_socket_handles_command_locally_without_available_target(
|
|
mock_environment,
|
|
):
|
|
driver = unixDriver.driver()
|
|
driver.env = mock_environment
|
|
client_sock = FakeClientSocket(b"command say local")
|
|
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="",
|
|
):
|
|
driver._handle_client(
|
|
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
|
|
)
|
|
|
|
event_queue.put.assert_called_once_with(
|
|
{
|
|
"Type": unixDriver.FenrirEventType.remote_incomming,
|
|
"data": "command say local",
|
|
}
|
|
)
|
|
|
|
|
|
def test_vcsa_main_socket_owner_handles_command_locally(mock_environment):
|
|
driver = unixDriver.driver()
|
|
driver.env = mock_environment
|
|
driver.bound_sockets = [(Mock(), unixDriver.MAIN_SOCKET_FILE)]
|
|
client_sock = FakeClientSocket(b"command say root")
|
|
event_queue = Mock()
|
|
|
|
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
|
|
)
|
|
|
|
find_available.assert_not_called()
|
|
event_queue.put.assert_called_once_with(
|
|
{
|
|
"Type": unixDriver.FenrirEventType.remote_incomming,
|
|
"data": "command say root",
|
|
}
|
|
)
|
|
|
|
|
|
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
|
|
client_sock = FakeClientSocket(b"command say private")
|
|
event_queue = Mock()
|
|
|
|
driver._handle_client(
|
|
client_sock, event_queue, "/tmp/fenrirscreenreader-222.sock"
|
|
)
|
|
|
|
event_queue.put.assert_called_once_with(
|
|
{
|
|
"Type": unixDriver.FenrirEventType.remote_incomming,
|
|
"data": "command say private",
|
|
}
|
|
)
|
|
|
|
|
|
def test_list_command_still_returns_response_from_main_socket(mock_environment):
|
|
driver = unixDriver.driver()
|
|
driver.env = mock_environment
|
|
mock_environment["runtime"]["RemoteManager"] = Mock()
|
|
client_sock = FakeClientSocket(b"command list")
|
|
event_queue = Mock()
|
|
mock_environment["runtime"][
|
|
"RemoteManager"
|
|
].handle_remote_incomming_with_response.return_value = {
|
|
"success": True,
|
|
"message": "pid=1",
|
|
}
|
|
|
|
driver._handle_client(client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE)
|
|
|
|
assert client_sock.sent == b"pid=1\n"
|
|
event_queue.put.assert_not_called()
|
|
|
|
|
|
def test_get_parent_pid_parses_process_names_with_spaces():
|
|
driver = unixDriver.driver()
|
|
|
|
stat_file = mock_open(
|
|
read_data="1234 (name with spaces) S 567 1 1 0 -1 0\n"
|
|
)
|
|
with patch("builtins.open", stat_file):
|
|
assert driver._get_parent_pid(1234) == 567
|
|
|
|
|
|
def test_find_available_private_socket_prefers_ancestor_socket(
|
|
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, "_is_socket_active", return_value=True):
|
|
assert (
|
|
driver._find_available_private_socket(
|
|
"/tmp/fenrirscreenreader-222.sock"
|
|
)
|
|
== "/tmp/fenrirscreenreader-222.sock"
|
|
)
|
|
|
|
|
|
def test_find_available_private_socket_skips_main_socket(
|
|
mock_environment,
|
|
):
|
|
driver = unixDriver.driver()
|
|
driver.env = mock_environment
|
|
|
|
with patch(
|
|
"fenrirscreenreader.remoteDriver.unixDriver.remoteInstanceRegistry.list_instances",
|
|
return_value=[
|
|
{
|
|
"pid": 111,
|
|
"socket_files": [
|
|
unixDriver.MAIN_SOCKET_FILE,
|
|
"/tmp/fenrirscreenreader-111.sock",
|
|
],
|
|
}
|
|
],
|
|
), patch.object(driver, "_is_socket_active", return_value=True):
|
|
assert (
|
|
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"
|
|
)
|