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.
This commit is contained in:
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user