WIP 보안 검토 7: FTPS 사설 CA 체인 시험 - 지문은 서버 인증서 (잎) 것 (검토 대기)

관리자 지시 (10/18): 사설 CA → 서버 인증서 체인을 시험 서버에 만들어 (tool/ftp_test_server.py --chain, cryptography 로 시험 전용 CA · 잎, 서버는 use_certificate_chain_file 로 체인 전체를 보냄)
onBadCertificate 가 넘기는 인증서가 잎인지 확인했다 → 잎이다 (고칠 것 없음). 이것을 시험으로 남긴다:
- 처음 연결: FtpCertException 의 지문 = 잎 지문 (CA 지문 아님), subject 에 127.0.0.1
- CA 지문을 믿고 있으면 "바뀜" 으로 막는다 (같은 CA 가 서명한 다른 인증서를 받지 않게)
- 잎 지문을 믿으면 명령 · 자료 연결 (PROT P) 모두 됨
시험 서버는 잎 지문 (FINGERPRINT) 과 함께 CA_FINGERPRINT 를 알린다.

시험: ftp_test (새 1) 통과.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

test/ftp_test.dart

@@ -64,15 +64,18 @@ void main() {
     Process? proc;
     late int port;
     late String fp;
+    late String caFp;
 
-    Future<void> start(String mode) async {
-      final pr = await Process.start(sftpTestPython, ['tool/ftp_test_server.py', '--root', root.path, '--mode', mode]);
+    Future<void> 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<void>();
       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<FtpCertException>().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<FtpCertException>().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('/'));

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():