ssh_client.py

"""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