117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
"""Per-host management credentials (override global Win/Linux admin)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.host import Host
|
|
from app.services.linux_admin_settings import (
|
|
LinuxAdminConfig,
|
|
get_effective_linux_admin_for_host,
|
|
)
|
|
from app.services.win_admin_settings import (
|
|
WinAdminConfig,
|
|
get_effective_win_admin_for_host,
|
|
normalize_win_admin_user,
|
|
)
|
|
|
|
|
|
def mask_secret(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
text = value.strip()
|
|
if not text:
|
|
return None
|
|
if len(text) <= 4:
|
|
return "****"
|
|
return f"{text[:2]}…{text[-2:]}"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HostMgmtAccessView:
|
|
host_id: int
|
|
has_override: bool
|
|
user: str | None
|
|
password_set: bool
|
|
password_hint: str | None
|
|
effective_source: str
|
|
effective_configured: bool
|
|
effective_user: str | None
|
|
|
|
|
|
def _effective_for_host(db: Session, host: Host) -> WinAdminConfig | LinuxAdminConfig:
|
|
os_family = (host.os_family or "").strip().lower()
|
|
if os_family == "windows" or (host.product or "").strip().lower() in {"rdp-login-monitor", "rdp"}:
|
|
return get_effective_win_admin_for_host(db, host)
|
|
if os_family == "linux" or (host.product or "").strip().lower() in {"ssh-monitor", "ssh"}:
|
|
return get_effective_linux_admin_for_host(db, host)
|
|
# Fallback: try host override first via win helper (same columns), then global win, then linux.
|
|
win_cfg = get_effective_win_admin_for_host(db, host)
|
|
if win_cfg.configured:
|
|
return win_cfg
|
|
return get_effective_linux_admin_for_host(db, host)
|
|
|
|
|
|
def get_host_mgmt_access_view(db: Session, host: Host) -> HostMgmtAccessView:
|
|
host_user = (host.mgmt_user or "").strip() or None
|
|
host_password = (host.mgmt_password or "").strip() or None
|
|
has_override = bool(host_user or host_password)
|
|
effective = _effective_for_host(db, host)
|
|
return HostMgmtAccessView(
|
|
host_id=host.id,
|
|
has_override=has_override,
|
|
user=host_user,
|
|
password_set=bool(host_password),
|
|
password_hint=mask_secret(host_password),
|
|
effective_source=effective.source,
|
|
effective_configured=effective.configured,
|
|
effective_user=effective.user or None,
|
|
)
|
|
|
|
|
|
def upsert_host_mgmt_credentials(
|
|
db: Session,
|
|
host: Host,
|
|
*,
|
|
user: str | None = None,
|
|
password: str | None = None,
|
|
clear: bool = False,
|
|
) -> HostMgmtAccessView:
|
|
"""Update per-host credentials.
|
|
|
|
- clear=True → wipe host override (use global Settings).
|
|
- user="" → clear username.
|
|
- password omitted (None) → keep existing password.
|
|
- password="" → clear password.
|
|
"""
|
|
if clear:
|
|
host.mgmt_user = None
|
|
host.mgmt_password = None
|
|
db.commit()
|
|
db.refresh(host)
|
|
return get_host_mgmt_access_view(db, host)
|
|
|
|
if user is not None:
|
|
cleaned = user.strip()
|
|
if not cleaned:
|
|
host.mgmt_user = None
|
|
else:
|
|
# Preserve DOMAIN\\user form for Windows; for Linux leave as-is after strip.
|
|
os_family = (host.os_family or "").strip().lower()
|
|
if os_family == "windows" or "\\" in cleaned or "@" in cleaned:
|
|
host.mgmt_user = normalize_win_admin_user(cleaned)
|
|
else:
|
|
host.mgmt_user = cleaned
|
|
|
|
if password is not None:
|
|
if not password.strip():
|
|
host.mgmt_password = None
|
|
else:
|
|
host.mgmt_password = password
|
|
|
|
db.commit()
|
|
db.refresh(host)
|
|
return get_host_mgmt_access_view(db, host)
|