133 lines
5.9 KiB
Python
133 lines
5.9 KiB
Python
# FTP · FTPS 시험 서버 (pyftpdlib 2.2.0 · pyOpenSSL 26.4.0 - 시험용 파이썬 venv ..\tools\sftp_test_py 에 둠, 앱에는 들어가지 않음).
|
|
# 사용: python tool/ftp_test_server.py --root 폴더 [--user jj --password pw] [--mode plain|explicit|implicit] [--cert 파일]
|
|
# 127.0.0.1 의 빈 포트에 열고 "PORT n" · "FINGERPRINT SHA256:AA:BB:…" 를 알린다. 표준 입력이 닫히면 끝난다.
|
|
# --cert 파일이 없으면 자체 서명 인증서를 만들어 그 파일에 둔다 (같은 파일이면 같은 지문).
|
|
import argparse
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
import threading
|
|
import logging
|
|
if os.environ.get('JJ_FTP_DEBUG'):
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
from pyftpdlib.authorizers import DummyAuthorizer
|
|
from pyftpdlib.handlers import FTPHandler
|
|
from pyftpdlib.servers import FTPServer
|
|
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument('--root', required=True)
|
|
ap.add_argument('--user', default='jj')
|
|
ap.add_argument('--password', default='pw')
|
|
ap.add_argument('--mode', default='explicit', choices=['plain', 'explicit', 'implicit'])
|
|
ap.add_argument('--cert', default='')
|
|
# 보안 검토 7: 사설 CA → 서버 인증서 (잎) 체인으로 (시험 전용). 지문은 잎 것 (FINGERPRINT) 과 CA 것 (CA_FINGERPRINT) 을 알린다
|
|
ap.add_argument('--chain', action='store_true')
|
|
a = ap.parse_args()
|
|
|
|
auth = DummyAuthorizer()
|
|
auth.add_user(a.user, a.password, a.root, perm='elradfmwMT')
|
|
fingerprint = ''
|
|
ca_fingerprint = ''
|
|
if a.mode == 'plain':
|
|
handler = FTPHandler
|
|
else:
|
|
from OpenSSL import crypto
|
|
from pyftpdlib.handlers import TLS_FTPHandler
|
|
cert = a.cert or os.path.join(a.root, '..', 'ftp_chain.pem' if a.chain else 'ftp_cert.pem')
|
|
def fp_of(x509):
|
|
der = crypto.dump_certificate(crypto.FILETYPE_ASN1, x509)
|
|
return 'SHA256:' + ':'.join('%02X' % b for b in hashlib.sha256(der).digest())
|
|
if a.chain and not os.path.exists(cert):
|
|
import datetime
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.x509.oid import NameOID
|
|
now = datetime.datetime.now(datetime.timezone.utc)
|
|
def make(cn, issuer, issuer_key, ca, serial):
|
|
k = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
|
|
c = (x509.CertificateBuilder().subject_name(name).issuer_name(issuer.subject if issuer else name)
|
|
.public_key(k.public_key()).serial_number(serial)
|
|
.not_valid_before(now - datetime.timedelta(minutes=5)).not_valid_after(now + datetime.timedelta(days=1))
|
|
.add_extension(x509.BasicConstraints(ca=ca, path_length=None), critical=True)
|
|
.sign(issuer_key or k, hashes.SHA256()))
|
|
return k, c
|
|
ca_k, ca_c = make('JJ Test Private CA', None, None, True, 1)
|
|
k, c = make('127.0.0.1', ca_c, ca_k, False, 2)
|
|
with open(cert, 'wb') as f:
|
|
f.write(k.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption())
|
|
+ c.public_bytes(serialization.Encoding.PEM) + ca_c.public_bytes(serialization.Encoding.PEM))
|
|
if a.chain:
|
|
pems = open(cert, 'rb').read().split(b'-----END CERTIFICATE-----')
|
|
ca_fingerprint = fp_of(crypto.load_certificate(crypto.FILETYPE_PEM, pems[1] + b'-----END CERTIFICATE-----'))
|
|
if not os.path.exists(cert):
|
|
k = crypto.PKey()
|
|
k.generate_key(crypto.TYPE_RSA, 2048)
|
|
c = crypto.X509()
|
|
c.get_subject().CN = '127.0.0.1'
|
|
c.set_serial_number(1)
|
|
c.gmtime_adj_notBefore(0)
|
|
c.gmtime_adj_notAfter(3600 * 24)
|
|
c.set_issuer(c.get_subject())
|
|
c.set_pubkey(k)
|
|
c.sign(k, 'sha256')
|
|
with open(cert, 'wb') as f:
|
|
f.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, k) + crypto.dump_certificate(crypto.FILETYPE_PEM, c))
|
|
der = crypto.dump_certificate(crypto.FILETYPE_ASN1, crypto.load_certificate(crypto.FILETYPE_PEM, open(cert, 'rb').read()))
|
|
fingerprint = 'SHA256:' + ':'.join('%02X' % b for b in hashlib.sha256(der).digest())
|
|
|
|
from pyftpdlib.handlers import TLS_DTPHandler
|
|
from pyftpdlib.handlers.ftps.ssl import SSLConnectionMixin
|
|
|
|
# pyftpdlib 2.2.0 + pyOpenSSL 26: 자료 연결의 TLS 닫기에서 빈 SSL.Error 를 "다시 해 보기" 로 보고 끝없이 되풀이해
|
|
# 올리기 (STOR) 가 끝나지 않았다 (226 이 오지 않음, 10/18). 시험 서버만: TLS 를 한 번 닫아 보고 바로 연결을 닫는다
|
|
class D(TLS_DTPHandler):
|
|
def _do_ssl_shutdown(self):
|
|
try:
|
|
self.socket.shutdown()
|
|
except Exception:
|
|
pass
|
|
super(SSLConnectionMixin, self).close()
|
|
|
|
class H(TLS_FTPHandler):
|
|
certfile = cert
|
|
dtp_handler = D
|
|
tls_control_required = True
|
|
tls_data_required = True
|
|
|
|
if a.mode == 'implicit':
|
|
# pyftpdlib 은 암시 FTPS 를 따로 두지 않는다 - 연결하자마자 TLS 로 (secure_connection)
|
|
class HI(H):
|
|
def handle(self):
|
|
self.secure_connection(self.ssl_context)
|
|
|
|
def handle_ssl_established(self):
|
|
FTPHandler.handle(self)
|
|
|
|
def ftp_AUTH(self, line):
|
|
self.respond('503 Already using TLS.')
|
|
|
|
handler = HI
|
|
else:
|
|
handler = H
|
|
|
|
handler.authorizer = auth
|
|
handler.passive_ports = None
|
|
handler.encoding = 'utf-8'
|
|
server = FTPServer(('127.0.0.1', 0), handler)
|
|
print('PORT', server.socket.getsockname()[1], flush=True)
|
|
print('FINGERPRINT', fingerprint, flush=True)
|
|
print('CA_FINGERPRINT', ca_fingerprint, flush=True)
|
|
|
|
|
|
def watch():
|
|
sys.stdin.read()
|
|
server.close_all()
|
|
os._exit(0)
|
|
|
|
|
|
threading.Thread(target=watch, daemon=True).start()
|
|
server.serve_forever(handle_exit=False)
|