"""Kali Linux VM icin SSH istemcisi. Tasarim notlari --------------- * Anahtar tabanli kimlik dogrulama (parola fallback yok, .env'de KALI_SSH_KEY zorunlu). Anahtar parolaliysa KALI_SSH_KEY_PASSPHRASE ile cozulur. * `run()` -> stdout, stderr, exit_code ve sureyi tek bir CommandResult'ta doner. * Context manager olarak kullanilabilir: with KaliSSH.from_config() as kali: res = kali.run("nmap -sV 10.0.0.1") * Bridged adapter senaryosunda IP degisirse sadece .env guncellenir. Bu dosya kendi basina test edilebilir; agent.py / tool_call.py bunu sarici olarak kullanacak. """ from __future__ import annotations import socket import time from dataclasses import dataclass from pathlib import Path from typing import Optional import paramiko from paramiko.ssh_exception import ( AuthenticationException, NoValidConnectionsError, SSHException, ) from config import SSHConfig # Logger sadece tip ipucu icin; runtime'da opsiyonel kalir (geri donus # importunu engellemek icin TYPE_CHECKING bloguna alindi). from typing import TYPE_CHECKING if TYPE_CHECKING: from logger import SessionLogger # Paramiko'nun denedigi anahtar tipleri (parola cozumu icin sirayla denenir). _KEY_LOADERS = ( paramiko.Ed25519Key, paramiko.ECDSAKey, paramiko.RSAKey, ) @dataclass class CommandResult: """Bir SSH komutunun calisma sonucu.""" command: str stdout: str stderr: str exit_code: int duration_s: float @property def ok(self) -> bool: return self.exit_code == 0 def combined_output(self, max_chars: int = 8000) -> str: """Modele geri verilecek tek string. Cok uzun ciktilar kirpilir.""" parts: list[str] = [] if self.stdout: parts.append(self.stdout.rstrip()) if self.stderr: parts.append(f"[stderr]\n{self.stderr.rstrip()}") text = "\n".join(parts) if parts else "(cikti yok)" if len(text) > max_chars: text = text[:max_chars] + f"\n...[CIKTI {len(text) - max_chars} KARAKTER KIRPILDI]" return text class SSHConnectionError(RuntimeError): """Bag­lan­ti / kimlik dogrulama hatalarini sarmalayan tek tip hata.""" def _load_private_key(key_path: Path, passphrase: Optional[str]) -> paramiko.PKey: """Anahtari tip otomatik tespit ederek yukle.""" if not key_path.exists(): raise SSHConnectionError(f"SSH anahtari bulunamadi: {key_path}") last_error: Exception | None = None for loader in _KEY_LOADERS: try: return loader.from_private_key_file(str(key_path), password=passphrase) except paramiko.PasswordRequiredException as exc: raise SSHConnectionError( "SSH anahtari parolali; .env icinde KALI_SSH_KEY_PASSPHRASE bos." ) from exc except SSHException as exc: last_error = exc continue raise SSHConnectionError( f"SSH anahtari yuklenemedi ({key_path}). Son hata: {last_error}" ) class KaliSSH: """Kali VM'e ince SSH sarmali.""" def __init__(self, cfg: SSHConfig, logger: Optional["SessionLogger"] = None): self.cfg = cfg self._client: Optional[paramiko.SSHClient] = None self.logger = logger # --- yasam dongusu -------------------------------------------------- @classmethod def from_config(cls, logger: Optional["SessionLogger"] = None) -> "KaliSSH": return cls(SSHConfig.from_env(), logger=logger) def attach_logger(self, logger: "SessionLogger") -> None: """Olusturulduktan sonra logger baglamak icin.""" self.logger = logger def __enter__(self) -> "KaliSSH": self.connect() return self def __exit__(self, exc_type, exc, tb) -> None: self.close() @property def is_connected(self) -> bool: if self._client is None: return False transport = self._client.get_transport() return transport is not None and transport.is_active() def connect(self) -> None: if self.is_connected: return pkey = _load_private_key(self.cfg.key_path, self.cfg.key_passphrase) client = paramiko.SSHClient() # Bilinmeyen host icin otomatik kabul (lokal VM senaryosu icin uygun; # production'da known_hosts kullanilmali). client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect( hostname=self.cfg.host, port=self.cfg.port, username=self.cfg.username, pkey=pkey, password=self.cfg.password, timeout=15, allow_agent=False, look_for_keys=False, ) except AuthenticationException as exc: if self.logger: self.logger.error( "ssh_auth_failed", host=self.cfg.host, user=self.cfg.username ) raise SSHConnectionError( f"Kimlik dogrulama basarisiz: {self.cfg.username}@{self.cfg.host}. " "Anahtarin Kali tarafindaki ~/.ssh/authorized_keys icinde oldugundan emin ol " "(veya gecici olarak .env icinde KALI_SSH_PASSWORD doldur)." ) from exc except NoValidConnectionsError as exc: if self.logger: self.logger.error( "ssh_connect_failed", host=self.cfg.host, port=self.cfg.port ) raise SSHConnectionError( f"Baglanti kurulamadi: {self.cfg.host}:{self.cfg.port}. " "Kali VM ayakta mi, sshd calisiyor mu, IP/port dogru mu?" ) from exc except (socket.timeout, TimeoutError) as exc: if self.logger: self.logger.error( "ssh_connect_timeout", host=self.cfg.host, port=self.cfg.port ) raise SSHConnectionError( f"Zaman asimi: {self.cfg.host}:{self.cfg.port}" ) from exc except SSHException as exc: if self.logger: self.logger.error("ssh_error", detail=str(exc)) raise SSHConnectionError(f"SSH hatasi: {exc}") from exc self._client = client if self.logger: self.logger.info( "ssh_connected", host=self.cfg.host, port=self.cfg.port, user=self.cfg.username, ) def close(self) -> None: if self._client is not None: try: self._client.close() finally: self._client = None # --- komut calistirma ---------------------------------------------- def run( self, command: str, *, timeout: Optional[int] = None, get_pty: bool = False, ) -> CommandResult: """Komutu Kali'de calistir, sonucu CommandResult olarak don. timeout None ise SSHConfig.command_timeout kullanilir. get_pty=True bazi tool'lar (ornek: msfconsole) icin gerekebilir; ancak ciktinin renk kodlari ile karismasina yol acar, varsayilan False. """ if not self.is_connected: self.connect() assert self._client is not None # type checker icin effective_timeout = timeout if timeout is not None else self.cfg.command_timeout start = time.monotonic() try: stdin, stdout, stderr = self._client.exec_command( command, timeout=effective_timeout, get_pty=get_pty, ) stdin.close() # exit_status_ready'i beklemeden read yaparsak deadlock olabilir; # paramiko'nun standart akisi: once read, exit_status sona. out = stdout.read().decode("utf-8", errors="replace") err = stderr.read().decode("utf-8", errors="replace") exit_code = stdout.channel.recv_exit_status() except socket.timeout as exc: if self.logger: self.logger.error( "command_timeout", command=command, timeout_s=effective_timeout ) raise SSHConnectionError( f"Komut zaman asimina ugradi ({effective_timeout}s): {command!r}" ) from exc except SSHException as exc: if self.logger: self.logger.error( "command_ssh_error", command=command, detail=str(exc) ) raise SSHConnectionError(f"Komut calistirma hatasi: {exc}") from exc result = CommandResult( command=command, stdout=out, stderr=err, exit_code=exit_code, duration_s=time.monotonic() - start, ) if self.logger: self.logger.command(result) return result def ping(self) -> bool: """Hizli saglik kontrolu. True donerse SSH kanali calisiyor demektir.""" try: res = self.run("echo pong", timeout=10) return res.ok and "pong" in res.stdout except SSHConnectionError: return False