Paramiko ile SSH Bağlantısı ve Uzak Komut Çalıştırma

Uzak sunucuları yönetmek, onlarca farklı makineye bağlanıp aynı komutları tek tek çalıştırmak… Bunu manuel yapıyorsan zaten bir şeyler yanlış gidiyor demektir. Python’un paramiko kütüphanesi tam da bu noktada devreye giriyor ve SSH üzerinden uzak sunucu yönetimini otomatize etmeni sağlıyor. Hem sysadminler hem de DevOps mühendisleri için vazgeçilmez bir araç haline gelmiş olan paramiko’yu bu yazıda en ince ayrıntısına kadar ele alacağız.

Paramiko Nedir ve Neden Kullanmalısın?

Paramiko, Python için geliştirilmiş bir SSH2 protokol kütüphanesidir. Saf Python ile yazılmıştır ve SSH bağlantısı kurma, dosya transferi (SFTP), tünel oluşturma gibi işlemleri programatik olarak yapmanı sağlar.

Alternatif olarak subprocess ile ssh komutu çağırabilirsin, ama bu yöntem taşınabilir değildir ve Windows’ta iş görmez. fabric kütüphanesi paramiko’nun üzerine kurulmuştur, yani temeli anlamak her iki araçta da sana avantaj sağlar. Paramiko’nun avantajları şöyle sıralanabilir:

• Platform bağımsızlığı: Windows, Linux ve macOS üzerinde aynı şekilde çalışır
• Şifre ve anahtar desteği: Hem password hem de SSH key authentication destekler
• SFTP dahil: Ayrı bir kütüphaneye gerek kalmadan dosya transferi yapabilirsin
• Gelişmiş kontrol: Timeout, banner, host key kontrolü gibi ince ayarlar mümkündür
• Aktif topluluk: Sürekli güncellenen ve yaygın kullanılan bir kütüphanedir

Kurulum ve Ortam Hazırlığı

Başlamadan önce sanal ortam oluşturmak her zaman iyi bir pratiktir:

# Sanal ortam oluştur
python3 -m venv ssh-otomasyon
source ssh-otomasyon/bin/activate  # Linux/macOS
# ssh-otomasyonScriptsactivate  # Windows

# Paramiko'yu kur
pip install paramiko

# Versiyon kontrolü
python -c "import paramiko; print(paramiko.__version__)"

Bazı sistemlerde cryptography paketi de gerekebilir, paramiko bunu bağımlılık olarak zaten çeker ama sorun yaşarsan:

pip install paramiko cryptography

İlk SSH Bağlantısı: Temel Kullanım

En basit haliyle bir SSH bağlantısı kurup komut çalıştıralım. Bu örnek, bir web sunucusunun uptime bilgisini çekiyor:

# baglanti_temel.py

import paramiko
import sys

def uzak_komut_calistir(host, port, kullanici, sifre, komut):
    """Uzak sunucuda komut çalıştırır ve sonucu döner."""
    
    ssh = paramiko.SSHClient()
    
    # Host key doğrulamasını ayarla
    # Üretim ortamında AutoAddPolicy kullanmaktan kaçın!
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        ssh.connect(
            hostname=host,
            port=port,
            username=kullanici,
            password=sifre,
            timeout=10
        )
        
        stdin, stdout, stderr = ssh.exec_command(komut)
        
        cikti = stdout.read().decode('utf-8').strip()
        hata = stderr.read().decode('utf-8').strip()
        
        if hata:
            print(f"HATA: {hata}", file=sys.stderr)
        
        return cikti
        
    except paramiko.AuthenticationException:
        print("Kimlik doğrulama başarısız!")
        return None
    except paramiko.SSHException as e:
        print(f"SSH hatası: {e}")
        return None
    finally:
        ssh.close()

# Kullanım örneği
sonuc = uzak_komut_calistir(
    host="192.168.1.50",
    port=22,
    kullanici="admin",
    sifre="gizli_sifre",
    komut="uptime && df -h /"
)

