161 lines
5.3 KiB
Python
161 lines
5.3 KiB
Python
import os
|
|
import unittest
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
from PyQt5.QtWidgets import QApplication, QInputDialog, QMessageBox
|
|
|
|
from pgu4_sync import (
|
|
ConnectionProfile,
|
|
ListWorker,
|
|
MainWindow,
|
|
is_protected_root_path,
|
|
)
|
|
|
|
|
|
class FakeSftp:
|
|
def __init__(self, resolved_path):
|
|
self.resolved_path = resolved_path
|
|
self.listdir_calls = []
|
|
|
|
def normalize(self, _path):
|
|
return self.resolved_path
|
|
|
|
def listdir_attr(self, path):
|
|
self.listdir_calls.append(path)
|
|
return []
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, sftp):
|
|
self.sftp = sftp
|
|
|
|
def open_sftp(self):
|
|
return self.sftp
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
|
|
class FakeFactory:
|
|
def __init__(self, sftp):
|
|
self.sftp = sftp
|
|
|
|
def open_client(self):
|
|
return FakeClient(self.sftp)
|
|
|
|
|
|
class RootAccessTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def test_root_and_children_are_protected(self):
|
|
self.assertTrue(is_protected_root_path("/root"))
|
|
self.assertTrue(is_protected_root_path("/root/"))
|
|
self.assertTrue(is_protected_root_path("/root/private/data"))
|
|
self.assertFalse(is_protected_root_path("/"))
|
|
self.assertFalse(is_protected_root_path("/rooted"))
|
|
self.assertFalse(is_protected_root_path("/home/root"))
|
|
|
|
def test_symlink_resolving_to_root_requires_auth_before_listing(self):
|
|
sftp = FakeSftp("/root")
|
|
worker = ListWorker(FakeFactory(sftp), "/home/root-link", allow_root=False)
|
|
required = []
|
|
listed = []
|
|
failed = []
|
|
worker.root_auth_required.connect(
|
|
lambda requested, resolved: required.append((requested, resolved))
|
|
)
|
|
worker.listed.connect(lambda path, entries: listed.append((path, entries)))
|
|
worker.failed.connect(failed.append)
|
|
|
|
worker.run()
|
|
|
|
self.assertEqual([("/home/root-link", "/root")], required)
|
|
self.assertEqual([], listed)
|
|
self.assertEqual([], failed)
|
|
self.assertEqual([], sftp.listdir_calls)
|
|
|
|
def test_authorized_worker_lists_resolved_root(self):
|
|
sftp = FakeSftp("/root/private")
|
|
worker = ListWorker(FakeFactory(sftp), "/home/root-link", allow_root=True)
|
|
listed = []
|
|
worker.listed.connect(lambda path, entries: listed.append((path, entries)))
|
|
|
|
worker.run()
|
|
|
|
self.assertEqual([("/root/private", [])], listed)
|
|
self.assertEqual(["/root/private"], sftp.listdir_calls)
|
|
|
|
def test_second_password_must_match_current_connection(self):
|
|
window = MainWindow()
|
|
window.factory.set_profile(
|
|
ConnectionProfile("192.0.2.1", 22, "root", "correct-secret")
|
|
)
|
|
original_get_text = QInputDialog.getText
|
|
original_warning = QMessageBox.warning
|
|
warnings = []
|
|
QMessageBox.warning = lambda *args, **kwargs: warnings.append(args) or QMessageBox.Ok
|
|
try:
|
|
QInputDialog.getText = lambda *args, **kwargs: ("wrong-secret", True)
|
|
self.assertFalse(window.request_root_access("/root"))
|
|
self.assertFalse(window._root_access_authorized)
|
|
self.assertEqual(1, len(warnings))
|
|
|
|
QInputDialog.getText = lambda *args, **kwargs: ("correct-secret", True)
|
|
self.assertTrue(window.request_root_access("/root"))
|
|
self.assertTrue(window._root_access_authorized)
|
|
finally:
|
|
QInputDialog.getText = original_get_text
|
|
QMessageBox.warning = original_warning
|
|
window.close()
|
|
self.app.processEvents()
|
|
|
|
def test_cancelled_second_password_denies_access(self):
|
|
window = MainWindow()
|
|
window.factory.set_profile(
|
|
ConnectionProfile("192.0.2.1", 22, "root", "correct-secret")
|
|
)
|
|
original_get_text = QInputDialog.getText
|
|
try:
|
|
QInputDialog.getText = lambda *args, **kwargs: ("", False)
|
|
self.assertFalse(window.request_root_access("/root"))
|
|
self.assertFalse(window._root_access_authorized)
|
|
finally:
|
|
QInputDialog.getText = original_get_text
|
|
window.close()
|
|
self.app.processEvents()
|
|
|
|
def test_denied_direct_root_path_does_not_start_list_worker(self):
|
|
window = MainWindow()
|
|
window.current_remote_path = "/home"
|
|
window.remote_path_edit.setText("/root")
|
|
window.factory.set_profile(
|
|
ConnectionProfile("192.0.2.1", 22, "root", "correct-secret")
|
|
)
|
|
original_get_text = QInputDialog.getText
|
|
original_warning = QMessageBox.warning
|
|
QMessageBox.warning = lambda *args, **kwargs: QMessageBox.Ok
|
|
try:
|
|
QInputDialog.getText = lambda *args, **kwargs: ("wrong-secret", True)
|
|
thread_count = len(window._threads)
|
|
|
|
window.load_remote("/root")
|
|
|
|
self.assertEqual(thread_count, len(window._threads))
|
|
self.assertEqual("/home", window.remote_path_edit.text())
|
|
self.assertFalse(window._root_access_authorized)
|
|
finally:
|
|
QInputDialog.getText = original_get_text
|
|
QMessageBox.warning = original_warning
|
|
window.close()
|
|
self.app.processEvents()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|