112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
"""API tests for RDG qwinsta/logoff via WinRM on client workstation."""
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import inspect as sa_inspect
|
|
|
|
from app.models import Event, Host
|
|
from app.services.winrm_connect import WinRmCmdResult
|
|
|
|
|
|
def _flap_event(db_session, *, gw: Host, ws: Host) -> Event:
|
|
event = Event(
|
|
event_id="ev-flap-1",
|
|
host_id=gw.id,
|
|
occurred_at=datetime.now(timezone.utc),
|
|
received_at=datetime.now(timezone.utc),
|
|
category="auth",
|
|
type="rdg.connection.disconnected",
|
|
severity="info",
|
|
title="RD Gateway event 303",
|
|
summary="",
|
|
payload={},
|
|
details={
|
|
"user": r"B26\papatramp",
|
|
"internal_ip": ws.ipv4,
|
|
"rdg_flap": True,
|
|
},
|
|
)
|
|
db_session.add(event)
|
|
db_session.commit()
|
|
db_session.refresh(event)
|
|
return event
|
|
|
|
|
|
def test_qwinsta_via_winrm_on_client_host(jwt_headers, client, db_session, monkeypatch):
|
|
monkeypatch.setenv("SAC_WIN_ADMIN_USER", r"B26\admin")
|
|
monkeypatch.setenv("SAC_WIN_ADMIN_PASSWORD", "pw")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
ws = Host(
|
|
hostname="Andrisonova-PC",
|
|
os_family="windows",
|
|
product="rdp-login-monitor",
|
|
ipv4="192.168.160.113",
|
|
)
|
|
gw = Host(hostname="K6A-DC3", os_family="windows", product="rdp-login-monitor", ipv4="192.168.160.40")
|
|
db_session.add_all([ws, gw])
|
|
db_session.commit()
|
|
ws_id = ws.id
|
|
event = _flap_event(db_session, gw=gw, ws=ws)
|
|
|
|
qwinsta_out = " SESSIONNAME USERNAME ID STATE\r\n rdp-tcp#0 B26\\papatramp 2 Active\r\n"
|
|
|
|
with patch("app.services.rdg_winrm_actions.run_winrm_on_host_targets") as mock_run:
|
|
mock_run.return_value = (
|
|
WinRmCmdResult(
|
|
ok=True,
|
|
message="WinRM OK (Andrisonova-PC), exit 0",
|
|
target="Andrisonova-PC",
|
|
stdout=qwinsta_out,
|
|
exit_code=0,
|
|
),
|
|
["Andrisonova-PC=OK"],
|
|
)
|
|
response = client.post(f"/api/v1/events/{event.id}/actions/qwinsta", headers=jwt_headers)
|
|
mock_run.assert_called_once()
|
|
called_identity = sa_inspect(mock_run.call_args.args[0]).identity
|
|
assert called_identity is not None
|
|
assert called_identity[0] == ws_id
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["status"] == "completed"
|
|
assert body["client_hostname"] == "Andrisonova-PC"
|
|
assert body["target"] == "Andrisonova-PC"
|
|
assert "papatramp" in (body["result_stdout"] or "")
|
|
|
|
|
|
def test_qwinsta_client_not_in_hosts(jwt_headers, client, db_session, monkeypatch):
|
|
monkeypatch.setenv("SAC_WIN_ADMIN_USER", r"B26\admin")
|
|
monkeypatch.setenv("SAC_WIN_ADMIN_PASSWORD", "pw")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
gw = Host(hostname="K6A-DC3", os_family="windows", product="rdp-login-monitor", ipv4="192.168.160.40")
|
|
db_session.add(gw)
|
|
db_session.commit()
|
|
|
|
event = Event(
|
|
event_id="ev-flap-2",
|
|
host_id=gw.id,
|
|
occurred_at=datetime.now(timezone.utc),
|
|
received_at=datetime.now(timezone.utc),
|
|
category="auth",
|
|
type="rdg.connection.disconnected",
|
|
severity="info",
|
|
title="303",
|
|
summary="",
|
|
payload={},
|
|
details={"user": r"B26\user", "internal_ip": "192.168.160.999", "rdg_flap": True},
|
|
)
|
|
db_session.add(event)
|
|
db_session.commit()
|
|
|
|
response = client.post(f"/api/v1/events/{event.id}/actions/qwinsta", headers=jwt_headers)
|
|
assert response.status_code == 404
|
|
assert "not found" in response.json()["detail"].lower()
|