if sonuc:
    print(sonuc)

SSH Key ile Bağlantı: Güvenli Yöntem

Üretim ortamında şifre yerine SSH key kullanmak hem daha güvenli hem de otomasyon için çok daha pratiktir. Script’lere şifre gömmek ciddi bir güvenlik açığıdır:

# key_baglanti.py

import paramiko
import os

def key_ile_baglan(host, kullanici, key_dosyasi=None, key_sifresi=None):
    """
    SSH private key ile bağlantı kurar.
    key_dosyasi belirtilmezse ~/.ssh/id_rsa kullanılır.
    """
    
    ssh = paramiko.SSHClient()
    
    # Bilinen host'ları sistem dosyasından yükle
    ssh.load_system_host_keys()
    
    # Yeni host'lar için politika belirle
    ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
    # Güvenli ortamlarda AutoAddPolicy yerine WarningPolicy de kullanılabilir
    
    if key_dosyasi is None:
        key_dosyasi = os.path.expanduser("~/.ssh/id_rsa")
    
    try:
        # Private key'i yükle
        if key_sifresi:
            pkey = paramiko.RSAKey.from_private_key_file(
                key_dosyasi, 
                password=key_sifresi
            )
        else:
            pkey = paramiko.RSAKey.from_private_key_file(key_dosyasi)
        
        ssh.connect(
            hostname=host,
            username=kullanici,
            pkey=pkey,
            timeout=15
        )
        
        print(f"[OK] {host} bağlantısı başarılı")
        return ssh
        
    except FileNotFoundError:
        print(f"Key dosyası bulunamadı: {key_dosyasi}")
        return None
    except paramiko.SSHException as e:
        print(f"SSH key hatası: {e}")
        return None

# Kullanım
ssh_client = key_ile_baglan("web01.sirket.com", "deploy")
if ssh_client:
    stdin, stdout, stderr = ssh_client.exec_command("hostname -f")
    print(stdout.read().decode())
    ssh_client.close()

Gerçek Dünya Senaryosu 1: Çoklu Sunucu Yönetimi

Diyelim ki 20 web sunucusunun disk kullanımını kontrol etmen gerekiyor. Manuel yapmak yerine şu script’i kullan:

# disk_kontrol.py
# Tüm sunucularda disk kullanımını kontrol eder ve uyarı verir

import paramiko
import json
from datetime import datetime

SUNUCULAR = [
    {"host": "web01.sirket.com", "port": 22, "kullanici": "admin"},
    {"host": "web02.sirket.com", "port": 22, "kullanici": "admin"},
    {"host": "db01.sirket.com",  "port": 2222, "kullanici": "dbadmin"},
    {"host": "cache01.sirket.com", "port": 22, "kullanici": "admin"},
]

ESIK_YUZDE = 80  # Bu değerin üzerindeyse uyarı ver
KEY_DOSYASI = "/home/monitor/.ssh/id_ed25519"

def disk_kullanimi_al(ssh):
    """df çıktısını parse eder, disk kullanım yüzdelerini döner."""
    komut = "df -h --output=source,pcent,target | grep -v tmpfs | tail -n+2"
    stdin, stdout, stderr = ssh.exec_command(komut)
    
    diskler = []
    for satir in stdout.read().decode().strip().split('n'):
        parcalar = satir.split()
        if len(parcalar) >= 3:
            yuzde = int(parcalar[1].replace('%', ''))
            diskler.append({
                "cihaz": parcalar[0],
                "kullanim": yuzde,
                "mount": parcalar[2]
            })
    return diskler

