-
Jobs
+
Jobs (auto-refresh 10s)
{
:key="job.id"
class="cursor-pointer hover:bg-slate-50"
:class="{ 'bg-blue-50': job.id === selectedJobId }"
- @click="watchJob(job.id)"
+ @click="viewJob(job.id)"
>
| #{{ job.id }} |
{{ job.type }} |
- {{ job.status }} |
+
+
+ {{ job.status }}
+
+ |
+ {{ job.progress_percent }}% |
|
- | Keine Jobs. |
+ Keine Jobs. |
@@ -96,13 +186,15 @@ async function watchJob(jobId: number): Promise
{
- Live-Log {{ selectedJobId ? `(Job #${selectedJobId})` : '' }}
+ Log {{ selectedJobId ? `(Job #${selectedJobId})` : '' }}
- Kein Job ausgewählt — klicke links einen Job an.
+ Kein Job ausgewählt - klicke links einen Job an.
+
+
+ {{ log.line }}
-
{{ log.line }}
diff --git a/satellite/README.md b/satellite/README.md
new file mode 100644
index 0000000..2328959
--- /dev/null
+++ b/satellite/README.md
@@ -0,0 +1,64 @@
+# Insight Updater Satellite
+
+Remote-Agent fuer Kundennetzwerke. Laedt Jobs von der zentralen Insight-Updater-Instanz,
+fuehrt sie lokal im Kundennetz aus (WinRM / SSH / CAU / Netzwerk-Scan) und meldet
+Logs und Ergebnisse zurueck. Kein Docker noetig - eine einzelne Binary genuegt.
+
+## Prinzip
+
+- Nur ausgehende HTTPS-Verbindungen zur Zentrale (keine Firewall-Loecher beim Kunden)
+- Pull-Modell: der Satellite pollt alle N Sekunden nach Jobs
+- Credentials (WinRM/SSH) liegen ausschliesslich lokal in `credentials.yaml`
+- 1-2 Satelliten pro Kunde reichen - sie steuern das ganze Netz (wie CAU im Cluster)
+
+## Setup (Development)
+
+```bash
+cd satellite
+pip install -e .
+cp config.example.yaml config.yaml
+cp credentials.example.yaml credentials.yaml
+# config.yaml: central_url + api_key eintragen (Key aus dem Dashboard)
+# credentials.yaml: WinRM-/SSH-Zugangsdaten des Kundennetzes pflegen
+insight-satellite --config config.yaml --credentials credentials.yaml
+```
+
+## Windows-Binary bauen
+
+```bash
+pip install -e ".[build]"
+pyinstaller --onefile --name insight-satellite satellite/runner.py
+# Ergebnis: dist/insight-satellite.exe
+```
+
+Auf dem Zielsystem (Windows-Server beim Kunden):
+
+```
+C:\insight-satellite\
+ insight-satellite.exe
+ config.yaml
+ credentials.yaml
+```
+
+Start als Scheduled Task (Beispiel, ohne Umlaute):
+
+```powershell
+schtasks /create /tn "InsightUpdaterSatellite" /tr "C:\insight-satellite\insight-satellite.exe --config C:\insight-satellite\config.yaml --credentials C:\insight-satellite\credentials.yaml" /sc onstart /ru SYSTEM /rl HIGHEST
+```
+
+## Job-Typen
+
+| Typ | Aktion |
+|---|---|
+| `windows_update` | Windows Update via WinRM (COM Microsoft.Update.Session) |
+| `linux_update` | apt/dnf/yum upgrade via SSH mit sudo |
+| `cau_run` | Invoke-CauRun auf einem Failover-Cluster |
+| `health_check` | Verbindungstest, Ergebnis geht an die Zentrale |
+| `network_scan` | Ping-Sweep + Port-Probe (5985/22), legt gefundene Hosts zentral an |
+
+## Ablauf pro Job
+
+1. `GET /api/satellite/poll` - Jobs abholen (werden dabei claimed)
+2. Lokal ausfuehren, Log-Zeilen sammeln
+3. `POST /api/satellite/logs` - Batches waehrend der Ausfuehrung
+4. `POST /api/satellite/result` - Abschluss (success/failed + Fehlertext)
diff --git a/satellite/config.example.yaml b/satellite/config.example.yaml
new file mode 100644
index 0000000..1b097c9
--- /dev/null
+++ b/satellite/config.example.yaml
@@ -0,0 +1,19 @@
+# Insight Updater Satellite
+# Zentrale Verbindung
+central_url: https://updater.insight-it.de
+api_key: "ius_..." # API-Key aus dem Dashboard (Satellite anlegen)
+
+# Verhalten
+poll_interval: 30 # Sekunden zwischen Job-Polls
+heartbeat_interval: 60 # Sekunden zwischen Heartbeats
+log_batch_size: 50 # Log-Zeilen pro Upload
+log_flush_interval: 10 # Sekunden, nach denen Log-Puffer geflusht wird
+
+# WinRM-Defaults (koennen pro Credential ueberschrieben werden)
+winrm_transport: ntlm # ntlm | kerberos | credssp
+winrm_cert_validation: ignore
+
+# Netzwerk-Scan Defaults
+scan_default_subnet: "" # z.B. 192.168.1.0/24 - leer = Job-Parameter pflicht
+scan_ping_timeout_ms: 500
+scan_port_timeout_ms: 1500
diff --git a/satellite/credentials.example.yaml b/satellite/credentials.example.yaml
new file mode 100644
index 0000000..de9add1
--- /dev/null
+++ b/satellite/credentials.example.yaml
@@ -0,0 +1,20 @@
+# Lokale Credentials - NUR auf dem Satellite, niemals zentral!
+# Server in der Zentrale referenzieren diese Namen ueber "credential_ref".
+#
+# Beispiele:
+winrm-admin:
+ type: winrm
+ username: "DOMAIN\\svc_update"
+ password: "geheim"
+ # transport: ntlm # optional, ueberschreibt config.yaml
+
+linux-root:
+ type: ssh
+ username: "root"
+ password: "geheim"
+
+linux-key:
+ type: ssh
+ username: "update"
+ private_key_path: "/etc/insight-satellite/id_ed25519"
+ passphrase: ""
diff --git a/satellite/pyproject.toml b/satellite/pyproject.toml
new file mode 100644
index 0000000..69e2dc4
--- /dev/null
+++ b/satellite/pyproject.toml
@@ -0,0 +1,32 @@
+[project]
+name = "insight-updater-satellite"
+version = "0.2.0"
+description = "Insight Updater Satellite - remote update agent for customer networks"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+ "httpx>=0.26.0",
+ "pywinrm>=0.4.3",
+ "asyncssh>=2.14.0",
+ "pyyaml>=6.0.1",
+ "structlog>=24.1.0",
+ "tenacity>=8.2.0",
+]
+
+[project.optional-dependencies]
+kerberos = ["pywinrm[kerberos]>=0.4.3"]
+build = ["pyinstaller>=6.0.0"]
+
+[project.scripts]
+insight-satellite = "satellite.runner:main"
+
+[build-system]
+requires = ["setuptools>=68.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+include = ["satellite*"]
+
+[tool.ruff]
+target-version = "py311"
+line-length = 100
diff --git a/satellite/satellite/__init__.py b/satellite/satellite/__init__.py
new file mode 100644
index 0000000..66d75ec
--- /dev/null
+++ b/satellite/satellite/__init__.py
@@ -0,0 +1,3 @@
+"""Insight Updater Satellite - remote agent for customer networks."""
+
+__version__ = "0.2.0"
diff --git a/satellite/satellite/cau_exec.py b/satellite/satellite/cau_exec.py
new file mode 100644
index 0000000..b0d5c61
--- /dev/null
+++ b/satellite/satellite/cau_exec.py
@@ -0,0 +1,29 @@
+"""CAU executor: Cluster-Aware Updating via PowerShell remoting."""
+
+from collections.abc import AsyncIterator
+
+from satellite.config import Config, Credential
+from satellite.winrm_exec import Target, WinRMExecutor
+
+
+class CAUError(Exception):
+ pass
+
+
+class CAUExecutor:
+ def __init__(self, config: Config, cluster_name: str, port: int, credential: Credential) -> None:
+ self.cluster_name = cluster_name
+ self.winrm = WinRMExecutor(config, Target(cluster_name, port), credential)
+
+ async def invoke_cau_run(self) -> AsyncIterator[str]:
+ script = f"""
+ Import-Module ClusterAwareUpdating
+ Invoke-CauRun -ClusterName {self.cluster_name} `
+ -Force -Confirm:$false -WaitForCompletion |
+ Out-String
+ """
+ try:
+ async for line in self.winrm.stream_powershell(script):
+ yield line
+ except Exception as exc:
+ raise CAUError(f"CAU-Lauf fehlgeschlagen: {exc}") from exc
diff --git a/satellite/satellite/client.py b/satellite/satellite/client.py
new file mode 100644
index 0000000..50e0906
--- /dev/null
+++ b/satellite/satellite/client.py
@@ -0,0 +1,79 @@
+"""HTTP client for the central Insight Updater API."""
+
+from datetime import UTC, datetime
+from typing import Any
+
+import httpx
+
+from satellite.config import Config
+
+
+class CentralClient:
+ def __init__(self, config: Config, version: str) -> None:
+ self.config = config
+ self.version = version
+ self._client = httpx.AsyncClient(
+ base_url=config.central_url,
+ headers={"X-Api-Key": config.api_key},
+ timeout=httpx.Timeout(30.0),
+ )
+
+ async def close(self) -> None:
+ await self._client.aclose()
+
+ async def heartbeat(self, hostname: str) -> None:
+ r = await self._client.post(
+ "/api/satellite/heartbeat",
+ json={"version": self.version, "hostname": hostname},
+ )
+ r.raise_for_status()
+
+ async def poll(self) -> list[dict[str, Any]]:
+ r = await self._client.get("/api/satellite/poll")
+ r.raise_for_status()
+ return list(r.json().get("jobs", []))
+
+ async def push_logs(
+ self,
+ job_id: int,
+ lines: list[tuple[str, str]], # (level, line)
+ progress_percent: int | None = None,
+ current_phase: str | None = None,
+ ) -> None:
+ payload: dict[str, Any] = {
+ "job_id": job_id,
+ "lines": [
+ {"timestamp": datetime.now(UTC).isoformat(), "level": lvl, "line": ln}
+ for lvl, ln in lines
+ ],
+ }
+ if progress_percent is not None:
+ payload["progress_percent"] = progress_percent
+ if current_phase is not None:
+ payload["current_phase"] = current_phase
+ r = await self._client.post("/api/satellite/logs", json=payload)
+ r.raise_for_status()
+
+ async def push_result(self, job_id: int, success: bool, error: str | None = None) -> None:
+ r = await self._client.post(
+ "/api/satellite/result",
+ json={
+ "job_id": job_id,
+ "status": "success" if success else "failed",
+ "error": error,
+ },
+ )
+ r.raise_for_status()
+
+ async def push_scan_result(self, job_id: int, hosts: list[dict[str, Any]]) -> None:
+ r = await self._client.post(
+ "/api/satellite/scan-result", json={"job_id": job_id, "hosts": hosts}
+ )
+ r.raise_for_status()
+
+ async def push_health_report(self, server_id: int, ok: bool, message: str) -> None:
+ r = await self._client.post(
+ "/api/satellite/health-report",
+ json={"server_id": server_id, "ok": ok, "message": message},
+ )
+ r.raise_for_status()
diff --git a/satellite/satellite/config.py b/satellite/satellite/config.py
new file mode 100644
index 0000000..1dc1048
--- /dev/null
+++ b/satellite/satellite/config.py
@@ -0,0 +1,67 @@
+"""Satellite configuration: config.yaml + credentials.yaml loading."""
+
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import yaml
+
+
+@dataclass
+class Credential:
+ name: str
+ type: str # "winrm" | "ssh"
+ username: str
+ password: str | None = None
+ private_key_path: str | None = None
+ passphrase: str | None = None
+ transport: str | None = None
+
+
+@dataclass
+class Config:
+ central_url: str
+ api_key: str
+ poll_interval: int = 30
+ heartbeat_interval: int = 60
+ log_batch_size: int = 50
+ log_flush_interval: int = 10
+ winrm_transport: str = "ntlm"
+ winrm_cert_validation: str = "ignore"
+ scan_default_subnet: str = ""
+ scan_ping_timeout_ms: int = 500
+ scan_port_timeout_ms: int = 1500
+ credentials: dict[str, Credential] = field(default_factory=dict)
+
+ def credential(self, ref: str | None) -> Credential | None:
+ if not ref:
+ return None
+ cred = self.credentials.get(ref)
+ if cred is None:
+ raise KeyError(f"Credential '{ref}' nicht in credentials.yaml gefunden")
+ return cred
+
+
+def load_config(config_path: str = "config.yaml", credentials_path: str = "credentials.yaml") -> Config:
+ raw = yaml.safe_load(Path(config_path).read_text(encoding="utf-8")) or {}
+
+ creds: dict[str, Credential] = {}
+ cred_file = Path(credentials_path)
+ if cred_file.exists():
+ raw_creds = yaml.safe_load(cred_file.read_text(encoding="utf-8")) or {}
+ for name, c in raw_creds.items():
+ creds[name] = Credential(name=name, **c)
+
+ return Config(
+ central_url=raw["central_url"].rstrip("/"),
+ api_key=raw["api_key"],
+ poll_interval=int(raw.get("poll_interval", 30)),
+ heartbeat_interval=int(raw.get("heartbeat_interval", 60)),
+ log_batch_size=int(raw.get("log_batch_size", 50)),
+ log_flush_interval=int(raw.get("log_flush_interval", 10)),
+ winrm_transport=raw.get("winrm_transport", "ntlm"),
+ winrm_cert_validation=raw.get("winrm_cert_validation", "ignore"),
+ scan_default_subnet=raw.get("scan_default_subnet", ""),
+ scan_ping_timeout_ms=int(raw.get("scan_ping_timeout_ms", 500)),
+ scan_port_timeout_ms=int(raw.get("scan_port_timeout_ms", 1500)),
+ credentials=creds,
+ )
diff --git a/satellite/satellite/runner.py b/satellite/satellite/runner.py
new file mode 100644
index 0000000..aa7978e
--- /dev/null
+++ b/satellite/satellite/runner.py
@@ -0,0 +1,205 @@
+"""Satellite runner: main loop - heartbeat, poll, execute, report."""
+
+import argparse
+import asyncio
+import socket
+import sys
+import time
+from typing import Any
+
+import structlog
+
+from satellite import __version__
+from satellite.cau_exec import CAUExecutor
+from satellite.client import CentralClient
+from satellite.config import Config, load_config
+from satellite.scanner import scan_subnet
+from satellite.ssh_exec import SSHExecutor
+from satellite.ssh_exec import Target as SSHTarget
+from satellite.winrm_exec import Target as WinRMTarget
+from satellite.winrm_exec import WinRMExecutor
+
+logger = structlog.get_logger(__name__)
+
+
+class LogBuffer:
+ """Collects log lines and flushes them in batches to the central."""
+
+ def __init__(self, client: CentralClient, config: Config, job_id: int) -> None:
+ self.client = client
+ self.config = config
+ self.job_id = job_id
+ self.lines: list[tuple[str, str]] = []
+ self.last_flush = time.monotonic()
+
+ async def add(self, line: str, level: str = "info") -> None:
+ self.lines.append((level, line))
+ logger.info("job.log", job_id=self.job_id, line=line)
+ if (
+ len(self.lines) >= self.config.log_batch_size
+ or time.monotonic() - self.last_flush >= self.config.log_flush_interval
+ ):
+ await self.flush()
+
+ async def flush(self, progress: int | None = None, phase: str | None = None) -> None:
+ if not self.lines and progress is None:
+ return
+ try:
+ await self.client.push_logs(
+ self.job_id, self.lines, progress_percent=progress, current_phase=phase
+ )
+ except Exception as exc: # noqa: BLE001 - nicht wegen Log-Upload abbrechen
+ logger.warning("log_upload_failed", error=str(exc))
+ self.lines = []
+ self.last_flush = time.monotonic()
+
+
+async def execute_job(client: CentralClient, config: Config, job: dict[str, Any]) -> None:
+ job_id: int = job["job_id"]
+ buf = LogBuffer(client, config, job_id)
+ await buf.add(f"Job #{job_id} gestartet: {job['type']}")
+
+ try:
+ if job["type"] == "network_scan":
+ await _run_scan(client, config, job, buf)
+ elif job["type"] == "health_check":
+ await _run_health_check(client, config, job, buf)
+ elif job["type"] == "linux_update":
+ await _run_linux_update(config, job, buf)
+ elif job["type"] == "windows_update":
+ await _run_windows_update(config, job, buf)
+ elif job["type"] == "cau_run":
+ await _run_cau(config, job, buf)
+ else:
+ raise ValueError(f"Unbekannter Job-Typ: {job['type']}")
+
+ await buf.flush(progress=100)
+ await client.push_result(job_id, success=True)
+ logger.info("job.done", job_id=job_id)
+ except Exception as exc: # noqa: BLE001
+ await buf.add(f"FEHLER: {exc}", level="error")
+ await buf.flush()
+ await client.push_result(job_id, success=False, error=str(exc))
+ logger.error("job.failed", job_id=job_id, error=str(exc))
+
+
+async def _run_scan(client: CentralClient, config: Config, job: dict, buf: LogBuffer) -> None:
+ subnet = job.get("scan_subnet") or config.scan_default_subnet
+ if not subnet:
+ raise ValueError("Kein Subnetz angegeben (scan_subnet oder scan_default_subnet)")
+ await buf.add(f"Scanne Subnetz {subnet} ...")
+ hosts = await scan_subnet(subnet, config.scan_ping_timeout_ms, config.scan_port_timeout_ms)
+ await buf.add(f"{len(hosts)} verwaltbare Hosts gefunden")
+ await buf.flush(progress=90)
+ await client.push_scan_result(job["job_id"], hosts)
+
+
+async def _run_health_check(client: CentralClient, config: Config, job: dict, buf: LogBuffer) -> None:
+ ok, message = await _test_connection(config, job)
+ await buf.add(message, level="info" if ok else "error")
+ server_id = job.get("server_id")
+ if server_id:
+ await client.push_health_report(server_id, ok, message)
+ if not ok:
+ raise ConnectionError(message)
+
+
+async def _run_linux_update(config: Config, job: dict, buf: LogBuffer) -> None:
+ cred = config.credential(job.get("credential_ref"))
+ if not cred or cred.type != "ssh":
+ raise ValueError("SSH-Credential erforderlich")
+ executor = SSHExecutor(SSHTarget(job["hostname"], job.get("port") or 22), cred)
+ async for line in executor.stream_updates(job.get("reboot_if_required", False)):
+ await buf.add(line)
+
+
+async def _run_windows_update(config: Config, job: dict, buf: LogBuffer) -> None:
+ cred = config.credential(job.get("credential_ref"))
+ if not cred or cred.type != "winrm":
+ raise ValueError("WinRM-Credential erforderlich")
+ executor = WinRMExecutor(config, WinRMTarget(job["hostname"], job.get("port") or 5985), cred)
+ async for line in executor.install_updates(job.get("reboot_if_required", False)):
+ await buf.add(line)
+
+
+async def _run_cau(config: Config, job: dict, buf: LogBuffer) -> None:
+ cred = config.credential(job.get("credential_ref"))
+ if not cred or cred.type != "winrm":
+ raise ValueError("WinRM-Credential erforderlich")
+ executor = CAUExecutor(config, job["hostname"], job.get("port") or 5985, cred)
+ async for line in executor.invoke_cau_run():
+ await buf.add(line)
+
+
+async def _test_connection(config: Config, job: dict) -> tuple[bool, str]:
+ cred = config.credential(job.get("credential_ref"))
+ if cred is None:
+ return False, "Kein Credential referenziert"
+ if cred.type == "ssh":
+ executor = SSHExecutor(SSHTarget(job["hostname"], job.get("port") or 22), cred)
+ return await executor.test_connection()
+ executor = WinRMExecutor(config, WinRMTarget(job["hostname"], job.get("port") or 5985), cred)
+ return await executor.test_connection()
+
+
+async def run(config_path: str, credentials_path: str) -> None:
+ config = load_config(config_path, credentials_path)
+ client = CentralClient(config, __version__)
+ hostname = socket.gethostname()
+ last_heartbeat = 0.0
+
+ logger.info(
+ "satellite.starting",
+ version=__version__,
+ central=config.central_url,
+ credentials=len(config.credentials),
+ )
+
+ try:
+ while True:
+ now = time.monotonic()
+ if now - last_heartbeat >= config.heartbeat_interval:
+ try:
+ await client.heartbeat(hostname)
+ last_heartbeat = now
+ except Exception as exc: # noqa: BLE001
+ logger.warning("heartbeat_failed", error=str(exc))
+
+ try:
+ jobs = await client.poll()
+ except Exception as exc: # noqa: BLE001
+ logger.warning("poll_failed", error=str(exc))
+ await asyncio.sleep(config.poll_interval)
+ continue
+
+ for job in jobs:
+ await execute_job(client, config, job)
+
+ if not jobs:
+ await asyncio.sleep(config.poll_interval)
+ finally:
+ await client.close()
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description="Insight Updater Satellite")
+ parser.add_argument("--config", default="config.yaml")
+ parser.add_argument("--credentials", default="credentials.yaml")
+ args = parser.parse_args()
+
+ structlog.configure(
+ processors=[
+ structlog.processors.TimeStamper(fmt="iso"),
+ structlog.processors.add_log_level,
+ structlog.dev.ConsoleRenderer(),
+ ]
+ )
+
+ try:
+ asyncio.run(run(args.config, args.credentials))
+ except KeyboardInterrupt:
+ sys.exit(0)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/satellite/satellite/scanner.py b/satellite/satellite/scanner.py
new file mode 100644
index 0000000..8bc71d7
--- /dev/null
+++ b/satellite/satellite/scanner.py
@@ -0,0 +1,72 @@
+"""Network scanner: ping sweep + WinRM/SSH port probe for auto-discovery."""
+
+import asyncio
+import ipaddress
+import shutil
+
+
+async def _ping(ip: str, timeout_ms: int) -> bool:
+ if not shutil.which("ping"):
+ return True # kein ping verfuegbar -> trotzdem Port-Check versuchen
+ proc = await asyncio.create_subprocess_exec(
+ "ping", "-c", "1", "-W", str(max(1, timeout_ms // 1000)), ip,
+ stdout=asyncio.subprocess.DEVNULL,
+ stderr=asyncio.subprocess.DEVNULL,
+ )
+ await proc.wait()
+ return proc.returncode == 0
+
+
+async def _port_open(ip: str, port: int, timeout_ms: int) -> bool:
+ try:
+ _reader, writer = await asyncio.wait_for(
+ asyncio.open_connection(ip, port), timeout=timeout_ms / 1000
+ )
+ writer.close()
+ await writer.wait_closed()
+ return True
+ except (TimeoutError, OSError):
+ return False
+
+
+async def _resolve(ip: str) -> str:
+ def _do() -> str:
+ import socket
+
+ try:
+ return socket.gethostbyaddr(ip)[0]
+ except OSError:
+ return ip
+
+ return await asyncio.to_thread(_do)
+
+
+async def scan_subnet(subnet: str, ping_timeout_ms: int, port_timeout_ms: int) -> list[dict]:
+ """Scan a subnet; return hosts with winrm_open / ssh_open flags."""
+ network = ipaddress.ip_network(subnet, strict=False)
+ hosts: list[dict] = []
+ sem = asyncio.Semaphore(64)
+
+ async def probe(ip: str) -> None:
+ async with sem:
+ if not await _ping(ip, ping_timeout_ms):
+ return
+ winrm, ssh = await asyncio.gather(
+ _port_open(ip, 5985, port_timeout_ms),
+ _port_open(ip, 22, port_timeout_ms),
+ )
+ if not winrm and not ssh:
+ return
+ hostname = await _resolve(ip)
+ hosts.append(
+ {
+ "hostname": hostname,
+ "ip": ip,
+ "os_guess": "windows" if winrm else ("linux" if ssh else None),
+ "winrm_open": winrm,
+ "ssh_open": ssh,
+ }
+ )
+
+ await asyncio.gather(*[probe(str(ip)) for ip in network.hosts()])
+ return sorted(hosts, key=lambda h: h["ip"])
diff --git a/satellite/satellite/ssh_exec.py b/satellite/satellite/ssh_exec.py
new file mode 100644
index 0000000..5c87b11
--- /dev/null
+++ b/satellite/satellite/ssh_exec.py
@@ -0,0 +1,84 @@
+"""SSH executor: run updates on Linux targets via asyncssh."""
+
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+
+from satellite.config import Credential
+
+
+class SSHError(Exception):
+ pass
+
+
+@dataclass
+class Target:
+ hostname: str
+ port: int = 22
+
+
+class SSHExecutor:
+ def __init__(self, target: Target, credential: Credential) -> None:
+ self.target = target
+ self.credential = credential
+
+ def _connect_kwargs(self) -> dict:
+ kwargs: dict = {
+ "host": self.target.hostname,
+ "port": self.target.port,
+ "known_hosts": None,
+ "connect_timeout": 30,
+ "username": self.credential.username,
+ }
+ if self.credential.password:
+ kwargs["password"] = self.credential.password
+ if self.credential.private_key_path:
+ kwargs["client_keys"] = [self.credential.private_key_path]
+ if self.credential.passphrase:
+ kwargs["passphrase"] = self.credential.passphrase
+ return kwargs
+
+ async def test_connection(self) -> tuple[bool, str]:
+ import asyncssh
+
+ try:
+ async with asyncssh.connect(**self._connect_kwargs()) as conn:
+ result = await conn.run("hostname", check=True)
+ return True, f"Verbunden mit {str(result.stdout).strip()}"
+ except Exception as exc: # noqa: BLE001
+ return False, str(exc)
+
+ async def stream_updates(self, reboot_if_required: bool = False) -> AsyncIterator[str]:
+ import asyncssh
+
+ async with asyncssh.connect(**self._connect_kwargs()) as conn:
+ pm = None
+ for candidate in ("apt-get", "dnf", "yum"):
+ result = await conn.run(f"command -v {candidate}", check=False)
+ if result.returncode == 0:
+ pm = candidate
+ break
+ if not pm:
+ raise SSHError("Kein unterstuetzter Paketmanager gefunden (apt/dnf/yum)")
+
+ yield f"Paketmanager: {pm}"
+ if pm == "apt-get":
+ cmd = "apt-get update && DEBIAN_FRONTEND=noninteractive apt-get upgrade -y"
+ else:
+ cmd = f"{pm} update -y"
+
+ password = self.credential.password or ""
+ full_cmd = f"echo '{password}' | sudo -S sh -c '{cmd}'"
+
+ async with conn.create_process(full_cmd) as process:
+ async for line in process.stdout: # type: ignore[union-attr]
+ yield str(line).rstrip()
+ await process.wait()
+ if process.returncode != 0:
+ raise SSHError(f"Update fehlgeschlagen (exit {process.returncode})")
+
+ if reboot_if_required:
+ async with asyncssh.connect(**self._connect_kwargs()) as conn:
+ check = await conn.run("test -f /var/run/reboot-required", check=False)
+ if check.returncode == 0:
+ yield "REBOOT erforderlich - wird ausgefuehrt..."
+ await conn.run(f"echo '{password}' | sudo -S reboot", check=False)
diff --git a/satellite/satellite/winrm_exec.py b/satellite/satellite/winrm_exec.py
new file mode 100644
index 0000000..e23937b
--- /dev/null
+++ b/satellite/satellite/winrm_exec.py
@@ -0,0 +1,96 @@
+"""WinRM executor: run PowerShell on Windows targets via pywinrm."""
+
+import asyncio
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+
+from satellite.config import Config, Credential
+
+
+class WinRMError(Exception):
+ pass
+
+
+@dataclass
+class Target:
+ hostname: str
+ port: int = 5985
+
+
+class WinRMExecutor:
+ def __init__(self, config: Config, target: Target, credential: Credential) -> None:
+ self.config = config
+ self.target = target
+ self.credential = credential
+
+ def _build_session(self): # type: ignore[no-untyped-def]
+ import winrm # pywinrm
+
+ scheme = "https" if self.target.port == 5986 else "http"
+ endpoint = f"{scheme}://{self.target.hostname}:{self.target.port}/wsman"
+ return winrm.Session(
+ endpoint,
+ auth=(self.credential.username, self.credential.password),
+ transport=self.credential.transport or self.config.winrm_transport,
+ server_cert_validation=self.config.winrm_cert_validation,
+ operation_timeout_sec=60,
+ read_timeout_sec=3600, # Windows Update kann lange laufen
+ )
+
+ async def run_powershell(self, script: str) -> str:
+ def _run() -> str:
+ session = self._build_session()
+ result = session.run_ps(script)
+ if result.status_code != 0:
+ err = result.std_err.decode(errors="replace").strip()
+ raise WinRMError(f"PowerShell exit {result.status_code}: {err}")
+ return result.std_out.decode(errors="replace")
+
+ return await asyncio.to_thread(_run)
+
+ async def stream_powershell(self, script: str) -> AsyncIterator[str]:
+ output = await self.run_powershell(script)
+ for line in output.splitlines():
+ yield line
+ await asyncio.sleep(0)
+
+ async def test_connection(self) -> tuple[bool, str]:
+ def _run() -> tuple[bool, str]:
+ try:
+ session = self._build_session()
+ result = session.run_ps("$env:COMPUTERNAME")
+ if result.status_code == 0:
+ name = result.std_out.decode(errors="replace").strip()
+ return True, f"Verbunden mit {name}"
+ return False, result.std_err.decode(errors="replace").strip()
+ except Exception as exc: # noqa: BLE001
+ return False, str(exc)
+
+ return await asyncio.to_thread(_run)
+
+ async def install_updates(self, reboot_if_required: bool = False) -> AsyncIterator[str]:
+ reboot_block = (
+ "if ($installResult.RebootRequired) { Write-Output 'REBOOT erforderlich - wird ausgefuehrt'; Restart-Computer -Force }"
+ if reboot_if_required
+ else "Write-Output \"RebootRequired: $($installResult.RebootRequired)\""
+ )
+ script = f"""
+ $session = New-Object -ComObject Microsoft.Update.Session
+ $searcher = $session.CreateUpdateSearcher()
+ $result = $searcher.Search("IsInstalled=0")
+ Write-Output "Gefundene Updates: $($result.Updates.Count)"
+ if ($result.Updates.Count -gt 0) {{
+ $toInstall = New-Object -ComObject Microsoft.Update.UpdateColl
+ $result.Updates | ForEach-Object {{
+ Write-Output " - $($_.Title)"
+ $toInstall.Add($_) | Out-Null
+ }}
+ $installer = $session.CreateUpdateInstaller()
+ $installer.Updates = $toInstall
+ $installResult = $installer.Install()
+ Write-Output "ResultCode: $($installResult.ResultCode)"
+ {reboot_block}
+ }}
+ """
+ async for line in self.stream_powershell(script):
+ yield line