fix: WinRM NTLM username normalize and hostname-first targets (0.10.3)
This commit is contained in:
+40
-16
@@ -15,7 +15,8 @@ from app.services.winrm_connect import (
|
||||
HostNotWindowsError,
|
||||
HostTargetMissingError,
|
||||
WinAdminNotConfiguredError,
|
||||
resolve_windows_host_target,
|
||||
WinRmTestResult,
|
||||
iter_winrm_targets,
|
||||
test_winrm_connection,
|
||||
)
|
||||
from app.services.host_health import (
|
||||
@@ -186,28 +187,51 @@ def test_host_winrm(
|
||||
raise HTTPException(status_code=400, detail="Windows domain admin is not configured")
|
||||
|
||||
try:
|
||||
target = resolve_windows_host_target(host)
|
||||
targets = iter_winrm_targets(host)
|
||||
except HostNotWindowsError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
except HostTargetMissingError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
try:
|
||||
result = test_winrm_connection(
|
||||
target=target,
|
||||
user=cfg.user,
|
||||
password=cfg.password,
|
||||
)
|
||||
except WinAdminNotConfiguredError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
except RuntimeError as exc:
|
||||
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
||||
last_result = None
|
||||
attempts: list[str] = []
|
||||
for target in targets:
|
||||
try:
|
||||
result = test_winrm_connection(
|
||||
target=target,
|
||||
user=cfg.user,
|
||||
password=cfg.password,
|
||||
)
|
||||
except WinAdminNotConfiguredError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
except RuntimeError as exc:
|
||||
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
||||
last_result = result
|
||||
attempts.append(f"{target}={'OK' if result.ok else 'fail'}")
|
||||
if result.ok:
|
||||
if len(targets) > 1:
|
||||
result = WinRmTestResult(
|
||||
ok=True,
|
||||
message=f"{result.message} (пробовали: {', '.join(attempts)})",
|
||||
target=result.target,
|
||||
hostname=result.hostname,
|
||||
)
|
||||
return HostWinRmTestResponse(
|
||||
ok=result.ok,
|
||||
message=result.message,
|
||||
target=result.target,
|
||||
hostname=result.hostname,
|
||||
)
|
||||
|
||||
assert last_result is not None
|
||||
message = last_result.message
|
||||
if len(attempts) > 1:
|
||||
message = f"{message} (пробовали: {', '.join(attempts)})"
|
||||
return HostWinRmTestResponse(
|
||||
ok=result.ok,
|
||||
message=result.message,
|
||||
target=result.target,
|
||||
hostname=result.hostname,
|
||||
ok=False,
|
||||
message=message,
|
||||
target=last_result.target,
|
||||
hostname=last_result.hostname,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -21,10 +21,22 @@ class WinAdminConfig:
|
||||
return bool(self.user.strip() and self.password.strip())
|
||||
|
||||
|
||||
def normalize_win_admin_user(user: str) -> str:
|
||||
"""DOMAIN\\user — один обратный слэш; убрать лишнее экранирование из env/UI."""
|
||||
text = user.strip()
|
||||
if not text:
|
||||
return ""
|
||||
text = text.replace("\\\\", "\\")
|
||||
if "\\" not in text and "@" not in text and "/" not in text:
|
||||
# NETBIOS\user без слэша — не трогаем (оператор допишет в UI)
|
||||
return text
|
||||
return text
|
||||
|
||||
|
||||
def _win_admin_from_env() -> WinAdminConfig:
|
||||
settings = get_settings()
|
||||
return WinAdminConfig(
|
||||
user=settings.sac_win_admin_user.strip(),
|
||||
user=normalize_win_admin_user(settings.sac_win_admin_user),
|
||||
password=settings.sac_win_admin_password,
|
||||
source="env",
|
||||
)
|
||||
@@ -45,7 +57,7 @@ def get_effective_win_admin_config(db: Session | None = None) -> WinAdminConfig:
|
||||
if row is None:
|
||||
return env_cfg
|
||||
|
||||
user = (row.win_admin_user or "").strip() or env_cfg.user
|
||||
user = normalize_win_admin_user((row.win_admin_user or "").strip() or env_cfg.user)
|
||||
password = (row.win_admin_password or "") or env_cfg.password
|
||||
has_db_values = bool((row.win_admin_user or "").strip() or (row.win_admin_password or "").strip())
|
||||
return WinAdminConfig(
|
||||
@@ -73,7 +85,7 @@ def upsert_win_admin_settings(
|
||||
db.add(row)
|
||||
|
||||
if user is not None and user.strip():
|
||||
row.win_admin_user = user.strip()
|
||||
row.win_admin_user = normalize_win_admin_user(user)
|
||||
if password is not None and password.strip():
|
||||
row.win_admin_password = password
|
||||
|
||||
|
||||
@@ -29,12 +29,35 @@ class WinRmTestResult:
|
||||
|
||||
|
||||
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 iter_winrm_targets(host: Host) -> list[str]:
|
||||
"""WinRM + NTLM: сначала имя машины (SPN), затем IPv4."""
|
||||
if not is_windows_host(host):
|
||||
raise HostNotWindowsError("Host is not Windows")
|
||||
target = (host.ipv4 or "").strip() or (host.hostname or "").strip()
|
||||
if not target:
|
||||
raise HostTargetMissingError("Host has no IPv4 or hostname for WinRM test")
|
||||
return target
|
||||
|
||||
seen: set[str] = set()
|
||||
ordered: list[str] = []
|
||||
|
||||
def add(value: str | None) -> None:
|
||||
text = (value or "").strip()
|
||||
if not text or text in seen:
|
||||
return
|
||||
seen.add(text)
|
||||
ordered.append(text)
|
||||
|
||||
inventory = host.inventory if isinstance(host.inventory, dict) else {}
|
||||
computer_name = inventory.get("computer_name")
|
||||
if isinstance(computer_name, str):
|
||||
add(computer_name)
|
||||
add(host.hostname)
|
||||
add(host.display_name)
|
||||
add(host.ipv4)
|
||||
return ordered
|
||||
|
||||
|
||||
def is_windows_host(host: Host) -> bool:
|
||||
@@ -55,6 +78,10 @@ def test_winrm_connection(
|
||||
if not target or not user or not password:
|
||||
raise WinAdminNotConfiguredError("Windows admin credentials or target missing")
|
||||
|
||||
from app.services.win_admin_settings import normalize_win_admin_user
|
||||
|
||||
user = normalize_win_admin_user(user)
|
||||
|
||||
try:
|
||||
import winrm
|
||||
except ImportError as exc:
|
||||
@@ -73,9 +100,16 @@ def test_winrm_connection(
|
||||
)
|
||||
result = session.run_cmd("hostname")
|
||||
except winrm.exceptions.WinRMTransportError as exc:
|
||||
detail = str(exc)
|
||||
hint = ""
|
||||
if "credentials were rejected" in detail.lower():
|
||||
hint = (
|
||||
" Проверьте логин (B26\\user), пароль и что учётка — admin на целевом ПК; "
|
||||
"при подключении по IP NTLM часто отклоняет creds — используйте имя хоста."
|
||||
)
|
||||
return WinRmTestResult(
|
||||
ok=False,
|
||||
message=f"WinRM transport error: {exc}",
|
||||
message=f"WinRM transport error ({target}): {detail}{hint}",
|
||||
target=target,
|
||||
)
|
||||
except (socket.timeout, TimeoutError):
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
"""Единый источник версии SAC (API, health, логи, OpenAPI)."""
|
||||
|
||||
APP_NAME = "Security Alert Center"
|
||||
APP_VERSION = "0.10.2"
|
||||
APP_VERSION = "0.10.3"
|
||||
APP_VERSION_LABEL = f"{APP_NAME} v.{APP_VERSION}"
|
||||
|
||||
Reference in New Issue
Block a user