def tum_sunuculari_kontrol_et():
    rapor = {
        "tarih": datetime.now().isoformat(),
        "sunucular": [],
        "uyarilar": []
    }
    
    for sunucu in SUNUCULAR:
        ssh = paramiko.SSHClient()
        ssh.load_system_host_keys()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        try:
            pkey = paramiko.Ed25519Key.from_private_key_file(KEY_DOSYASI)
            ssh.connect(
                hostname=sunucu["host"],
                port=sunucu["port"],
                username=sunucu["kullanici"],
                pkey=pkey,
                timeout=10
            )
            
            diskler = disk_kullanimi_al(ssh)
            sunucu_raporu = {
                "host": sunucu["host"],
                "durum": "OK",
                "diskler": diskler
            }
            
            for disk in diskler:
                if disk["kullanim"] >= ESIK_YUZDE:
                    uyari = f"UYARI: {sunucu['host']} - {disk['mount']} - %{disk['kullanim']} dolu!"
                    rapor["uyarilar"].append(uyari)
                    print(uyari)
                    sunucu_raporu["durum"] = "UYARI"
            
            rapor["sunucular"].append(sunucu_raporu)
            
        except Exception as e:
            hata = {"host": sunucu["host"], "durum": "HATA", "mesaj": str(e)}
            rapor["sunucular"].append(hata)
            print(f"HATA: {sunucu['host']} - {e}")
        finally:
            ssh.close()
    
    return rapor

if __name__ == "__main__":
    rapor = tum_sunuculari_kontrol_et()
    
    # Raporu JSON olarak kaydet
    with open(f"disk_rapor_{datetime.now().strftime('%Y%m%d_%H%M')}.json", "w") as f:
        json.dump(rapor, f, indent=2, ensure_ascii=False)
    
    print(f"nToplam {len(rapor['uyarilar'])} uyarı var.")

SFTP ile Dosya Transferi

Paramiko’nun SFTP modülü, uzak sunuculara dosya yükleme ve indirme işlemlerini kolaylaştırır:

# sftp_transfer.py

import paramiko
import os
from pathlib import Path

class SFTPYonetici:
    def __init__(self, host, kullanici, key_dosyasi, port=22):
        self.host = host
        self.kullanici = kullanici
        self.key_dosyasi = key_dosyasi
        self.port = port
        self.ssh = None
        self.sftp = None
    
    def baglan(self):
        self.ssh = paramiko.SSHClient()
        self.ssh.load_system_host_keys()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        pkey = paramiko.RSAKey.from_private_key_file(self.key_dosyasi)
        self.ssh.connect(
            hostname=self.host,
            port=self.port,
            username=self.kullanici,
            pkey=pkey
        )
        self.sftp = self.ssh.open_sftp()
        return self
    
    def yukle(self, yerel_dosya, uzak_yol):
        """Yerel dosyayı uzak sunucuya yükler."""
        dosya_adi = Path(yerel_dosya).name
        uzak_tam_yol = f"{uzak_yol}/{dosya_adi}"
        
        print(f"Yükleniyor: {yerel_dosya} -> {self.host}:{uzak_tam_yol}")
        self.sftp.put(yerel_dosya, uzak_tam_yol)
        print(f"[OK] Yükleme tamamlandı")
        return uzak_tam_yol
    
    def indir(self, uzak_dosya, yerel_yol):
        """Uzak sunucudan dosya indirir."""
        dosya_adi = os.path.basename(uzak_dosya)
        yerel_tam_yol = os.path.join(yerel_yol, dosya_adi)
        
        print(f"İndiriliyor: {self.host}:{uzak_dosya} -> {yerel_tam_yol}")
        self.sftp.get(uzak_dosya, yerel_tam_yol)
        print(f"[OK] İndirme tamamlandı")
        return yerel_tam_yol
    
    def dizin_listele(self, uzak_yol):
        """Uzak dizindeki dosyaları listeler."""
        return self.sftp.listdir_attr(uzak_yol)
    
    def kapat(self):
        if self.sftp:
            self.sftp.close()
        if self.ssh:
            self.ssh.close()
    
    def __enter__(self):
        return self.baglan()
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.kapat()

