| """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
|
|
|
|
|
|
|
| from typing import TYPE_CHECKING
|
|
|
| if TYPE_CHECKING:
|
| from logger import SessionLogger
|
|
|
|
|
|
|
|
|
| _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):
|
| """Baglanti / 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
|
|
|
|
|
| @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()
|
|
|
|
|
| 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
|
|
|
|
|
| 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
|
|
|
| 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()
|
|
|
|
|
| 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
|
|
|
