diff --git a/test/ftp_test.dart b/test/ftp_test.dart index 23dcaad..c8a6348 100644 --- a/test/ftp_test.dart +++ b/test/ftp_test.dart @@ -64,15 +64,18 @@ void main() { Process? proc; late int port; late String fp; + late String caFp; - Future start(String mode) async { - final pr = await Process.start(sftpTestPython, ['tool/ftp_test_server.py', '--root', root.path, '--mode', mode]); + Future start(String mode, {bool chain = false}) async { + final pr = await Process.start(sftpTestPython, ['tool/ftp_test_server.py', '--root', root.path, '--mode', mode, if (chain) '--chain']); proc = pr; final lines = pr.stdout.transform(utf8.decoder).transform(const LineSplitter()).asBroadcastStream(); pr.stderr.drain(); port = int.parse((await lines.firstWhere((l) => l.startsWith('PORT ')).timeout(const Duration(seconds: 20))).split(' ')[1]); final f = (await lines.firstWhere((l) => l.startsWith('FINGERPRINT')).timeout(const Duration(seconds: 5))).split(' '); fp = f.length > 1 ? f[1] : ''; + final ca = (await lines.firstWhere((l) => l.startsWith('CA_FINGERPRINT')).timeout(const Duration(seconds: 5))).split(' '); + caFp = ca.length > 1 ? ca[1] : ''; } setUp(() { @@ -131,6 +134,21 @@ void main() { expect(Directory(p.join(root.path, '영상')).existsSync(), isFalse); } + test('보안 검토 7: 사설 CA 가 서명한 서버 인증서 - 지문은 서버 인증서 (잎) 것, CA 지문을 믿어도 같은 CA 의 다른 인증서를 받지 않는다', () async { + await start('explicit', chain: true); + expect(caFp, isNotEmpty); + expect(caFp, isNot(fp)); + final e = await err(() => FtpRemote(srv()).list('/')); + expect(e, isA().having((x) => x.fingerprint, 'fp', fp).having((x) => x.subject, 'subject', contains('127.0.0.1'))); + // CA 의 지문을 믿고 있어도 막는다 (같은 CA 가 서명한 다른 서버로 바뀌어도 모르게 되지 않게) + final viaCa = await err(() => FtpRemote(srv(cert: caFp)).list('/')); + expect(viaCa, isA().having((x) => x.changed, 'changed', isTrue).having((x) => x.fingerprint, 'fp', fp)); + // 잎 지문을 믿으면 명령 · 자료 연결 모두 + final c = FtpRemote(srv(cert: fp)); + await roundTrip(c); + c.close(); + }, skip: hasPy ? false : '시험용 파이썬 없음'); + test('FTPS (명시): 처음 보는 자체 서명 인증서는 지문을 보이고 막음 → 믿으면 폴더 · 쓰기 · 목록 · 부분 읽기 · 옮기기 · 지우기', () async { await start('explicit'); final e = await err(() => FtpRemote(srv()).list('/')); diff --git a/tool/ftp_test_server.py b/tool/ftp_test_server.py index 975adf8..872da06 100644 --- a/tool/ftp_test_server.py +++ b/tool/ftp_test_server.py @@ -21,17 +21,47 @@ 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_cert.pem') + 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) @@ -89,6 +119,7 @@ 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():