# Context manager ile kullanım
with SFTPYonetici("web01.sirket.com", "deploy", "~/.ssh/id_rsa") as sftp:
    # Config dosyası yükle
    sftp.yukle("/etc/nginx/nginx.conf", "/tmp/yedek")
    
    # Log dosyası indir
    sftp.indir("/var/log/nginx/error.log", "/tmp/loglar")
    
    # Dizin listele
    dosyalar = sftp.dizin_listele("/var/www/html")
    for dosya in dosyalar:
        print(f"{dosya.filename} - {dosya.st_size} bytes")

Gerçek Dünya Senaryosu 2: Otomatik Deployment Script’i

Bir web uygulamasını birden fazla sunucuya deploy eden gerçekçi bir örnek:

# deployment.py
# Basit bir web app deployment otomasyonu

import paramiko
import time
import sys

class Deployer:
    def __init__(self, sunucular, kullanici, key_dosyasi):
        self.sunucular = sunucular
        self.kullanici = kullanici
        self.key_dosyasi = key_dosyasi
        self.basarili = []
        self.basarisiz = []
    
    def _baglan(self, host, port=22):
        ssh = paramiko.SSHClient()
        ssh.load_system_host_keys()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        pkey = paramiko.RSAKey.from_private_key_file(self.key_dosyasi)
        ssh.connect(hostname=host, port=port, username=self.kullanici, pkey=pkey, timeout=15)
        return ssh
    
    def _komut_calistir(self, ssh, komut, timeout=60):
        """Komutu çalıştırır, exit code'u kontrol eder."""
        stdin, stdout, stderr = ssh.exec_command(komut)
        
        # Exit code için kanalı bekle
        exit_status = stdout.channel.recv_exit_status()
        
        cikti = stdout.read().decode('utf-8').strip()
        hata = stderr.read().decode('utf-8').strip()
        
        return exit_status, cikti, hata
    
    def sunucu_deploy(self, host, port, uygulama_yolu, git_branch="main"):
        print(f"n{'='*50}")
        print(f"Sunucu: {host}")
        print(f"{'='*50}")
        
        ssh = None
        try:
            ssh = self._baglan(host, port)
            
            adimlar = [
                ("Dizine geç", f"cd {uygulama_yolu}"),
                ("Git pull", f"cd {uygulama_yolu} && git pull origin {git_branch}"),
                ("Bağımlılıkları yükle", f"cd {uygulama_yolu} && pip install -r requirements.txt -q"),
                ("Migration çalıştır", f"cd {uygulama_yolu} && python manage.py migrate --no-input"),
                ("Static dosyaları topla", f"cd {uygulama_yolu} && python manage.py collectstatic --no-input"),
                ("Servisi yeniden başlat", "sudo systemctl restart gunicorn"),
                ("Servis durumunu kontrol et", "systemctl is-active gunicorn"),
            ]
            
            for adim_adi, komut in adimlar:
                print(f"  --> {adim_adi}...", end=" ", flush=True)
                exit_code, cikti, hata = self._komut_calistir(ssh, komut)
                
                if exit_code == 0:
                    print("OK")
                else:
                    print(f"HATA (exit: {exit_code})")
                    if hata:
                        print(f"      {hata[:200]}")
                    raise Exception(f"Adım başarısız: {adim_adi}")
            
            self.basarili.append(host)
            print(f"[BASARILI] {host} deploy tamamlandı")
            
        except Exception as e:
            self.basarisiz.append({"host": host, "hata": str(e)})
            print(f"[BASARISIZ] {host}: {e}")
        finally:
            if ssh:
                ssh.close()
    
    def deploy_baslat(self, uygulama_yolu, git_branch="main"):
        print(f"Deployment başlıyor - {len(self.sunucular)} sunucu")
        baslangic = time.time()
        
        for sunucu in self.sunucular:
            self.sunucu_deploy(
                sunucu["host"], 
                sunucu.get("port", 22),
                uygulama_yolu,
                git_branch
            )
        
        gecen_sure = time.time() - baslangic
        
        print(f"n{'='*50}")
        print(f"DEPLOYMENT ÖZETI")
        print(f"Süre: {gecen_sure:.1f} saniye")
        print(f"Başarılı: {len(self.basarili)} sunucu")
        print(f"Başarısız: {len(self.basarisiz)} sunucu")
        
        if self.basarisiz:
            print("nHatalı sunucular:")
            for item in self.basarisiz:
                print(f"  - {item['host']}: {item['hata']}")
            sys.exit(1)

