From 2e839e0b160277cbb2a5ad69da4dc928c06bd776 Mon Sep 17 00:00:00 2001 From: PTah Date: Sat, 20 Jun 2026 00:36:12 +1000 Subject: [PATCH] fix: SSH sudo without TTY for non-root Linux admin (0.11.1) Probe commands run without sudo; privileged update uses sudo -S. Add SSH connect and Linux admin API tests. --- backend/app/services/ssh_connect.py | 28 +++- backend/app/version.py | 2 +- backend/tests/test_health.py | 4 +- backend/tests/test_linux_admin_settings.py | 122 ++++++++++++--- backend/tests/test_ssh_connect.py | 170 +++++++++++++++++++++ frontend/src/version.ts | 2 +- 6 files changed, 293 insertions(+), 35 deletions(-) create mode 100644 backend/tests/test_ssh_connect.py diff --git a/backend/app/services/ssh_connect.py b/backend/app/services/ssh_connect.py index 40bd240..76c4524 100644 --- a/backend/app/services/ssh_connect.py +++ b/backend/app/services/ssh_connect.py @@ -67,11 +67,25 @@ def _truncate_output(text: str) -> str: return text[:SSH_OUTPUT_MAX_LEN] + "\n… (truncated)" -def _remote_shell_command(user: str, remote_cmd: str) -> str: - safe_cmd = remote_cmd.replace("'", "'\"'\"'") - if user == "root": - return f"bash -lc '{safe_cmd}'" - return f"bash -lc 'sudo -n {safe_cmd} 2>/dev/null || sudo {safe_cmd}'" +def _shell_single_quote(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def _remote_shell_command( + user: str, + remote_cmd: str, + password: str, + *, + need_root: bool = False, +) -> str: + """Build remote shell command. Sudo only when need_root and user is not root.""" + safe_cmd = _shell_single_quote(remote_cmd) + if user == "root" or not need_root: + return f"bash -lc {safe_cmd}" + + safe_pw = _shell_single_quote(password) + inner = f"printf '%s\\n' {safe_pw} | sudo -S -p '' bash -lc {safe_cmd}" + return f"bash -lc {_shell_single_quote(inner)}" def run_ssh_command( @@ -82,6 +96,7 @@ def run_ssh_command( remote_cmd: str, connect_timeout_sec: int = 15, command_timeout_sec: int = 600, + need_root: bool = False, ) -> SshCommandResult: target = target.strip() user = user.strip() @@ -95,7 +110,7 @@ def run_ssh_command( client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - shell_cmd = _remote_shell_command(user, remote_cmd) + shell_cmd = _remote_shell_command(user, remote_cmd, password, need_root=need_root) try: client.connect( @@ -209,4 +224,5 @@ def run_ssh_monitor_update( password=password, remote_cmd=script, command_timeout_sec=900, + need_root=True, ) diff --git a/backend/app/version.py b/backend/app/version.py index 6044ca5..b6ac1de 100644 --- a/backend/app/version.py +++ b/backend/app/version.py @@ -1,5 +1,5 @@ """Единый источник версии SAC (API, health, логи, OpenAPI).""" APP_NAME = "Security Alert Center" -APP_VERSION = "0.11.0" +APP_VERSION = "0.11.1" APP_VERSION_LABEL = f"{APP_NAME} v.{APP_VERSION}" diff --git a/backend/tests/test_health.py b/backend/tests/test_health.py index ad4b7b8..f39377d 100644 --- a/backend/tests/test_health.py +++ b/backend/tests/test_health.py @@ -4,6 +4,6 @@ from app.version import APP_NAME, APP_VERSION, APP_VERSION_LABEL def test_version_constants(): - assert APP_VERSION == "0.11.0" + assert APP_VERSION == "0.11.1" assert APP_NAME == "Security Alert Center" - assert APP_VERSION_LABEL == "Security Alert Center v.0.11.0" + assert APP_VERSION_LABEL == "Security Alert Center v.0.11.1" diff --git a/backend/tests/test_linux_admin_settings.py b/backend/tests/test_linux_admin_settings.py index b0bbf8f..70bcda8 100644 --- a/backend/tests/test_linux_admin_settings.py +++ b/backend/tests/test_linux_admin_settings.py @@ -1,16 +1,68 @@ -"""Linux admin settings and SSH host actions.""" +"""Additional Linux admin API tests.""" from unittest.mock import patch -from app.models import Host -from app.models.ui_settings import UI_SETTINGS_ROW_ID, UiSettings -from app.services.ssh_connect import SshCommandResult, iter_ssh_targets + +def test_get_linux_admin_settings_env_default(jwt_headers, client, monkeypatch): + monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "") + monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "") + from app.config import get_settings + + get_settings.cache_clear() + + response = client.get("/api/v1/settings/linux-admin", headers=jwt_headers) + assert response.status_code == 200 + body = response.json() + assert body["configured"] is False + assert body["source"] == "env" + + +def test_host_ssh_test_requires_admin(jwt_monitor_headers, client): + response = client.post("/api/v1/hosts/1/actions/ssh-test", headers=jwt_monitor_headers) + assert response.status_code == 403 + + +def test_host_ssh_test_not_linux(jwt_headers, client, db_session, monkeypatch): + monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "root") + monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "pw") + from app.config import get_settings + from app.models import Host + + get_settings.cache_clear() + + host = Host(hostname="PC", os_family="windows", product="rdp-login-monitor", ipv4="10.0.0.1") + db_session.add(host) + db_session.commit() + db_session.refresh(host) + + response = client.post(f"/api/v1/hosts/{host.id}/actions/ssh-test", headers=jwt_headers) + assert response.status_code == 400 + assert "not Linux" in response.json()["detail"] + + +def test_host_agent_update_not_configured(jwt_headers, client, db_session, monkeypatch): + monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "") + monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "") + from app.config import get_settings + from app.models import Host + + get_settings.cache_clear() + + host = Host(hostname="ubabuba", os_family="linux", product="ssh-monitor", ipv4="10.0.0.5") + db_session.add(host) + db_session.commit() + db_session.refresh(host) + + response = client.post(f"/api/v1/hosts/{host.id}/actions/agent-update", headers=jwt_headers) + assert response.status_code == 400 + assert "not configured" in response.json()["detail"].lower() def test_put_linux_admin_settings_persists(jwt_headers, client, db_session, monkeypatch): monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "") monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "") from app.config import get_settings + from app.models.ui_settings import UI_SETTINGS_ROW_ID, UiSettings get_settings.cache_clear() @@ -24,6 +76,7 @@ def test_put_linux_admin_settings_persists(jwt_headers, client, db_session, monk assert body["configured"] is True assert body["user"] == "root" assert body["source"] == "db" + assert body["password_hint"] row = db_session.get(UiSettings, UI_SETTINGS_ROW_ID) assert row is not None @@ -31,31 +84,50 @@ def test_put_linux_admin_settings_persists(jwt_headers, client, db_session, monk assert row.linux_admin_password == "secret-pass" -def test_iter_ssh_targets_hostname_before_ip(): - host = Host( - hostname="ubabuba", - os_family="linux", - product="ssh-monitor", - ipv4="10.0.0.5", - ) - targets = iter_ssh_targets(host) - assert targets[0] == "ubabuba" - assert targets[-1] == "10.0.0.5" - - -def test_host_agent_update_via_ssh(jwt_headers, client, db_session, monkeypatch): +def test_host_ssh_test_success(jwt_headers, client, db_session, monkeypatch): monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "root") monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "pw") from app.config import get_settings + from app.models import Host + from app.services.ssh_connect import SshCommandResult get_settings.cache_clear() - host = Host( - hostname="ubabuba", - os_family="linux", - product="ssh-monitor", - ipv4="10.0.0.5", - ) + host = Host(hostname="ubabuba", os_family="linux", product="ssh-monitor", ipv4="10.0.0.5") + db_session.add(host) + db_session.commit() + db_session.refresh(host) + + with patch("app.api.v1.hosts.test_ssh_connection") as mock_test: + mock_test.return_value = SshCommandResult( + ok=True, + message="SSH OK, hostname=ubabuba", + target="ubabuba", + stdout="ubabuba\n", + exit_code=0, + ) + response = client.post( + f"/api/v1/hosts/{host.id}/actions/ssh-test", + headers=jwt_headers, + ) + + assert response.status_code == 200 + body = response.json() + assert body["ok"] is True + assert "hostname=ubabuba" in body["message"] + assert body["target"] == "ubabuba" + + +def test_host_agent_update_success(jwt_headers, client, db_session, monkeypatch): + monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "root") + monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "pw") + from app.config import get_settings + from app.models import Host + from app.services.ssh_connect import SshCommandResult + + get_settings.cache_clear() + + host = Host(hostname="ubabuba", os_family="linux", product="ssh-monitor", ipv4="10.0.0.5") db_session.add(host) db_session.commit() db_session.refresh(host) @@ -63,9 +135,9 @@ def test_host_agent_update_via_ssh(jwt_headers, client, db_session, monkeypatch) with patch("app.api.v1.hosts.run_ssh_monitor_update") as mock_update: mock_update.return_value = SshCommandResult( ok=True, - message="SSH OK", + message="SSH OK (ubabuba), exit 0\nSUMMARY updated", target="ubabuba", - stdout="updated", + stdout="SUMMARY updated\n", exit_code=0, ) response = client.post( diff --git a/backend/tests/test_ssh_connect.py b/backend/tests/test_ssh_connect.py new file mode 100644 index 0000000..fb52705 --- /dev/null +++ b/backend/tests/test_ssh_connect.py @@ -0,0 +1,170 @@ +"""SSH connect service tests (paramiko mocked via sys.modules).""" + +import sys +from unittest.mock import MagicMock + +from app.models import Host +from app.services import ssh_connect +from app.services.ssh_connect import ( + HostNotLinuxError, + HostTargetMissingError, + _remote_shell_command, + iter_ssh_targets, + run_ssh_command, + run_ssh_monitor_update, +) + + +class _AuthError(Exception): + pass + + +def _install_fake_paramiko(monkeypatch, *, connect_side_effect=None, exec_setup=None): + mock_client_cls = MagicMock() + client = MagicMock() + if connect_side_effect is not None: + client.connect.side_effect = connect_side_effect + if exec_setup is not None: + exec_setup(client) + mock_client_cls.return_value = client + + fake = MagicMock() + fake.SSHClient = mock_client_cls + fake.AutoAddPolicy = MagicMock() + fake.AuthenticationException = _AuthError + monkeypatch.setitem(sys.modules, "paramiko", fake) + return client + + +def test_remote_shell_command_non_root_probe_has_no_sudo(): + cmd = _remote_shell_command("deploy", "hostname", "secret", need_root=False) + assert "sudo" not in cmd + assert "hostname" in cmd + + +def test_remote_shell_command_non_root_privileged_uses_sudo_s(): + cmd = _remote_shell_command("deploy", "/opt/scripts/update_ssh_monitor.sh", "secret", need_root=True) + assert "sudo -S" in cmd + assert "secret" in cmd + assert "update_ssh_monitor.sh" in cmd + + +def test_remote_shell_command_root_skips_sudo_even_when_need_root(): + cmd = _remote_shell_command("root", "/opt/scripts/update_ssh_monitor.sh", "secret", need_root=True) + assert "sudo" not in cmd + + +def test_probe_ssh_connection_non_root_does_not_use_sudo(monkeypatch): + captured: list[str] = [] + + def setup(client): + stdout = MagicMock() + stdout.read.return_value = b"ubabuba\n" + stdout.channel.recv_exit_status.return_value = 0 + stderr = MagicMock() + stderr.read.return_value = b"" + + def capture_exec(cmd, **kwargs): + captured.append(cmd) + return (None, stdout, stderr) + + client.exec_command.side_effect = capture_exec + + _install_fake_paramiko(monkeypatch, exec_setup=setup) + + result = ssh_connect.test_ssh_connection(target="10.10.36.9", user="deploy", password="pw") + assert result.ok is True + assert captured + assert "sudo" not in captured[0] + + +def test_iter_ssh_targets_rejects_windows(): + host = Host(hostname="PC", os_family="windows", product="rdp-login-monitor", ipv4="1.2.3.4") + try: + iter_ssh_targets(host) + raise AssertionError("expected HostNotLinuxError") + except HostNotLinuxError: + pass + + +def test_iter_ssh_targets_requires_address(): + host = Host(hostname="", os_family="linux", product="ssh-monitor", ipv4=None) + try: + iter_ssh_targets(host) + raise AssertionError("expected HostTargetMissingError") + except HostTargetMissingError: + pass + + +def test_probe_ssh_connection_success(monkeypatch): + def setup(client): + stdout = MagicMock() + stdout.read.return_value = b"ubabuba\n" + stdout.channel.recv_exit_status.return_value = 0 + stderr = MagicMock() + stderr.read.return_value = b"" + client.exec_command.return_value = (None, stdout, stderr) + + client = _install_fake_paramiko(monkeypatch, exec_setup=setup) + + result = ssh_connect.test_ssh_connection(target="ubabuba", user="root", password="pw") + assert result.ok is True + assert "hostname=ubabuba" in result.message + client.connect.assert_called_once() + client.close.assert_called_once() + + +def test_run_ssh_command_auth_failure(monkeypatch): + client = _install_fake_paramiko(monkeypatch, connect_side_effect=_AuthError("bad creds")) + + result = run_ssh_command(target="10.0.0.1", user="root", password="wrong", remote_cmd="hostname") + assert result.ok is False + assert "auth failed" in result.message.lower() + client.close.assert_called_once() + + +def test_run_ssh_monitor_update_missing_script(monkeypatch): + monkeypatch.setattr( + ssh_connect, + "run_ssh_command", + lambda **kwargs: ssh_connect.SshCommandResult( + ok=True, + message="ok", + target="h", + stdout="missing\n", + ), + ) + + result = run_ssh_monitor_update(target="ubabuba", user="root", password="pw") + assert result.ok is False + assert "not found" in result.message.lower() + + +def test_run_ssh_monitor_update_runs_script(monkeypatch): + calls = 0 + + def fake_run(**kwargs): + nonlocal calls + calls += 1 + if calls == 1: + return ssh_connect.SshCommandResult( + ok=True, + message="ok", + target="h", + stdout="ready\n", + ) + assert kwargs["remote_cmd"] == "/opt/scripts/update_ssh_monitor.sh" + assert kwargs.get("need_root") is True + return ssh_connect.SshCommandResult( + ok=True, + message="SSH OK", + target="ubabuba", + stdout="SUMMARY updated\n", + exit_code=0, + ) + + monkeypatch.setattr(ssh_connect, "run_ssh_command", fake_run) + + result = run_ssh_monitor_update(target="ubabuba", user="root", password="pw") + assert result.ok is True + assert calls == 2 diff --git a/frontend/src/version.ts b/frontend/src/version.ts index 52ae265..4b181c6 100644 --- a/frontend/src/version.ts +++ b/frontend/src/version.ts @@ -1,4 +1,4 @@ /** Fallback до загрузки /health; при релизе держите в sync с backend/app/version.py */ export const APP_NAME = "Security Alert Center"; -export const APP_VERSION = "0.11.0"; +export const APP_VERSION = "0.11.1"; export const APP_VERSION_LABEL = `${APP_NAME} v.${APP_VERSION}`;