본문 바로가기
카테고리 없음

[Python] paramiko 활용 SFTP 연결 예시

by 쿡노트 2024. 3. 19.
반응형

Client에서, 원격 서버로 보안 파일 전송 프로토콜(Secure File Transfer Protocol, SFTP) 로 간단하게 파일 다운로드하는 Python 예제를 작성하겠습니다.

 

테스트 환경

Clinet : Windows11, python 3.9.13, paramiko 3.4.0
Server : Linux

 

환경 구성
Visual Studio Code에서 작업을 했고, 설치 및 사용법은 생략하겠습니다.

SFTP 접속 방식 종류

1. ID, Password 접속 (소스코드에 직접 기재 시 계정 노출 우려가 있다. 또는 패스워드 변경 시 소스 코드 변경)
2. SSH 인증키를 활용한 접속

Python3 설치 여부 확인 (Visual Studio Code 하단 터미널 또는 cmd에서 확인 가능)

python3 --version

python 설치가 되어 있지 않으면, 하단 사이트에서 필요한 버전 다운로드하여서 설치 가능
https://www.python.org/downloads/

paramiko 설치 (Visual Studio Code 하단 터미널 또는 cmd 접속)

pip3 install paramiko
pip3 list

코드 예시

라이브러리 import

import logging
import paramiko #python paramiko 라이브러리 import

 

초기화(패스워드 인증, SSH 인증키 두 가지 테스트 목적으로 작성되었다.)

def __init__(self, hostname, port, username, password, private_key_path):
        self.hostname = hostname
        self.port = port
        self.username = username
        self.password = password #Password 접속 초기화
        self.private_key_path = private_key_path #SSH 인증키 접속 초기화
        self.ssh = None
        self.sftp = None
        logging.debug("Init SFTP")  

 

ID, Password 인증 접속

#ID, Password 인증
    def password_connect(self):
        self.ssh = paramiko.SSHClient()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            self.ssh.connect(self.hostname, self.port, self.username, self.password)
            self.sftp = self.ssh.open_sftp()
        except Exception as e:
            print(f"Connection Error: {e}")

 

SSH 인증키 접속

#SSH 인증키 접속
    def private_key_connect(self):
        self.ssh = paramiko.SSHClient()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:  
            #self.ssh.connect(self.hostname, self.port, self.username, pkey=self.private_key_path)
            self.ssh.connect(self.hostname, self.port, self.username, key_filename=self.private_key_path)
            self.sftp = self.ssh.open_sftp()
        except Exception as e:
            print(f"Connection Error: {e}")

 

self.ssh.connect 구문에 private key를 전달하는 변수를 두 가지 방법으로 사용이 가능하며,

1. key_filename=self.private_key_path
2. pkey=self.private_key_path

 

main에서 private_key 값을 전달하는 방식에 차이가 발생한다.
--key_filename 사용 시
private_key_path = 'C:/user_home_path/.ssh/id_rsa'
--pkey 사용 시
private_key_path = paramiko.RSAKey.from_private_key_file('C:/user_home_path/.ssh/id_rsa')

paramiko 공식문서의 설명에서, key_filename을 사용하면, private key와 일치하는 이름의 ~.pub 파일을 동시에 로드하여, 인증을 시도한다고 한다. 별도의 복호화 과정이 필요 없다.
https://docs.paramiko.org/en/2.4/api/client.html

 

SSH 인증키 접속 & 서버 File 다운로드 함수 호출

def main():
    hostname = '서버 IP'  #192.168.xx.xxx
    port = 22
    username = 'root'
    password = '' #패스워드 인증 접속시 패스워드 기재
    private_key_path = 'C:/user_home_path/.ssh/id_rsa'
   
    remote_file_path = '/remote_path/test.txt'
    local_file_path = 'C:/(Client)local_path/test.txt'

    sftp_client = GetFromSftp(hostname, port, username, password, private_key_path)
    #sftp_client.password_connect() #ID, Password 접속 수행
    sftp_client.private_key_connect()
    sftp_client.download_file(remote_file_path, local_file_path)
    sftp_client.close()

 