# Kullanım
sunucular = [
    {"host": "app01.sirket.com", "port": 22},
    {"host": "app02.sirket.com", "port": 22},
    {"host": "app03.sirket.com", "port": 22},
]

deployer = Deployer(
    sunucular=sunucular,
    kullanici="deploy",
    key_dosyasi="/home/ci/.ssh/id_ed25519"
)

deployer.deploy_baslat(
    uygulama_yolu="/var/www/myapp",
    git_branch="main"
)

İnteraktif Shell ve PTY Kullanımı

Bazı komutlar (özellikle sudo gerektiren veya interaktif çıktı veren) pseudo-terminal (PTY) gerektirir:

# pty_kullanim.py

import paramiko
import time

def sudo_komut_calistir(ssh, komut, sudo_sifre):
    """
    Sudo gerektiren komutları çalıştırmak için PTY kullanır.
    Mümkünse bunun yerine sudoers'a NOPASSWD eklemek daha temizdir.
    """
    
    # PTY ile kanal aç
    kanal = ssh.get_transport().open_session()
    kanal.get_pty()
    kanal.exec_command(f"sudo -S {komut}")
    
    # Sudo şifre prompt'unu bekle
    time.sleep(0.5)
    
    # Şifreyi gönder
    kanal.send(f"{sudo_sifre}n")
    
    # Çıktıyı oku
    cikti = b""
    while True:
        if kanal.recv_ready():
            cikti += kanal.recv(1024)
        if kanal.exit_status_ready():
            break
        time.sleep(0.1)
    
    exit_code = kanal.recv_exit_status()
    kanal.close()
    
    return exit_code, cikti.decode('utf-8')

# Örnek: Servis yeniden başlatma
ssh = paramiko.SSHClient()
ssh.load_system_host_keys()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("sunucu.com", username="admin", key_filename="~/.ssh/id_rsa")

exit_code, cikti = sudo_komut_calistir(ssh, "systemctl restart nginx", "sudo_sifrem")
print(f"Exit code: {exit_code}")
print(f"Çıktı: {cikti}")

ssh.close()

Önemli Güvenlik Notları

Paramiko kullanırken dikkat etmen gereken güvenlik konuları:

• AutoAddPolicy kullanımı: AutoAddPolicy() Man-in-the-Middle saldırılarına karşı savunmasızdır. Üretim ortamında known_hosts dosyasını kullan ve RejectPolicy() veya WarningPolicy() tercih et
• Şifreleri kaynak koduna gömme: Şifreleri asla direkt koda yazma. os.environ.get(), python-dotenv, HashiCorp Vault veya benzeri araçlar kullan
• SSH key izinleri: Private key dosyaları chmod 600 olmalı, yoksa paramiko bağlantıyı reddedebilir
• Timeout değerleri: Her zaman timeout belirt, yoksa script sonsuza kadar asılı kalabilir
• Logging: Tüm bağlantı ve komut loglarını kaydet, hem güvenlik hem de debug için kritik
• Ed25519 key kullanımı: RSA yerine Ed25519 key’leri tercih et, daha modern ve güvenlidir

Bağlantı Havuzu ve Performans

Çok sayıda sunucuya bağlanırken sıralı işlem yerine paralel bağlantı kullanabilirsin:

# paralel_baglanti.py

import paramiko
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading

# Thread-safe print için lock
print_lock = threading.Lock()

