131 lines
5.0 KiB
Python
131 lines
5.0 KiB
Python
"""
|
|
Serverseitiger SSH-Verbindungsaufbau (asyncssh).
|
|
|
|
Zentrales Sicherheitsprinzip (Konzept 4.2/4.4/6.4): der private Schluessel
|
|
wird pro Verbindung aus der DB geladen, entschluesselt, an asyncssh
|
|
uebergeben und danach nicht weiter referenziert -- er verlaesst den
|
|
Serverprozess nie und wird nicht geloggt. Strict Host Key Checking ist
|
|
Pflicht: ohne gepinnten Fingerprint wird die Verbindung abgelehnt.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import asyncssh
|
|
import aiosqlite
|
|
|
|
from app.security.crypto import decrypt_secret
|
|
|
|
logger = logging.getLogger("jumphost.ssh_proxy")
|
|
|
|
|
|
class HostNotConfiguredError(Exception):
|
|
pass
|
|
|
|
|
|
class HostKeyMismatchError(Exception):
|
|
def __init__(self, expected: str | None, observed: str | None) -> None:
|
|
self.expected = expected
|
|
self.observed = observed
|
|
super().__init__(f"Host-Key-Mismatch: erwartet={expected!r} beobachtet={observed!r}")
|
|
|
|
|
|
class _PinnedHostKeyClient(asyncssh.SSHClient):
|
|
"""Erzwingt Strict Host Key Checking gegen einen fest hinterlegten
|
|
SHA-256-Fingerprint. Kein automatisches Trust-on-First-Use (TOFU)."""
|
|
|
|
def __init__(self, expected_fingerprint: str | None, *, discovery_mode: bool = False) -> None:
|
|
self.expected_fingerprint = expected_fingerprint
|
|
self.discovery_mode = discovery_mode
|
|
self.observed_fingerprint: str | None = None
|
|
|
|
def validate_host_public_key(self, host, addr, port, key) -> bool: # noqa: D102
|
|
self.observed_fingerprint = key.get_fingerprint("sha256")
|
|
if self.discovery_mode:
|
|
# Nur ueber den expliziten Admin-Discovery-Endpunkt erreichbar,
|
|
# niemals im regulaeren Verbindungspfad (siehe admin/routes.py).
|
|
return True
|
|
if not self.expected_fingerprint:
|
|
return False
|
|
return self.observed_fingerprint == self.expected_fingerprint
|
|
|
|
|
|
async def load_host(conn: aiosqlite.Connection, host_id: int) -> dict:
|
|
cursor = await conn.execute(
|
|
"SELECT id, hostname, address, port, os_type, protocol, ssh_host_key_fingerprint, "
|
|
"ssh_username, file_transfer_enabled, host_group_id FROM hosts WHERE id = ? AND is_active = 1",
|
|
(host_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
if row is None:
|
|
raise HostNotConfiguredError(f"Host {host_id} nicht gefunden oder inaktiv")
|
|
keys = (
|
|
"id", "hostname", "address", "port", "os_type", "protocol",
|
|
"ssh_host_key_fingerprint", "ssh_username", "file_transfer_enabled", "host_group_id",
|
|
)
|
|
return dict(zip(keys, row))
|
|
|
|
|
|
async def load_private_key_for_host(conn: aiosqlite.Connection, host_id: int) -> asyncssh.SSHKey:
|
|
cursor = await conn.execute(
|
|
"SELECT sk.private_key_enc FROM ssh_keys sk "
|
|
"JOIN host_ssh_key_map m ON m.ssh_key_id = sk.id "
|
|
"WHERE m.host_id = ? LIMIT 1",
|
|
(host_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
if row is None:
|
|
raise HostNotConfiguredError(f"Kein SSH-Schluessel fuer Host {host_id} hinterlegt")
|
|
pem = decrypt_secret(row[0], associated_data=b"ssh_private_key")
|
|
try:
|
|
return asyncssh.import_private_key(pem)
|
|
finally:
|
|
# Bestpraxis: Referenz auf den Klartext-PEM-Bytes so schnell wie moeglich loslassen.
|
|
del pem
|
|
|
|
|
|
async def connect_to_host(conn: aiosqlite.Connection, host_id: int) -> asyncssh.SSHClientConnection:
|
|
host = await load_host(conn, host_id)
|
|
if host["protocol"] != "ssh":
|
|
raise HostNotConfiguredError("Host ist kein SSH-Ziel")
|
|
|
|
private_key = await load_private_key_for_host(conn, host_id)
|
|
client_factory = lambda: _PinnedHostKeyClient(host["ssh_host_key_fingerprint"])
|
|
|
|
try:
|
|
connection = await asyncssh.connect(
|
|
host["address"],
|
|
port=host["port"],
|
|
username=host["ssh_username"],
|
|
client_keys=[private_key],
|
|
known_hosts=None, # Validierung erfolgt ausschliesslich ueber validate_host_public_key
|
|
client_factory=client_factory,
|
|
connect_timeout=10,
|
|
)
|
|
except asyncssh.Error as exc:
|
|
logger.warning("SSH-Verbindungsfehler zu Host %s: %s", host_id, exc)
|
|
raise
|
|
return connection
|
|
|
|
|
|
async def discover_and_store_host_key(
|
|
conn: aiosqlite.Connection, host_id: int, *, admin_user_id: int
|
|
) -> str:
|
|
"""Verbindet EINMALIG ohne Pinning, um den Host-Key-Fingerprint zu erfassen
|
|
und in der DB zu hinterlegen. Nur ueber einen dedizierten, admin-only
|
|
Endpunkt aufrufbar -- jeder Aufruf ist eine bewusste Vertrauensentscheidung
|
|
und wird im Audit-Log als solche vermerkt (siehe admin/routes.py)."""
|
|
host = await load_host(conn, host_id)
|
|
client = _PinnedHostKeyClient(None, discovery_mode=True)
|
|
connection = await asyncssh.connect(
|
|
host["address"], port=host["port"], username=host["ssh_username"],
|
|
known_hosts=None, client_factory=lambda: client, connect_timeout=10,
|
|
)
|
|
connection.close()
|
|
fingerprint = client.observed_fingerprint
|
|
await conn.execute(
|
|
"UPDATE hosts SET ssh_host_key_fingerprint = ? WHERE id = ?", (fingerprint, host_id)
|
|
)
|
|
await conn.commit()
|
|
return fingerprint
|