PTah
68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
"""Actor user extraction for event list UI."""
|
|
|
|
import uuid
|
|
|
|
from app.services.event_actor_user import extract_event_actor_user
|
|
|
|
|
|
def test_ssh_login_user():
|
|
assert extract_event_actor_user(
|
|
"ssh.login.success",
|
|
{"user": "deploy", "source_ip": "10.0.0.1"},
|
|
) == "deploy"
|
|
|
|
|
|
def test_rdp_login_username_fallback():
|
|
assert extract_event_actor_user(
|
|
"rdp.login.failed",
|
|
{"username": "DOMAIN\\admin"},
|
|
) == "DOMAIN\\admin"
|
|
|
|
|
|
def test_sudo_user():
|
|
assert extract_event_actor_user(
|
|
"privilege.sudo.command",
|
|
{"user": "root", "command": "apt update"},
|
|
) == "root"
|
|
|
|
|
|
def test_shadow_shadower_user():
|
|
assert extract_event_actor_user(
|
|
"rdp.shadow.control.started",
|
|
{"shadower_user": "admin", "target_user": "user1"},
|
|
) == "admin"
|
|
|
|
|
|
def test_heartbeat_no_user():
|
|
assert extract_event_actor_user("agent.heartbeat", {"user": "ignored"}) is None
|
|
|
|
|
|
def test_auth_event_missing_user():
|
|
assert extract_event_actor_user("ssh.login.failed", {}) is None
|
|
|
|
|
|
def test_events_list_includes_actor_user(client, auth_headers, jwt_headers):
|
|
event_id = str(uuid.uuid4())
|
|
payload = {
|
|
"schema_version": "1.0",
|
|
"event_id": event_id,
|
|
"occurred_at": "2026-05-27T10:00:00+03:00",
|
|
"source": {"product": "ssh-monitor", "product_version": "1.2.3-SAC"},
|
|
"host": {"hostname": "actor-host", "os_family": "linux"},
|
|
"category": "auth",
|
|
"type": "ssh.login.success",
|
|
"severity": "info",
|
|
"title": "SSH ok",
|
|
"summary": "login",
|
|
"details": {"user": "alice", "source_ip": "192.0.2.1"},
|
|
}
|
|
assert client.post("/api/v1/events", json=payload, headers=auth_headers).status_code == 201
|
|
|
|
listed = client.get("/api/v1/events?type=ssh.login.success&hostname=actor", headers=jwt_headers)
|
|
assert listed.status_code == 200
|
|
items = listed.json()["items"]
|
|
assert any(i["actor_user"] == "alice" for i in items)
|
|
|
|
dash = client.get("/api/v1/dashboards/recent-events?limit=5", headers=jwt_headers)
|
|
assert any(i.get("actor_user") == "alice" for i in dash.json())
|