SSH 인증키 접속 Client 수행 작업

--Windows 터미널(cmd) 접속
--SSH 키 생성
C:\User\사용자명> ssh-keygen -t rsa

--.ssh 디렉터리로 이동
C:\User\사용자명> cd .ssh

--.ssh 디렉터리 내에 private key, public key SSH 키 한쌍이 생성 되었다.
C:\User\사용자명> dir
2024-03-15  오후 01:14             2,602 id_rsa #Private Key
2024-03-15  오후 01:14               566 id_rsa.pub #Public Key

SSH 인증키 접속 Server 수행 작업

--생성된 id_rsa.pub(public key) 를 접속하고자 하는 원격 서버에 authorized_keys 파일에 복사 
**id_rsa.pub 파일의 내용을 복사한 다음 원격 서버 사용자 홈 디렉터리 하위 .ssh 의 authorized_keys 파일 하단에 붙여 넣거나, 명령문으로 전송한다.(리눅스 명령문 예시 $ ssh-copy-id root@서버IP / Windows public key 전송 명령어는 추후 업데이트)

--접속할 원격 서버 작업
$ vi $HOME/.ssh/authorized_keys
or
$ vi ~/.ssh/authorized_keys
ex) '/root/.ssh/authorized_keys'

 

아래는 전체 작성된 코드입니다.(간단한 예시지만 보통 부분 코드만 검색되고, 전체 코드는 없어서 올려 봅니다.)

import logging
import paramiko #python paramiko 라이브러리 import

class GetFromSftp:
    
    def __init__(self, hostname, port, username, password, private_key_path):
        self.hostname = hostname
        self.port = port
        self.username = username
        self.password = password #Password 접속 초기화
        self.private_key_path = private_key_path #SSH 인증키 접속 초기화
        self.ssh = None
        self.sftp = None
        logging.debug("Init SFTP")    
    
    #패스워드 접속
    def password_connect(self):
        self.ssh = paramiko.SSHClient()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            self.ssh.connect(self.hostname, self.port, self.username, self.password)
            self.sftp = self.ssh.open_sftp()
        except Exception as e:
            print(f"Connection Error: {e}")
    
    #SSH 인증키 접속
    def private_key_connect(self):
        self.ssh = paramiko.SSHClient()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:  
            #self.ssh.connect(self.hostname, self.port, self.username, pkey=self.private_key_path) #pkey connect 예시
            self.ssh.connect(self.hostname, self.port, self.username, key_filename=self.private_key_path)
            self.sftp = self.ssh.open_sftp()
        except Exception as e:
            print(f"Connection Error: {e}")        
    
    #서버 File 다운로드
    def download_file(self, remote_file_path, local_file_path):

        try:
            self.sftp.get(remote_file_path, local_file_path)
            print(f"File downloaded: {local_file_path}")

        except Exception as e:
            print(f"Error downloading file: {e}")    

    def close(self):
        if self.sftp:
            self.sftp.close()
        if self.ssh:
            self.ssh.close()            

def main():
    hostname = '서버 IP'
    port = 22
    username = 'root'
    password = '' #패스워드 접속 구현시 패스워드 기재
    #private_key_path = paramiko.RSAKey.from_private_key_file('C:/user_home_path/.ssh/id_rsa')
    private_key_path = 'C:/user_home_path/.ssh/id_rsa'
    remote_file_path = '/remote_path/test.txt'
    local_file_path = 'C:/local_path/test.txt'

    sftp_client = GetFromSftp(hostname, port, username, password, private_key_path)
    #sftp_client.password_connect() #ID, Password 접속 수행
    sftp_client.private_key_connect()
    sftp_client.download_file(remote_file_path, local_file_path)
    sftp_client.close()

if __name__ == "__main__":
    main()

 

 

반응형