diff --git a/backend/app/services/host_sessions.py b/backend/app/services/host_sessions.py index f476108..509a1f5 100644 --- a/backend/app/services/host_sessions.py +++ b/backend/app/services/host_sessions.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import re from dataclasses import dataclass @@ -49,6 +50,8 @@ _LOGIND_STATES = frozenset( "unknown", } ) +_LOGIND_TTY_RE = re.compile(r"^(pts|tty)/", re.IGNORECASE) +_NO_SESSION_MARKERS = ("no session", "unknown session", "does not exist") @dataclass(frozen=True) @@ -93,6 +96,86 @@ def is_windows_host_event(event: Event) -> bool: return host is not None and is_windows_host(host) +def _normalize_logind_tty(value: object) -> str | None: + if value is None: + return None + text = str(value).strip() + if not text or text == "-": + return None + return text + + +def _logind_row_priority(row: HostSessionRow) -> tuple[int, int]: + """Prefer interactive TTY sessions over ephemeral scope/SSH-exec rows (tty «-»).""" + has_tty = 1 if row.tty and _LOGIND_TTY_RE.match(row.tty) else 0 + try: + sid_num = int(row.session_id) + except ValueError: + sid_num = 0 + return (has_tty, sid_num) + + +def filter_logind_session_rows(rows: list[HostSessionRow]) -> list[HostSessionRow]: + """Drop no-TTY duplicates when the same user already has a pts/tty session.""" + if len(rows) < 2: + return rows + + by_user: dict[str, list[HostSessionRow]] = {} + for row in rows: + key = row.user.casefold() + by_user.setdefault(key, []).append(row) + + out: list[HostSessionRow] = [] + for group in by_user.values(): + if len(group) == 1: + out.append(group[0]) + continue + with_tty = [r for r in group if r.tty and _LOGIND_TTY_RE.match(r.tty)] + if with_tty: + out.append(max(with_tty, key=_logind_row_priority)) + continue + out.append(max(group, key=_logind_row_priority)) + out.sort(key=lambda r: (r.user.casefold(), -_logind_row_priority(r)[1])) + return out + + +def parse_loginctl_sessions_json(stdout: str) -> list[HostSessionRow]: + text = stdout.strip() + if not text: + return [] + try: + payload = json.loads(text) + except json.JSONDecodeError: + return [] + if not isinstance(payload, list): + return [] + + rows: list[HostSessionRow] = [] + for item in payload: + if not isinstance(item, dict): + continue + sid = str(item.get("session") or "").strip() + user = str(item.get("user") or "").strip() + if not sid or not user: + continue + if not _LOGIND_SESSION_ID_RE.match(sid): + continue + if not _LOGIND_USER_RE.match(user): + continue + state = str(item.get("state") or "").strip().lower() or None + if state and state not in _LOGIND_STATES: + continue + rows.append( + HostSessionRow( + session_id=sid, + user=user, + tty=_normalize_logind_tty(item.get("tty")), + state=state, + ) + ) + return filter_logind_session_rows(rows) + + def _looks_like_loginctl_row(parts: list[str]) -> bool: if len(parts) < 6: return False @@ -122,11 +205,134 @@ def parse_loginctl_sessions(stdout: str) -> list[HostSessionRow]: HostSessionRow( session_id=sid, user=user, - tty=tty, + tty=_normalize_logind_tty(tty), state=state, ) ) - return rows + return filter_logind_session_rows(rows) + + +def _shell_quote(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def _linux_session_exists_message(stderr: str, stdout: str) -> bool: + blob = f"{stderr}\n{stdout}".casefold() + return any(marker in blob for marker in _NO_SESSION_MARKERS) + + +def _linux_show_session_user(host: Host, cfg: LinuxAdminConfig, session_id: str) -> str | None: + sid = session_id.strip() + if not sid: + return None + remote_cmd = f"loginctl show-session {_shell_quote(sid)} -p Name --value" + targets = iter_ssh_targets(host) + for target in targets: + result = run_ssh_command( + target=target, + user=cfg.user, + password=cfg.password, + remote_cmd=remote_cmd, + connect_timeout_sec=15, + command_timeout_sec=30, + need_root=True, + login_shell=False, + ) + if not result.ok: + continue + user = result.stdout.strip().splitlines()[-1].strip() if result.stdout.strip() else "" + if user and _LOGIND_USER_RE.match(user): + return user + return None + + +def list_linux_sessions(host: Host, cfg: LinuxAdminConfig) -> tuple[list[HostSessionRow], SshCommandResult]: + if not is_linux_host(host): + raise SshHostNotLinuxError("Host is not Linux") + targets = iter_ssh_targets(host) + json_cmd = "loginctl list-sessions --output=json --no-legend" + text_cmd = "loginctl list-sessions --no-legend --no-pager" + last: SshCommandResult | None = None + for target in targets: + result = run_ssh_command( + target=target, + user=cfg.user, + password=cfg.password, + remote_cmd=json_cmd, + connect_timeout_sec=15, + command_timeout_sec=45, + need_root=True, + login_shell=False, + ) + last = result + if result.ok and result.stdout.strip(): + rows = parse_loginctl_sessions_json(result.stdout) + if rows: + return rows, result + rows = parse_loginctl_sessions(result.stdout) + if rows: + return rows, result + result = run_ssh_command( + target=target, + user=cfg.user, + password=cfg.password, + remote_cmd=text_cmd, + connect_timeout_sec=15, + command_timeout_sec=45, + need_root=True, + login_shell=False, + ) + last = result + if result.ok and result.stdout.strip(): + return parse_loginctl_sessions(result.stdout), result + assert last is not None + return [], last + + +def terminate_linux_session( + host: Host, + cfg: LinuxAdminConfig, + session_id: str, +) -> SshCommandResult: + sid = session_id.strip() + if not sid: + return SshCommandResult(ok=False, message="session_id is required", target="") + if not is_linux_host(host): + raise SshHostNotLinuxError("Host is not Linux") + targets = iter_ssh_targets(host) + remote_cmd = f"loginctl terminate-session {_shell_quote(sid)}" + last: SshCommandResult | None = None + session_user = _linux_show_session_user(host, cfg, sid) + for target in targets: + result = run_ssh_command( + target=target, + user=cfg.user, + password=cfg.password, + remote_cmd=remote_cmd, + connect_timeout_sec=15, + command_timeout_sec=45, + need_root=True, + login_shell=False, + ) + last = result + if result.ok: + return result + if session_user: + return _terminate_linux_user_sessions(host, cfg, session_user) + if last is not None and _linux_session_exists_message(last.stderr, last.stdout): + return SshCommandResult( + ok=False, + message=( + f"Сессия {sid} не найдена (устарела). Обновите список и повторите " + "или завершите все сессии пользователя." + ), + target=last.target, + stdout=last.stdout, + stderr=last.stderr, + exit_code=last.exit_code, + ) + assert last is not None + return last def parse_qwinsta_sessions(stdout: str, *, filter_user: str | None = None) -> list[HostSessionRow]: @@ -167,65 +373,6 @@ def parse_qwinsta_sessions(stdout: str, *, filter_user: str | None = None) -> li return parse_qwinsta_sessions(stdout, filter_user=None) -def list_linux_sessions(host: Host, cfg: LinuxAdminConfig) -> tuple[list[HostSessionRow], SshCommandResult]: - if not is_linux_host(host): - raise SshHostNotLinuxError("Host is not Linux") - targets = iter_ssh_targets(host) - cmd = "loginctl list-sessions --no-legend --no-pager" - last: SshCommandResult | None = None - for target in targets: - result = run_ssh_command( - target=target, - user=cfg.user, - password=cfg.password, - remote_cmd=cmd, - connect_timeout_sec=15, - command_timeout_sec=45, - need_root=True, - login_shell=False, - ) - last = result - if result.ok and result.stdout.strip(): - return parse_loginctl_sessions(result.stdout), result - assert last is not None - return [], last - - -def terminate_linux_session( - host: Host, - cfg: LinuxAdminConfig, - session_id: str, -) -> SshCommandResult: - sid = session_id.strip() - if not sid: - return SshCommandResult(ok=False, message="session_id is required", target="") - if not is_linux_host(host): - raise SshHostNotLinuxError("Host is not Linux") - targets = iter_ssh_targets(host) - remote_cmd = f"loginctl terminate-session {_shell_quote(sid)}" - last: SshCommandResult | None = None - for target in targets: - result = run_ssh_command( - target=target, - user=cfg.user, - password=cfg.password, - remote_cmd=remote_cmd, - connect_timeout_sec=15, - command_timeout_sec=45, - need_root=True, - login_shell=False, - ) - last = result - if result.ok: - return result - assert last is not None - return last - - -def _shell_quote(value: str) -> str: - return "'" + value.replace("'", "'\"'\"'") + "'" - - def list_windows_sessions(host: Host, cfg: WinAdminConfig) -> tuple[list[HostSessionRow], WinRmCmdResult | None]: if not is_windows_host(host): raise HostNotWindowsError("Host is not Windows") diff --git a/backend/app/version.py b/backend/app/version.py index 977bc27..bd61d58 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.20.33" +APP_VERSION = "0.20.34" APP_VERSION_LABEL = f"{APP_NAME} v.{APP_VERSION}" diff --git a/backend/tests/test_health.py b/backend/tests/test_health.py index c0e406a..eaf452f 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.20.33" + assert APP_VERSION == "0.20.34" assert APP_NAME == "Security Alert Center" - assert APP_VERSION_LABEL == "Security Alert Center v.0.20.33" + assert APP_VERSION_LABEL == "Security Alert Center v.0.20.34" diff --git a/backend/tests/test_host_sessions.py b/backend/tests/test_host_sessions.py index 6f19355..551a83d 100644 --- a/backend/tests/test_host_sessions.py +++ b/backend/tests/test_host_sessions.py @@ -2,7 +2,9 @@ from app.services.host_sessions import ( event_supports_session_terminate, + filter_logind_session_rows, parse_loginctl_sessions, + parse_loginctl_sessions_json, parse_qwinsta_sessions, ) @@ -16,18 +18,37 @@ 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 - +def test_parse_loginctl_sessions_prefers_pts_over_ephemeral_duplicate(): + stdout = """21166 1000 papatramp - pts/0 active no - +21169 1000 papatramp - - active no - """ rows = parse_loginctl_sessions(stdout) + assert len(rows) == 1 + assert rows[0].session_id == "21166" + assert rows[0].tty == "pts/0" + + +def test_parse_loginctl_sessions_json(): + stdout = """[ + {"session":"21166","uid":1000,"user":"papatramp","seat":null,"tty":"pts/0","state":"active","idle":false,"since":null}, + {"session":"21169","uid":1000,"user":"papatramp","seat":null,"tty":null,"state":"active","idle":false,"since":null} +]""" + rows = parse_loginctl_sessions_json(stdout) + assert len(rows) == 1 + assert rows[0].session_id == "21166" + assert rows[0].tty == "pts/0" + + +def test_filter_logind_session_rows_keeps_distinct_users(): + from app.services.host_sessions import HostSessionRow + + rows = filter_logind_session_rows( + [ + HostSessionRow(session_id="1", user="alice", tty="pts/0", state="active"), + HostSessionRow(session_id="2", user="bob", tty=None, state="active"), + ] + ) 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(): diff --git a/frontend/src/version.ts b/frontend/src/version.ts index efcc304..db2a2f8 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.20.33"; +export const APP_VERSION = "0.20.34"; export const APP_VERSION_LABEL = `${APP_NAME} v.${APP_VERSION}`;