693 lines
21 KiB
Python
693 lines
21 KiB
Python
"""WinRM connectivity test for Windows hosts using domain admin credentials."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import socket
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from app.config import get_settings
|
|
from app.models import Host
|
|
from app.services.agent_git_release import _cache_dir, clone_or_update_repo
|
|
from app.services.rdp_bundle_delivery import build_rdp_bundle_zip, register_rdp_bundle_zip
|
|
|
|
_CLIXML_PROGRESS_NOISE = frozenset(
|
|
{
|
|
"Preparing modules for first use.",
|
|
"Preparing modules fo",
|
|
}
|
|
)
|
|
_CLIXML_MARKER = "#< CLIXML"
|
|
_CYRILLIC_RE = re.compile(r"[\u0400-\u04FF]")
|
|
RDP_REMOTE_STAGING_DIRNAME = "sac-rdp-staging"
|
|
RDP_LEGACY_STAGING = r"C:\ProgramData\RDP-login-monitor\_sac_staging"
|
|
RDP_BUNDLE_FILES = (
|
|
"Login_Monitor.ps1",
|
|
"Sac-Client.ps1",
|
|
"RdpMonitor-TaskQuery.ps1",
|
|
"version.txt",
|
|
"Deploy-LoginMonitor.ps1",
|
|
"Restart-RdpLoginMonitor.ps1",
|
|
"login_monitor.settings.example.ps1",
|
|
)
|
|
RDP_BUNDLE_REQUIRED = frozenset(
|
|
{
|
|
"Login_Monitor.ps1",
|
|
"Sac-Client.ps1",
|
|
"version.txt",
|
|
"Deploy-LoginMonitor.ps1",
|
|
}
|
|
)
|
|
|
|
|
|
class WinAdminNotConfiguredError(Exception):
|
|
pass
|
|
|
|
|
|
class HostNotWindowsError(Exception):
|
|
pass
|
|
|
|
|
|
class HostTargetMissingError(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WinRmTestResult:
|
|
ok: bool
|
|
message: str
|
|
target: str
|
|
hostname: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WinRmCmdResult:
|
|
ok: bool
|
|
message: str
|
|
target: str
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
exit_code: int | None = None
|
|
|
|
|
|
def _winrm_session(
|
|
target: str,
|
|
user: str,
|
|
password: str,
|
|
*,
|
|
timeout_sec: int = 15,
|
|
):
|
|
from app.services.win_admin_settings import normalize_win_admin_user
|
|
|
|
user = normalize_win_admin_user(user.strip())
|
|
target = target.strip()
|
|
if not target or not user or not password:
|
|
raise WinAdminNotConfiguredError("Windows admin credentials or target missing")
|
|
|
|
try:
|
|
import winrm
|
|
except ImportError as exc:
|
|
raise RuntimeError("pywinrm is not installed on SAC server") from exc
|
|
|
|
operation_timeout_sec = max(5, timeout_sec)
|
|
read_timeout_sec = operation_timeout_sec + 15
|
|
settings = get_settings()
|
|
scheme = "https" if settings.sac_winrm_use_https else "http"
|
|
port = 5986 if settings.sac_winrm_use_https else 5985
|
|
endpoint = f"{scheme}://{target}:{port}/wsman"
|
|
cert_validation = (settings.sac_winrm_server_cert_validation or "validate").strip().lower()
|
|
if cert_validation not in {"validate", "ignore"}:
|
|
cert_validation = "validate"
|
|
session = winrm.Session(
|
|
endpoint,
|
|
auth=(user, password),
|
|
transport="ntlm",
|
|
read_timeout_sec=read_timeout_sec,
|
|
operation_timeout_sec=operation_timeout_sec,
|
|
server_cert_validation=cert_validation,
|
|
)
|
|
return session, winrm
|
|
|
|
|
|
def _clixml_to_plain(text: str) -> str:
|
|
"""Turn WinRM/PowerShell CLIXML blobs into readable text; drop progress noise."""
|
|
raw = text or ""
|
|
if _CLIXML_MARKER not in raw:
|
|
return raw.strip()
|
|
|
|
messages = re.findall(r'<S N="Message">([^<]*)</S>', raw, flags=re.IGNORECASE)
|
|
messages = [
|
|
msg.strip()
|
|
for msg in messages
|
|
if msg.strip() and msg.strip() not in _CLIXML_PROGRESS_NOISE
|
|
]
|
|
to_strings = re.findall(r"<ToString>([^<]*)</ToString>", raw, flags=re.IGNORECASE)
|
|
to_strings = [
|
|
text.strip()
|
|
for text in to_strings
|
|
if text.strip() and "Preparing modules" not in text
|
|
]
|
|
parts = [*messages, *to_strings]
|
|
if parts:
|
|
return "\n".join(parts)
|
|
|
|
without = re.sub(r"#< CLIXML.*", "", raw, flags=re.DOTALL).strip()
|
|
return without
|
|
|
|
|
|
def _decode_text_score(text: str) -> int:
|
|
cyrillic = len(_CYRILLIC_RE.findall(text))
|
|
suspicious = len(re.findall(r"\?{3,}", text))
|
|
return cyrillic * 5 - text.count("\ufffd") * 20 - suspicious * 15 - max(0, text.count("?") - cyrillic) * 2
|
|
|
|
|
|
def _decode_winrm_bytes(data: bytes) -> str:
|
|
if not data:
|
|
return ""
|
|
for encoding in ("utf-8-sig", "utf-8"):
|
|
try:
|
|
text = data.decode(encoding)
|
|
except UnicodeDecodeError:
|
|
break
|
|
score = _decode_text_score(text)
|
|
if text.isascii() or score > 0:
|
|
return text
|
|
best_score = -10_000
|
|
best_text = ""
|
|
for encoding in ("cp866", "cp1251"):
|
|
try:
|
|
text = data.decode(encoding)
|
|
except UnicodeDecodeError:
|
|
continue
|
|
score = _decode_text_score(text)
|
|
if score > best_score:
|
|
best_score = score
|
|
best_text = text
|
|
if best_text:
|
|
return best_text
|
|
return data.decode("utf-8", errors="replace")
|
|
|
|
|
|
def _winrm_failure_detail(stdout: str, stderr: str, exit_code: int) -> str:
|
|
stdout_plain = _clixml_to_plain(stdout)
|
|
stderr_plain = _clixml_to_plain(stderr)
|
|
|
|
for text in (stderr_plain, stdout_plain):
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith("ERROR:"):
|
|
return stripped[6:].strip() or stripped
|
|
|
|
for candidate in (stderr_plain, stdout_plain):
|
|
if candidate and _CLIXML_MARKER not in candidate:
|
|
lines = [line.strip() for line in candidate.splitlines() if line.strip()]
|
|
if lines:
|
|
return lines[-1][:2000]
|
|
|
|
if stdout_plain:
|
|
return stdout_plain[:2000]
|
|
if stderr_plain:
|
|
return stderr_plain[:2000]
|
|
return (
|
|
f"PowerShell exit code {exit_code} "
|
|
"(CLIXML без текста ошибки — проверьте Deploy-LoginMonitor.ps1 на ПК)"
|
|
)
|
|
|
|
|
|
def _sanitize_winrm_streams(stdout: str, stderr: str) -> tuple[str, str]:
|
|
return _clixml_to_plain(stdout), _clixml_to_plain(stderr)
|
|
|
|
|
|
def _finalize_winrm_run(
|
|
*,
|
|
target: str,
|
|
stdout: str,
|
|
stderr: str,
|
|
exit_code: int,
|
|
) -> WinRmCmdResult:
|
|
ok = exit_code == 0
|
|
if ok:
|
|
message = f"WinRM OK ({target}), exit 0"
|
|
if stdout.strip():
|
|
message = f"{message}\n{stdout.strip().splitlines()[0][:200]}"
|
|
else:
|
|
detail = _winrm_failure_detail(stdout, stderr, exit_code)
|
|
message = f"WinRM command failed ({target}): {detail}"
|
|
return WinRmCmdResult(
|
|
ok=ok,
|
|
message=message,
|
|
target=target,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
exit_code=exit_code,
|
|
)
|
|
|
|
|
|
_SYSTEM32_CMDS = {
|
|
"hostname": "hostname.exe",
|
|
"qwinsta": "qwinsta.exe",
|
|
"logoff": "logoff.exe",
|
|
}
|
|
|
|
|
|
def _resolve_winrm_cmd(remote_cmd: str) -> str:
|
|
"""WinRM cmd shell may have minimal PATH — use explicit System32 for built-ins."""
|
|
text = remote_cmd.strip()
|
|
if not text:
|
|
return text
|
|
parts = text.split(None, 1)
|
|
name = parts[0].lower()
|
|
rest = parts[1] if len(parts) > 1 else ""
|
|
exe = _SYSTEM32_CMDS.get(name)
|
|
if exe:
|
|
suffix = f" {rest}" if rest else ""
|
|
return f"%SystemRoot%\\System32\\{exe}{suffix}"
|
|
return text
|
|
|
|
|
|
def run_winrm_ps(
|
|
*,
|
|
target: str,
|
|
user: str,
|
|
password: str,
|
|
script: str,
|
|
timeout_sec: int = 30,
|
|
) -> WinRmCmdResult:
|
|
target = target.strip()
|
|
try:
|
|
session, winrm = _winrm_session(target, user, password, timeout_sec=timeout_sec)
|
|
result = session.run_ps(script)
|
|
except winrm.exceptions.WinRMTransportError as exc:
|
|
detail = str(exc)
|
|
hint = ""
|
|
if "credentials were rejected" in detail.lower():
|
|
hint = " Проверьте логин (B26\\user), пароль и WinRM на ПК."
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message=f"WinRM transport error ({target}): {detail}{hint}",
|
|
target=target,
|
|
)
|
|
except (socket.timeout, TimeoutError):
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message=f"WinRM timeout ({timeout_sec}s) to {target}:5985",
|
|
target=target,
|
|
)
|
|
except WinAdminNotConfiguredError as exc:
|
|
return WinRmCmdResult(ok=False, message=str(exc), target=target)
|
|
except RuntimeError as exc:
|
|
return WinRmCmdResult(ok=False, message=str(exc), target=target)
|
|
except Exception as exc:
|
|
return WinRmCmdResult(ok=False, message=f"WinRM error ({target}): {exc}", target=target)
|
|
|
|
stdout = _decode_winrm_bytes(result.std_out or b"")
|
|
stderr = _decode_winrm_bytes(result.std_err or b"")
|
|
exit_code = int(result.status_code)
|
|
stdout, stderr = _sanitize_winrm_streams(stdout, stderr)
|
|
return _finalize_winrm_run(
|
|
target=target,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
exit_code=exit_code,
|
|
)
|
|
|
|
|
|
def run_winrm_cmd(
|
|
*,
|
|
target: str,
|
|
user: str,
|
|
password: str,
|
|
remote_cmd: str,
|
|
timeout_sec: int = 30,
|
|
) -> WinRmCmdResult:
|
|
target = target.strip()
|
|
try:
|
|
session, winrm = _winrm_session(target, user, password, timeout_sec=timeout_sec)
|
|
result = session.run_cmd(_resolve_winrm_cmd(remote_cmd))
|
|
except winrm.exceptions.WinRMTransportError as exc:
|
|
detail = str(exc)
|
|
hint = ""
|
|
if "credentials were rejected" in detail.lower():
|
|
hint = " Проверьте логин (B26\\user), пароль и WinRM на ПК."
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message=f"WinRM transport error ({target}): {detail}{hint}",
|
|
target=target,
|
|
)
|
|
except (socket.timeout, TimeoutError):
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message=f"WinRM timeout ({timeout_sec}s) to {target}:5985",
|
|
target=target,
|
|
)
|
|
except WinAdminNotConfiguredError as exc:
|
|
return WinRmCmdResult(ok=False, message=str(exc), target=target)
|
|
except RuntimeError as exc:
|
|
return WinRmCmdResult(ok=False, message=str(exc), target=target)
|
|
except Exception as exc:
|
|
return WinRmCmdResult(ok=False, message=f"WinRM error ({target}): {exc}", target=target)
|
|
|
|
stdout = _decode_winrm_bytes(result.std_out or b"")
|
|
stderr = _decode_winrm_bytes(result.std_err or b"")
|
|
exit_code = int(result.status_code)
|
|
stdout, stderr = _sanitize_winrm_streams(stdout, stderr)
|
|
return _finalize_winrm_run(
|
|
target=target,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
exit_code=exit_code,
|
|
)
|
|
|
|
|
|
def run_winrm_qwinsta(*, target: str, user: str, password: str) -> WinRmCmdResult:
|
|
return run_winrm_cmd(target=target, user=user, password=password, remote_cmd="qwinsta", timeout_sec=45)
|
|
|
|
|
|
def run_winrm_logoff(*, target: str, user: str, password: str, session_id: int) -> WinRmCmdResult:
|
|
if session_id < 0:
|
|
return WinRmCmdResult(ok=False, message="Invalid session_id", target=target)
|
|
return run_winrm_cmd(
|
|
target=target,
|
|
user=user,
|
|
password=password,
|
|
remote_cmd=f"logoff {session_id} /v",
|
|
timeout_sec=45,
|
|
)
|
|
|
|
|
|
def _powershell_script_prelude() -> str:
|
|
return (
|
|
"$ProgressPreference = 'SilentlyContinue'\n"
|
|
"$InformationPreference = 'SilentlyContinue'\n"
|
|
"$utf8NoBom = New-Object System.Text.UTF8Encoding $false\n"
|
|
"[Console]::OutputEncoding = $utf8NoBom\n"
|
|
"$OutputEncoding = $utf8NoBom\n"
|
|
)
|
|
|
|
|
|
def _powershell_literal_path(path: str) -> str:
|
|
return path.replace("'", "''")
|
|
|
|
|
|
def _wrap_powershell_body(body: str) -> str:
|
|
indented = "\n".join(f" {line}" if line else "" for line in body.splitlines())
|
|
return (
|
|
_powershell_script_prelude()
|
|
+ "try {\n"
|
|
+ indented
|
|
+ "\n} catch {\n"
|
|
+ " Write-Output ('ERROR: ' + $_.Exception.Message)\n"
|
|
+ " exit 1\n"
|
|
+ "}\n"
|
|
)
|
|
|
|
|
|
def _custom_deploy_body(script_path: str) -> str:
|
|
literal = _powershell_literal_path(script_path.strip())
|
|
return (
|
|
f"$p = '{literal}'\n"
|
|
"if (-not (Test-Path -LiteralPath $p)) { throw \"Deploy script not found: $p\" }\n"
|
|
"Write-Output \"Running: $p\"\n"
|
|
"& $p"
|
|
)
|
|
|
|
|
|
def _rdp_staging_setup_ps() -> str:
|
|
dirname = _powershell_literal_path(RDP_REMOTE_STAGING_DIRNAME)
|
|
legacy = _powershell_literal_path(RDP_LEGACY_STAGING)
|
|
return (
|
|
f"$staging = Join-Path $env:TEMP '{dirname}'\n"
|
|
f"$legacyStaging = '{legacy}'\n"
|
|
"if (Test-Path -LiteralPath $legacyStaging) {\n"
|
|
" Remove-Item -LiteralPath $legacyStaging -Recurse -Force -ErrorAction SilentlyContinue\n"
|
|
"}\n"
|
|
)
|
|
|
|
|
|
def _prepare_staging_body() -> str:
|
|
return (
|
|
_rdp_staging_setup_ps()
|
|
+ "if (Test-Path -LiteralPath $staging) { Remove-Item -LiteralPath $staging -Recurse -Force }\n"
|
|
+ "New-Item -ItemType Directory -Path $staging -Force | Out-Null\n"
|
|
+ "Write-Output \"Staging ready: $staging\"\n"
|
|
)
|
|
|
|
|
|
def _deploy_from_staging_body() -> str:
|
|
return (
|
|
_rdp_staging_setup_ps()
|
|
+ "$deploy = Join-Path $staging 'Deploy-LoginMonitor.ps1'\n"
|
|
+ "if (-not (Test-Path -LiteralPath $staging)) { throw 'Staging directory missing' }\n"
|
|
+ "if (-not (Test-Path -LiteralPath $deploy)) { throw \"Deploy-LoginMonitor.ps1 missing in staging\" }\n"
|
|
+ "$logPath = Join-Path $env:ProgramData 'RDP-login-monitor\\Logs\\deploy.log'\n"
|
|
+ "Write-Output \"Running: $deploy -SourceShareRoot $staging\"\n"
|
|
+ "powershell.exe -NoProfile -ExecutionPolicy Bypass -File $deploy -SourceShareRoot $staging\n"
|
|
+ "if (Test-Path -LiteralPath $logPath) {\n"
|
|
+ " Write-Output ''\n"
|
|
+ " Write-Output '--- deploy.log ---'\n"
|
|
+ " Get-Content -LiteralPath $logPath -Encoding UTF8 | Select-Object -Last 60\n"
|
|
+ "}\n"
|
|
)
|
|
|
|
|
|
def _bundle_download_url(token: str) -> str:
|
|
settings = get_settings()
|
|
base = (settings.sac_agent_bundle_base_url or settings.sac_public_url).rstrip("/")
|
|
return f"{base}/api/v1/agent/rdp-bundle/{token}"
|
|
|
|
|
|
def _download_bundle_body(bundle_url: str) -> str:
|
|
url = _powershell_literal_path(bundle_url)
|
|
return (
|
|
_rdp_staging_setup_ps()
|
|
+ "if (-not (Test-Path -LiteralPath $staging)) { throw 'Staging directory missing' }\n"
|
|
+ f"$url = '{url}'\n"
|
|
+ "$zip = Join-Path $env:TEMP ('sac-rdp-bundle-' + [Guid]::NewGuid().ToString() + '.zip')\n"
|
|
+ "[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12\n"
|
|
+ "Write-Output \"Downloading bundle: $url\"\n"
|
|
+ "Invoke-WebRequest -Uri $url -OutFile $zip -UseBasicParsing\n"
|
|
+ "if (-not (Test-Path -LiteralPath $zip)) { throw 'Bundle download produced no file' }\n"
|
|
+ "$zipSize = (Get-Item -LiteralPath $zip).Length\n"
|
|
+ "if ($zipSize -lt 100) { throw \"Bundle too small ($zipSize bytes)\" }\n"
|
|
+ "Add-Type -AssemblyName System.IO.Compression.FileSystem\n"
|
|
+ "[System.IO.Compression.ZipFile]::ExtractToDirectory($zip, $staging)\n"
|
|
+ "Remove-Item -LiteralPath $zip -Force -ErrorAction SilentlyContinue\n"
|
|
+ "Write-Output \"Bundle extracted to $staging\"\n"
|
|
)
|
|
|
|
|
|
def _fetch_rdp_bundle_dir(repo_url: str, git_branch: str) -> Path:
|
|
settings = get_settings()
|
|
return clone_or_update_repo(
|
|
repo_url,
|
|
git_branch,
|
|
_cache_dir(),
|
|
token=(settings.sac_agent_git_token or "").strip(),
|
|
)
|
|
|
|
|
|
def run_winrm_rdp_monitor_update(
|
|
*,
|
|
target: str,
|
|
user: str,
|
|
password: str,
|
|
script_path: str = "",
|
|
repo_url: str = "",
|
|
git_branch: str = "main",
|
|
) -> WinRmCmdResult:
|
|
target = target.strip()
|
|
if script_path.strip():
|
|
return run_winrm_ps(
|
|
target=target,
|
|
user=user,
|
|
password=password,
|
|
script=_wrap_powershell_body(_custom_deploy_body(script_path)),
|
|
timeout_sec=900,
|
|
)
|
|
|
|
effective_repo = (
|
|
repo_url or "https://git.papatramp.ru/PapaTramp/RDP-login-monitor.git"
|
|
).strip()
|
|
effective_branch = (git_branch or "main").strip() or "main"
|
|
if not effective_repo:
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message="rdp_git_repo_url is not configured in SAC",
|
|
target=target,
|
|
)
|
|
|
|
try:
|
|
repo_dir = _fetch_rdp_bundle_dir(effective_repo, effective_branch)
|
|
except Exception as exc:
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message=f"SAC failed to fetch RDP bundle from git: {exc}",
|
|
target=target,
|
|
)
|
|
|
|
missing = [name for name in RDP_BUNDLE_REQUIRED if not (repo_dir / name).is_file()]
|
|
if missing:
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message=f"RDP bundle incomplete on SAC: missing {', '.join(missing)}",
|
|
target=target,
|
|
)
|
|
|
|
zip_bytes = build_rdp_bundle_zip(repo_dir, RDP_BUNDLE_FILES)
|
|
if not zip_bytes:
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message="RDP bundle zip is empty on SAC",
|
|
target=target,
|
|
)
|
|
bundle_token = register_rdp_bundle_zip(zip_bytes)
|
|
bundle_url = _bundle_download_url(bundle_token)
|
|
staging_label = f"%TEMP%\\{RDP_REMOTE_STAGING_DIRNAME}"
|
|
|
|
prep = run_winrm_ps(
|
|
target=target,
|
|
user=user,
|
|
password=password,
|
|
script=_wrap_powershell_body(_prepare_staging_body()),
|
|
timeout_sec=120,
|
|
)
|
|
if not prep.ok:
|
|
return prep
|
|
|
|
download = run_winrm_ps(
|
|
target=target,
|
|
user=user,
|
|
password=password,
|
|
script=_wrap_powershell_body(_download_bundle_body(bundle_url)),
|
|
timeout_sec=300,
|
|
)
|
|
if not download.ok:
|
|
return WinRmCmdResult(
|
|
ok=False,
|
|
message=(
|
|
f"Failed to download RDP bundle on client: {download.message} "
|
|
"(bundle URL omitted from logs)"
|
|
),
|
|
target=target,
|
|
stdout="\n\n".join(part.strip() for part in [prep.stdout, download.stdout] if part.strip()),
|
|
stderr=download.stderr,
|
|
exit_code=download.exit_code,
|
|
)
|
|
|
|
deploy = run_winrm_ps(
|
|
target=target,
|
|
user=user,
|
|
password=password,
|
|
script=_wrap_powershell_body(_deploy_from_staging_body()),
|
|
timeout_sec=900,
|
|
)
|
|
header = f"SAC served bundle from git; client staging {staging_label}"
|
|
stdout = "\n\n".join(
|
|
part.strip()
|
|
for part in [header, prep.stdout, download.stdout, deploy.stdout]
|
|
if part.strip()
|
|
)
|
|
return WinRmCmdResult(
|
|
ok=deploy.ok,
|
|
message=deploy.message,
|
|
target=target,
|
|
stdout=stdout,
|
|
stderr=deploy.stderr,
|
|
exit_code=deploy.exit_code,
|
|
)
|
|
|
|
|
|
def run_winrm_on_host_targets(
|
|
host: Host,
|
|
*,
|
|
user: str,
|
|
password: str,
|
|
action,
|
|
) -> tuple[WinRmCmdResult | None, list[str]]:
|
|
"""Try WinRM targets in hostname-first order; action(target) -> WinRmCmdResult."""
|
|
attempts: list[str] = []
|
|
last: WinRmCmdResult | None = None
|
|
for target in iter_winrm_targets(host):
|
|
result = action(target=target)
|
|
last = result
|
|
attempts.append(f"{target}={'OK' if result.ok else 'fail'}")
|
|
if result.ok:
|
|
return result, attempts
|
|
return last, attempts
|
|
|
|
|
|
def resolve_windows_host_target(host: Host) -> str:
|
|
targets = iter_winrm_targets(host)
|
|
if not targets:
|
|
raise HostTargetMissingError("Host has no hostname or IPv4 for WinRM test")
|
|
return targets[0]
|
|
|
|
|
|
def _netbios_name_variants(name: str | None) -> list[str]:
|
|
"""Короткое имя ПК (NetBIOS), как Enter-PSSession -ComputerName Andrisonova-PC."""
|
|
text = (name or "").strip()
|
|
if not text:
|
|
return []
|
|
short = text.split(".")[0].split("\\")[0].strip()
|
|
if not short:
|
|
return []
|
|
if short.casefold() == text.casefold():
|
|
return [short]
|
|
return [short, text]
|
|
|
|
|
|
def _looks_like_computer_name(value: str) -> bool:
|
|
text = value.strip()
|
|
if not text or " " in text:
|
|
return False
|
|
return len(text) <= 63
|
|
|
|
|
|
def iter_winrm_targets(host: Host) -> list[str]:
|
|
"""WinRM + NTLM: сначала имя хоста (как Enter-PSSession -ComputerName), IP — последним."""
|
|
if not is_windows_host(host):
|
|
raise HostNotWindowsError("Host is not Windows")
|
|
|
|
seen: set[str] = set()
|
|
ordered: list[str] = []
|
|
|
|
def add(value: str | None) -> None:
|
|
text = (value or "").strip()
|
|
if not text or text.casefold() in seen:
|
|
return
|
|
seen.add(text.casefold())
|
|
ordered.append(text)
|
|
|
|
for variant in _netbios_name_variants(host.hostname):
|
|
add(variant)
|
|
|
|
inventory = host.inventory if isinstance(host.inventory, dict) else {}
|
|
computer_name = inventory.get("computer_name")
|
|
if isinstance(computer_name, str):
|
|
for variant in _netbios_name_variants(computer_name):
|
|
add(variant)
|
|
|
|
if host.display_name and _looks_like_computer_name(host.display_name):
|
|
for variant in _netbios_name_variants(host.display_name):
|
|
add(variant)
|
|
|
|
add(host.ipv4)
|
|
return ordered
|
|
|
|
|
|
def is_windows_host(host: Host) -> bool:
|
|
if (host.os_family or "").strip().lower() == "windows":
|
|
return True
|
|
return (host.product or "").strip() == "rdp-login-monitor"
|
|
|
|
|
|
def test_winrm_connection(
|
|
*,
|
|
target: str,
|
|
user: str,
|
|
password: str,
|
|
timeout_sec: int = 15,
|
|
) -> WinRmTestResult:
|
|
result = run_winrm_cmd(
|
|
target=target,
|
|
user=user,
|
|
password=password,
|
|
remote_cmd="hostname",
|
|
timeout_sec=timeout_sec,
|
|
)
|
|
if result.ok and result.stdout.strip():
|
|
host = result.stdout.strip().splitlines()[0]
|
|
return WinRmTestResult(
|
|
ok=True,
|
|
message=f"WinRM OK, hostname={host}",
|
|
target=result.target,
|
|
hostname=host,
|
|
)
|
|
return WinRmTestResult(
|
|
ok=result.ok,
|
|
message=result.message,
|
|
target=result.target,
|
|
hostname=None,
|
|
)
|