275 lines
10 KiB
Python
275 lines
10 KiB
Python
"""88: SFTP 시험 서버 (paramiko). 단위 시험이 띄워 쓴다 - 앱에는 들어가지 않는다.
|
|
|
|
사용: python sftp_test_server.py --root <폴더> --hostkey <키 파일> [--user u] [--password pw] [--pubkey <공개 키 파일>]
|
|
[--legacy-only] [--port 0]
|
|
- 첫 줄에 "PORT <번호>" 를 찍는다 (port 0 이면 빈 포트).
|
|
- --legacy-only: 호스트 키 서명을 ssh-rsa (SHA-1) 만 (오래된 서버 흉내, RSA 호스트 키일 때).
|
|
- 호스트 키 파일이 없으면 만든다 (--keytype ed25519 · rsa).
|
|
- --fixed-hostkey: 실행마다 같은 ed25519 호스트 키 (캡처의 지문 글이 흔들리지 않게 - 시험 전용, 파일에 남기지 않음).
|
|
paramiko 3.5.1 이 필요하다 (tools\\sftp_test_py 의 venv: python -m pip install "paramiko==3.5.1").
|
|
5.0 은 ssh-rsa (SHA-1) · SHA-1 키 교환을 빼서 --legacy-only (오래된 서버 흉내) 가 아무 방식도 내놓지 못한다.
|
|
"""
|
|
import argparse
|
|
import os
|
|
import socket
|
|
import sys
|
|
import threading
|
|
|
|
import paramiko
|
|
from paramiko import SFTPAttributes, SFTPHandle, SFTPServer, SFTPServerInterface, SFTP_OK, SFTP_NO_SUCH_FILE, SFTP_PERMISSION_DENIED
|
|
|
|
|
|
def _err(e):
|
|
if isinstance(e, FileNotFoundError):
|
|
return SFTP_NO_SUCH_FILE
|
|
if isinstance(e, PermissionError):
|
|
return SFTP_PERMISSION_DENIED
|
|
return SFTPServer.convert_errno(e.errno) if getattr(e, 'errno', None) else paramiko.SFTP_FAILURE
|
|
|
|
|
|
class _Handle(SFTPHandle):
|
|
def stat(self):
|
|
try:
|
|
return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno() if hasattr(self, 'readfile') else self.writefile.fileno()))
|
|
except OSError as e:
|
|
return _err(e)
|
|
|
|
def chattr(self, attr):
|
|
return SFTP_OK
|
|
|
|
|
|
class _Sftp(SFTPServerInterface):
|
|
ROOT = '.'
|
|
|
|
def _real(self, path):
|
|
path = self.canonicalize(path)
|
|
return os.path.join(self.ROOT, path.lstrip('/').replace('/', os.sep))
|
|
|
|
def canonicalize(self, path):
|
|
# 홈 = 맨 위 ("/home/jj" 처럼 보이게 하지 않고 "/" 로)
|
|
p = path.replace('\\', '/')
|
|
if not p.startswith('/'):
|
|
p = '/' + p
|
|
parts = []
|
|
for x in p.split('/'):
|
|
if x in ('', '.'):
|
|
continue
|
|
if x == '..':
|
|
if parts:
|
|
parts.pop()
|
|
continue
|
|
parts.append(x)
|
|
return '/' + '/'.join(parts)
|
|
|
|
def list_folder(self, path):
|
|
real = self._real(path)
|
|
try:
|
|
out = []
|
|
for name in os.listdir(real):
|
|
a = SFTPAttributes.from_stat(os.lstat(os.path.join(real, name)))
|
|
a.filename = name
|
|
out.append(a)
|
|
return out
|
|
except OSError as e:
|
|
return _err(e)
|
|
|
|
def stat(self, path):
|
|
try:
|
|
return SFTPAttributes.from_stat(os.stat(self._real(path)))
|
|
except OSError as e:
|
|
return _err(e)
|
|
|
|
def lstat(self, path):
|
|
try:
|
|
return SFTPAttributes.from_stat(os.lstat(self._real(path)))
|
|
except OSError as e:
|
|
return _err(e)
|
|
|
|
def open(self, path, flags, attr):
|
|
real = self._real(path)
|
|
try:
|
|
binary = getattr(os, 'O_BINARY', 0)
|
|
fd = os.open(real, flags | binary, 0o666)
|
|
except OSError as e:
|
|
return _err(e)
|
|
if flags & os.O_CREAT and attr is not None:
|
|
attr._flags &= ~attr.FLAG_PERMISSIONS
|
|
if flags & os.O_WRONLY:
|
|
mode = 'ab' if flags & os.O_APPEND else 'wb'
|
|
elif flags & os.O_RDWR:
|
|
mode = 'a+b' if flags & os.O_APPEND else 'r+b'
|
|
else:
|
|
mode = 'rb'
|
|
try:
|
|
f = os.fdopen(fd, mode)
|
|
except OSError as e:
|
|
return _err(e)
|
|
h = _Handle(flags)
|
|
h.filename = real
|
|
h.readfile = f
|
|
h.writefile = f
|
|
return h
|
|
|
|
def remove(self, path):
|
|
try:
|
|
os.remove(self._real(path))
|
|
except OSError as e:
|
|
return _err(e)
|
|
return SFTP_OK
|
|
|
|
def rename(self, old, new):
|
|
# SFTP v3: 대상이 있으면 실패 (OpenSSH 와 같게)
|
|
if os.path.exists(self._real(new)):
|
|
return paramiko.SFTP_FAILURE
|
|
try:
|
|
os.rename(self._real(old), self._real(new))
|
|
except OSError as e:
|
|
return _err(e)
|
|
return SFTP_OK
|
|
|
|
def posix_rename(self, old, new):
|
|
try:
|
|
os.replace(self._real(old), self._real(new))
|
|
except OSError as e:
|
|
return _err(e)
|
|
return SFTP_OK
|
|
|
|
def mkdir(self, path, attr):
|
|
try:
|
|
os.mkdir(self._real(path))
|
|
except OSError as e:
|
|
return _err(e)
|
|
return SFTP_OK
|
|
|
|
def rmdir(self, path):
|
|
try:
|
|
os.rmdir(self._real(path))
|
|
except OSError as e:
|
|
return _err(e)
|
|
return SFTP_OK
|
|
|
|
def chattr(self, path, attr):
|
|
try:
|
|
if attr.atime is not None and attr.mtime is not None:
|
|
os.utime(self._real(path), (attr.atime, attr.mtime))
|
|
except OSError as e:
|
|
return _err(e)
|
|
return SFTP_OK
|
|
|
|
|
|
class _Server(paramiko.ServerInterface):
|
|
def __init__(self, user, password, pubkey):
|
|
self.user, self.password, self.pubkey = user, password, pubkey
|
|
|
|
def get_allowed_auths(self, username):
|
|
auths = []
|
|
if self.password is not None:
|
|
auths.append('password')
|
|
if self.pubkey is not None:
|
|
auths.append('publickey')
|
|
return ','.join(auths)
|
|
|
|
def check_auth_password(self, username, password):
|
|
if username == self.user and self.password is not None and password == self.password:
|
|
return paramiko.AUTH_SUCCESSFUL
|
|
return paramiko.AUTH_FAILED
|
|
|
|
def check_auth_publickey(self, username, key):
|
|
if username == self.user and self.pubkey is not None and key.get_base64() == self.pubkey.get_base64():
|
|
return paramiko.AUTH_SUCCESSFUL
|
|
return paramiko.AUTH_FAILED
|
|
|
|
def check_channel_request(self, kind, chanid):
|
|
return paramiko.OPEN_SUCCEEDED if kind == 'session' else paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
|
|
|
|
|
|
def _load_or_make_key(path, keytype):
|
|
if os.path.exists(path):
|
|
try:
|
|
return paramiko.Ed25519Key(filename=path)
|
|
except Exception:
|
|
return paramiko.RSAKey(filename=path)
|
|
if keytype == 'rsa':
|
|
k = paramiko.RSAKey.generate(2048)
|
|
k.write_private_key_file(path)
|
|
return k
|
|
# ed25519 는 paramiko 로 만들 수 없어 cryptography 로
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ed25519
|
|
pk = ed25519.Ed25519PrivateKey.generate()
|
|
pem = pk.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.OpenSSH, serialization.NoEncryption())
|
|
with open(path, 'wb') as f:
|
|
f.write(pem)
|
|
return paramiko.Ed25519Key(filename=path)
|
|
|
|
|
|
def _fixed_key():
|
|
# 정한 글에서 만든 ed25519 키 - 캡처 시험이 실행마다 같은 지문을 보게 (시험 서버 전용)
|
|
import hashlib
|
|
import io
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ed25519
|
|
pk = ed25519.Ed25519PrivateKey.from_private_bytes(hashlib.sha256(b'jj-mkvmaker sftp shot host key').digest())
|
|
pem = pk.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.OpenSSH, serialization.NoEncryption())
|
|
return paramiko.Ed25519Key(file_obj=io.StringIO(pem.decode()))
|
|
|
|
|
|
def _public_key(path):
|
|
text = open(path, encoding='ascii').read().split()
|
|
kind, data = text[0], text[1]
|
|
return paramiko.PKey.from_type_string(kind, __import__('base64').b64decode(data))
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument('--root', required=True)
|
|
ap.add_argument('--hostkey', required=True)
|
|
ap.add_argument('--keytype', default='ed25519')
|
|
ap.add_argument('--user', default='jj')
|
|
ap.add_argument('--password')
|
|
ap.add_argument('--pubkey')
|
|
ap.add_argument('--legacy-only', action='store_true')
|
|
ap.add_argument('--port', type=int, default=0)
|
|
ap.add_argument('--fixed-hostkey', action='store_true')
|
|
a = ap.parse_args()
|
|
_Sftp.ROOT = os.path.abspath(a.root)
|
|
host_key = _fixed_key() if a.fixed_hostkey else _load_or_make_key(a.hostkey, a.keytype)
|
|
pub = _public_key(a.pubkey) if a.pubkey else None
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
sock.bind(('127.0.0.1', a.port))
|
|
sock.listen(16)
|
|
print('PORT %d' % sock.getsockname()[1], flush=True)
|
|
print('FINGERPRINT SHA256:%s %s' % (
|
|
__import__('base64').b64encode(__import__('hashlib').sha256(host_key.asbytes()).digest()).decode().rstrip('='),
|
|
host_key.get_name()), flush=True)
|
|
|
|
def serve(conn):
|
|
t = paramiko.Transport(conn)
|
|
if a.legacy_only:
|
|
# 오래된 서버: 호스트 키 서명은 ssh-rsa (SHA-1) 만 · 키 교환도 SHA-1 만
|
|
t.disabled_algorithms = {'pubkeys': ['rsa-sha2-256', 'rsa-sha2-512'],
|
|
'keys': ['rsa-sha2-256', 'rsa-sha2-512'],
|
|
'kex': ['curve25519-sha256@libssh.org', 'curve25519-sha256', 'ecdh-sha2-nistp256',
|
|
'ecdh-sha2-nistp384', 'ecdh-sha2-nistp521',
|
|
'diffie-hellman-group-exchange-sha256', 'diffie-hellman-group14-sha256',
|
|
'diffie-hellman-group16-sha512']}
|
|
t.add_server_key(host_key)
|
|
t.set_subsystem_handler('sftp', SFTPServer, _Sftp)
|
|
try:
|
|
t.start_server(server=_Server(a.user, a.password, pub))
|
|
while t.is_active():
|
|
t.join(1)
|
|
except Exception as e:
|
|
print('ERR', e, file=sys.stderr, flush=True)
|
|
|
|
# 표준 입력이 닫히면 (시험이 끝나면) 끝난다
|
|
threading.Thread(target=lambda: (sys.stdin.read(), os._exit(0)), daemon=True).start()
|
|
while True:
|
|
conn, _ = sock.accept()
|
|
threading.Thread(target=serve, args=(conn,), daemon=True).start()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|