huty
dc2d2acc82
Scaffold an MVP of the natural-language ops terminal: inventory + intent template registry, SSH/WinRM/local connectors, risk-gated executor with SQLite audit log, Claude-driven agent layer using Function Calling, plus a Typer CLI and FastAPI surface. Includes 10 cross-OS intents (disk/system/service) and example inventory. Verified end-to-end on the local Windows host: hosts/intents listing, check_disk_usage execution, and WRITE-class confirmation gating. Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>
64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
"""Probe OS facts on first connect; cache in-memory so we only run uname once."""
|
|
from __future__ import annotations
|
|
|
|
from .models import Host, OSFamily
|
|
|
|
# In-memory cache. For production, persist to DB so facts survive restarts.
|
|
_FACTS_CACHE: dict[str, dict] = {}
|
|
|
|
|
|
def get_cached_facts(host: Host) -> dict | None:
|
|
return _FACTS_CACHE.get(host.name)
|
|
|
|
|
|
def set_cached_facts(host: Host, facts: dict) -> None:
|
|
_FACTS_CACHE[host.name] = facts
|
|
|
|
|
|
async def probe_facts(host: Host, connector) -> dict:
|
|
"""Run a tiny shell snippet to discover OS family + distribution.
|
|
|
|
`connector` is an already-open connector instance for this host.
|
|
Returns a facts dict and caches it.
|
|
"""
|
|
cached = get_cached_facts(host)
|
|
if cached:
|
|
return cached
|
|
|
|
facts: dict = {"os_family": host.os_family.value}
|
|
|
|
# If we already know it from inventory, trust it.
|
|
if host.os_family != OSFamily.UNKNOWN:
|
|
set_cached_facts(host, facts)
|
|
return facts
|
|
|
|
# Heuristic: try Linux/Unix first, then Windows.
|
|
# Note: on Windows hosts with Git-Bash installed, `uname -s` may return
|
|
# MINGW/MSYS/CYGWIN — those still mean the underlying OS is Windows.
|
|
try:
|
|
res = await connector.run("uname -s", timeout=5)
|
|
if res.exit_code == 0 and res.stdout.strip():
|
|
kernel = res.stdout.strip().lower()
|
|
if any(tok in kernel for tok in ("mingw", "msys", "cygwin")):
|
|
facts["os_family"] = OSFamily.WINDOWS.value
|
|
elif "linux" in kernel:
|
|
facts["os_family"] = OSFamily.LINUX.value
|
|
elif "darwin" in kernel:
|
|
facts["os_family"] = OSFamily.DARWIN.value
|
|
elif "aix" in kernel:
|
|
facts["os_family"] = OSFamily.AIX.value
|
|
|
|
if facts["os_family"] == OSFamily.UNKNOWN.value:
|
|
# Try Windows PowerShell as last resort.
|
|
res = await connector.run(
|
|
"(Get-CimInstance Win32_OperatingSystem).Caption", timeout=5
|
|
)
|
|
if res.exit_code == 0 and res.stdout.strip():
|
|
facts["os_family"] = OSFamily.WINDOWS.value
|
|
facts["os_caption"] = res.stdout.strip()
|
|
except Exception as e: # noqa: BLE001 — probing is best-effort
|
|
facts["probe_error"] = str(e)
|
|
|
|
set_cached_facts(host, facts)
|
|
return facts
|