fix(backend): non-login SSH and strict loginctl parse, SSH retry (0.20.31)
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -4,6 +4,6 @@ from app.version import APP_NAME, APP_VERSION, APP_VERSION_LABEL
|
||||
|
||||
|
||||
def test_version_constants():
|
||||
assert APP_VERSION == "0.20.30"
|
||||
assert APP_VERSION == "0.20.31"
|
||||
assert APP_NAME == "Security Alert Center"
|
||||
assert APP_VERSION_LABEL == "Security Alert Center v.0.20.30"
|
||||
assert APP_VERSION_LABEL == "Security Alert Center v.0.20.31"
|
||||
|
||||
@@ -16,6 +16,20 @@ def test_parse_loginctl_sessions():
|
||||
assert rows[0].tty == "pts/0"
|
||||
|
||||
|
||||
def test_parse_loginctl_sessions_ignores_motd_noise():
|
||||
stdout = """This server is powered by FASTPANEL
|
||||
Ubuntu Operating LTS
|
||||
configuration By can be not
|
||||
11090 1000 papatramp - - active -
|
||||
11102 1000 papatramp - - active -
|
||||
"""
|
||||
rows = parse_loginctl_sessions(stdout)
|
||||
assert len(rows) == 2
|
||||
assert rows[0].session_id == "11090"
|
||||
assert rows[1].session_id == "11102"
|
||||
assert all(row.user == "papatramp" for row in rows)
|
||||
|
||||
|
||||
def test_parse_qwinsta_sessions_filters_user():
|
||||
stdout = """SESSIONNAME USERNAME ID STATE
|
||||
console Administrator 1 Active
|
||||
|
||||
@@ -36,6 +36,16 @@ def _install_fake_paramiko(monkeypatch, *, connect_side_effect=None, exec_setup=
|
||||
return client
|
||||
|
||||
|
||||
def test_ssh_output_hostname_ignores_motd():
|
||||
stdout = "Welcome!\nFASTPANEL\n\ncz-server\n"
|
||||
assert ssh_connect._ssh_output_hostname(stdout) == "cz-server"
|
||||
|
||||
|
||||
def test_remote_shell_command_non_login_for_sessions():
|
||||
cmd = _remote_shell_command("root", "loginctl list-sessions", "secret", login_shell=False)
|
||||
assert cmd == "bash -c 'loginctl list-sessions'"
|
||||
|
||||
|
||||
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
|
||||
@@ -125,6 +135,53 @@ def test_iter_ssh_targets_requires_address():
|
||||
pass
|
||||
|
||||
|
||||
def test_run_ssh_command_retries_transient_no_existing_session(monkeypatch):
|
||||
attempts = {"count": 0}
|
||||
|
||||
class _NoSessionError(Exception):
|
||||
pass
|
||||
|
||||
def connect_side_effect(*args, **kwargs):
|
||||
attempts["count"] += 1
|
||||
if attempts["count"] == 1:
|
||||
raise _NoSessionError("No existing session")
|
||||
|
||||
def setup(client):
|
||||
stdout = MagicMock()
|
||||
stdout.read.return_value = b"ready\n"
|
||||
stdout.channel.recv_exit_status.return_value = 0
|
||||
stderr = MagicMock()
|
||||
stderr.read.return_value = b""
|
||||
|
||||
def exec_ok(cmd, **kwargs):
|
||||
return (None, stdout, stderr)
|
||||
|
||||
client.exec_command.side_effect = exec_ok
|
||||
|
||||
fake = MagicMock()
|
||||
fake.SSHClient = MagicMock()
|
||||
fake.AutoAddPolicy = MagicMock()
|
||||
fake.AuthenticationException = _AuthError
|
||||
|
||||
def make_client():
|
||||
client = MagicMock()
|
||||
client.connect.side_effect = connect_side_effect
|
||||
setup(client)
|
||||
return client
|
||||
|
||||
fake.SSHClient.side_effect = make_client
|
||||
monkeypatch.setitem(sys.modules, "paramiko", fake)
|
||||
|
||||
result = run_ssh_command(
|
||||
target="185.87.149.9",
|
||||
user="root",
|
||||
password="pw",
|
||||
remote_cmd="hostname",
|
||||
)
|
||||
assert result.ok is True
|
||||
assert attempts["count"] == 2
|
||||
|
||||
|
||||
def test_probe_ssh_connection_success(monkeypatch):
|
||||
def setup(client):
|
||||
stdout = MagicMock()
|
||||
|
||||
Reference in New Issue
Block a user