def guvenli_print(*args, **kwargs):
    with print_lock:
        print(*args, **kwargs)

def sunucuda_calistir(sunucu_bilgi):
    host = sunucu_bilgi["host"]
    komut = sunucu_bilgi["komut"]
    key_dosyasi = sunucu_bilgi.get("key", "~/.ssh/id_rsa")
    
    ssh = paramiko.SSHClient()
    ssh.load_system_host_keys()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        pkey = paramiko.RSAKey.from_private_key_file(key_dosyasi)
        ssh.connect(hostname=host, username="admin", pkey=pkey, timeout=10)
        
        stdin, stdout, stderr = ssh.exec_command(komut)
        stdout.channel.recv_exit_status()
        
        cikti = stdout.read().decode().strip()
        guvenli_print(f"[{host}] {cikti}")
        
        return {"host": host, "durum": "OK", "cikti": cikti}
        
    except Exception as e:
        guvenli_print(f"[{host}] HATA: {e}")
        return {"host": host, "durum": "HATA", "mesaj": str(e)}
    finally:
        ssh.close()

# Paralel çalıştırma
sunucular = [
    {"host": f"web{i:02d}.sirket.com", "komut": "uptime", "key": "~/.ssh/id_ed25519"}
    for i in range(1, 11)
]

sonuclar = []
# max_workers değerini sunucu sayısına ve ağ kapasitesine göre ayarla
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(sunucuda_calistir, s): s for s in sunucular}
    
    for future in as_completed(futures):
        sonuc = future.result()
        sonuclar.append(sonuc)

print(f"nToplam: {len(sonuclar)} işlem tamamlandı")
basarili = sum(1 for s in sonuclar if s["durum"] == "OK")
print(f"Başarılı: {basarili}, Başarısız: {len(sonuclar) - basarili}")

Hata Yönetimi ve Yeniden Deneme Mantığı

Ağ ortamında geçici hatalar kaçınılmazdır. Basit bir retry mekanizması eklemek script’lerini çok daha sağlam hale getirir:

• paramiko.AuthenticationException: Yanlış şifre veya key sorunu
• paramiko.NoValidConnectionsError: Sunucuya ulaşılamadı
• paramiko.SSHException: Genel SSH protokol hatası
• socket.timeout: Bağlantı zaman aşımı
• EOFError: Bağlantı beklenmedik şekilde kapandı

Bu hataları yakalayıp uygun şekilde loglamak, production scriptlerinde gece yarısı sizi uyandırmayacak sağlıklı bir otomasyon altyapısının temelidir.

Sonuç

Paramiko, Python ile SSH otomasyonu için gerçekten güçlü ve esnek bir araç. Temel kullanımdan başlayıp SFTP, paralel bağlantı ve deployment pipeline’larına kadar uzanan geniş bir kullanım alanı var.

Birkaç önemli noktayı tekrar vurgulayalım: Üretim ortamında şifreler yerine SSH key kullan, AutoAddPolicy’yi dikkatli kullan, her bağlantıya timeout ekle ve tüm işlemleri logla. Paralel bağlantı kullanırken sunucu sayısına göre max_workers değerini makul tut, 50 sunucuya aynı anda bağlanmaya çalışmak ağı zorlayabilir.

Paramiko’yu öğrendikten sonra sıradaki adım fabric kütüphanesi olabilir. Fabric, paramiko’nun üzerine daha yüksek seviye bir arayüz sunar ve özellikle deployment senaryoları için işleri daha da kolaylaştırır. Ama temeli anlamak her zaman önce gelir, fabric altında paramiko var ve bir şeyler ters gittiğinde ne aradığını bilmek büyük avantaj sağlar.

Bu script’leri cron job’larla, CI/CD pipeline’larıyla veya monitoring sistemlerinizle entegre ederek tam bir otomasyon altyapısı kurabilirsin. Onlarca sunucuyu tek bir Python script’iyle yönetmek, işin tadını çıkarmanın en güzel yollarından biri.