test/suite_decryption.py

   1 #
   2 # -*- coding: utf-8 -*-
   3 # Wireshark tests
   4 # By Gerald Combs <gerald@wireshark.org>
   5 #
   6 # Ported from a set of Bash scripts which were copyright 2005 Ulf Lamping
   7 #
   8 # SPDX-License-Identifier: GPL-2.0-or-later
   9 #
  10 '''Decryption tests'''
  11
  12 import config
  13 import os.path
  14 import subprocesstest
  15 import unittest
  16
  17 class case_decrypt_80211(subprocesstest.SubprocessTestCase):
  18     def test_80211_wpa_psk(self):
  19         '''IEEE 802.11 WPA PSK'''
  20         # https://wiki.wireshark.org/SampleCaptures?action=AttachFile&do=view&target=wpa-Induction.pcap
  21         capture_file = os.path.join(config.capture_dir, 'wpa-Induction.pcap.gz')
  22         self.runProcess((config.cmd_tshark,
  23                 '-o', 'wlan.enable_decryption: TRUE',
  24                 '-Tfields',
  25                 '-e', 'http.request.uri',
  26                 '-r', capture_file,
  27                 '-Y', 'http',
  28             ),
  29             env=config.test_env)
  30         self.assertTrue(self.grepOutput('favicon.ico'))
  31
  32     def test_80211_wpa_eap(self):
  33         '''IEEE 802.11 WPA EAP (EAPOL Rekey)'''
  34         # Included in git sources test/captures/wpa-eap-tls.pcap.gz
  35         capture_file = os.path.join(config.capture_dir, 'wpa-eap-tls.pcap.gz')
  36         self.runProcess((config.cmd_tshark,
  37                 '-o', 'wlan.enable_decryption: TRUE',
  38                 '-r', capture_file,
  39                 '-Y', 'wlan.analysis.tk==7d9987daf5876249b6c773bf454a0da7',
  40                 ),
  41             env=config.test_env)
  42         self.assertTrue(self.grepOutput('Group Message'))
  43
  44     def test_80211_wpa_eapol_incomplete_rekeys(self):
  45         '''WPA decode with message1+2 only and secure bit set on message 2'''
  46         # Included in git sources test/captures/wpa-test-decode.pcap.gz
  47         capture_file = os.path.join(config.capture_dir, 'wpa-test-decode.pcap.gz')
  48         self.runProcess((config.cmd_tshark,
  49                 '-o', 'wlan.enable_decryption: TRUE',
  50                 '-r', capture_file,
  51                 '-Y', 'icmp.resp_to == 4263',
  52                 ),
  53             env=config.test_env)
  54         self.assertTrue(self.grepOutput('Echo'))
  55
  56     def test_80211_wpa_psk_mfp(self):
  57         '''WPA decode management frames with MFP enabled (802.11w)'''
  58         # Included in git sources test/captures/wpa-test-decode-mgmt.pcap.gz
  59         capture_file = os.path.join(config.capture_dir, 'wpa-test-decode-mgmt.pcap.gz')
  60         self.runProcess((config.cmd_tshark,
  61                 '-o', 'wlan.enable_decryption: TRUE',
  62                 '-r', capture_file,
  63                 '-Y', 'wlan.fixed.reason_code == 2 || wlan.fixed.category_code == 3',
  64                 ),
  65             env=config.test_env)
  66         self.assertEqual(self.countOutput('802.11.*SN=.*FN=.*Flags='), 3)
  67
  68
  69     def test_80211_wpa_tdls(self):
  70         '''WPA decode traffic in a TDLS (Tunneled Direct-Link Setup) session (802.11z)'''
  71         if not config.have_libgcrypt16:
  72             self.skipTest('Requires GCrypt 1.6 or later.')
  73         # Included in git sources test/captures/wpa-test-decode-tdls.pcap.gz
  74         capture_file = os.path.join(config.capture_dir, 'wpa-test-decode-tdls.pcap.gz')
  75         self.runProcess((config.cmd_tshark,
  76                 #'-ouat:80211_keys:"wpa-pwd","12345678"',
  77                 '-o', 'wlan.enable_decryption: TRUE',
  78                 '-r', capture_file,
  79                 '-Y', 'icmp',
  80                 ),
  81             env=config.test_env)
  82         self.assertEqual(self.countOutput('ICMP.*Echo .ping'), 2)
  83
  84 class case_decrypt_dtls(subprocesstest.SubprocessTestCase):
  85     def test_dtls(self):
  86         '''DTLS'''
  87         # https://wiki.wireshark.org/SampleCaptures?action=AttachFile&do=view&target=snakeoil.tgz
  88         capture_file = os.path.join(config.capture_dir, 'snakeoil-dtls.pcap')
  89         self.runProcess((config.cmd_tshark,
  90                 '-r', capture_file,
  91                 '-Tfields',
  92                 '-e', 'data.data',
  93                 '-Y', 'data',
  94             ),
  95             env=config.test_env)
  96         self.assertTrue(self.grepOutput('69:74:20:77:6f:72:6b:20:21:0a'))
  97
  98     def test_dtls_psk_aes128ccm8(self):
  99         '''DTLS 1.2 with PSK, AES-128-CCM-8'''
 100         capture_file = os.path.join(config.capture_dir, 'dtls12-aes128ccm8.pcap')
 101         self.runProcess((config.cmd_tshark,
 102                 '-r', capture_file,
 103                 '-o', 'dtls.psk:ca19e028a8a372ad2d325f950fcaceed',
 104                 '-x'
 105             ),
 106             env=config.test_env)
 107         dt_count = self.countOutput('Decrypted DTLS')
 108         wfm_count = self.countOutput('Works for me!.')
 109         self.assertTrue(dt_count == 7 and wfm_count == 2)
 110
 111     def test_dtls_udt(self):
 112         '''UDT over DTLS 1.2 with RSA key'''
 113         capture_file = os.path.join(config.capture_dir, 'udt-dtls.pcapng.gz')
 114         key_file = os.path.join(config.key_dir, 'udt-dtls.key')
 115         self.runProcess((config.cmd_tshark,
 116                 '-r', capture_file,
 117                 '-o', 'dtls.keys_list:0.0.0.0,0,data,{}'.format(key_file),
 118                 '-Y', 'dtls && udt.type==ack',
 119             ),
 120             env=config.test_env)
 121         self.assertTrue(self.grepOutput('UDT'))
 122
 123 class case_decrypt_tls(subprocesstest.SubprocessTestCase):
 124     def test_tls(self):
 125         '''TLS using the server's private key'''
 126         # https://wiki.wireshark.org/SampleCaptures?action=AttachFile&do=view&target=snakeoil2_070531.tgz
 127         capture_file = os.path.join(config.capture_dir, 'rsasnakeoil2.pcap')
 128         self.runProcess((config.cmd_tshark,
 129                 '-r', capture_file,
 130                 '-Tfields',
 131                 '-e', 'http.request.uri',
 132                 '-Y', 'http',
 133             ),
 134             env=config.test_env)
 135         self.assertTrue(self.grepOutput('favicon.ico'))
 136
 137     def test_tls_rsa_pq(self):
 138         '''TLS using the server's private key with p < q
 139         (test whether libgcrypt is correctly called)'''
 140         capture_file = os.path.join(config.capture_dir, 'rsa-p-lt-q.pcap')
 141         key_file = os.path.join(config.key_dir, 'rsa-p-lt-q.key')
 142         self.runProcess((config.cmd_tshark,
 143                 '-r', capture_file,
 144                 '-o', 'tls.keys_list:0.0.0.0,443,http,{}'.format(key_file),
 145                 '-Tfields',
 146                 '-e', 'http.request.uri',
 147                 '-Y', 'http',
 148             ),
 149             env=config.test_env)
 150         self.assertTrue(self.grepOutput('/'))
 151
 152     def test_tls_with_password(self):
 153         '''TLS using the server's private key with password'''
 154         capture_file = os.path.join(config.capture_dir, 'dmgr.pcapng')
 155         self.runProcess((config.cmd_tshark,
 156                 '-r', capture_file,
 157                 '-Tfields',
 158                 '-e', 'http.request.uri',
 159                 '-Y', 'http',
 160             ),
 161             env=config.test_env)
 162         self.assertTrue(self.grepOutput('unsecureLogon.jsp'))
 163
 164     def test_tls_master_secret(self):
 165         '''TLS using the master secret and ssl.keylog_file preference aliasing'''
 166         capture_file = os.path.join(config.capture_dir, 'dhe1.pcapng.gz')
 167         key_file = os.path.join(config.key_dir, 'dhe1_keylog.dat')
 168         self.runProcess((config.cmd_tshark,
 169                 '-r', capture_file,
 170                 '-o', 'ssl.keylog_file: {}'.format(key_file),
 171                 '-o', 'tls.desegment_ssl_application_data: FALSE',
 172                 '-o', 'http.tls.port: 443',
 173                 '-Tfields',
 174                 '-e', 'http.request.method',
 175                 '-e', 'http.request.uri',
 176                 '-e', 'http.request.version',
 177                 '-Y', 'http',
 178             ),
 179             env=config.test_env)
 180         self.assertTrue(self.grepOutput('GET\s+/test\s+HTTP/1.0'))
 181
 182     def test_tls12_renegotiation(self):
 183         '''TLS 1.2 with renegotiation'''
 184         capture_file = os.path.join(config.capture_dir, 'tls-renegotiation.pcap')
 185         key_file = os.path.join(config.key_dir, 'rsasnakeoil2.key')
 186         self.runProcess((config.cmd_tshark,
 187                 '-r', capture_file,
 188                 '-o', 'tls.keys_list:0.0.0.0,4433,http,{}'.format(key_file),
 189                 '-Tfields',
 190                 '-e', 'http.content_length',
 191                 '-Y', 'http',
 192             ),
 193             env=config.test_env)
 194         count_0 = self.countOutput('^0$')
 195         count_2151 = self.countOutput('^2151$')
 196         self.assertTrue(count_0 == 1 and count_2151 == 1)
 197
 198     def test_tls12_psk_aes128ccm(self):
 199         '''TLS 1.2 with PSK, AES-128-CCM'''
 200         capture_file = os.path.join(config.capture_dir, 'tls12-aes128ccm.pcap')
 201         self.runProcess((config.cmd_tshark,
 202                 '-r', capture_file,
 203                 '-o', 'tls.psk:ca19e028a8a372ad2d325f950fcaceed',
 204                 '-q',
 205                 '-z', 'follow,tls,ascii,0',
 206             ),
 207             env=config.test_env)
 208         self.assertTrue(self.grepOutput('http://www.gnu.org/software/gnutls'))
 209
 210     def test_tls12_psk_aes256gcm(self):
 211         '''TLS 1.2 with PSK, AES-256-GCM'''
 212         capture_file = os.path.join(config.capture_dir, 'tls12-aes256gcm.pcap')
 213         self.runProcess((config.cmd_tshark,
 214                 '-r', capture_file,
 215                 '-o', 'tls.psk:ca19e028a8a372ad2d325f950fcaceed',
 216                 '-q',
 217                 '-z', 'follow,tls,ascii,0',
 218             ),
 219             env=config.test_env)
 220         self.assertTrue(self.grepOutput('http://www.gnu.org/software/gnutls'))
 221
 222     def test_tls12_chacha20poly1305(self):
 223         '''TLS 1.2 with ChaCha20-Poly1305'''
 224         if not config.have_libgcrypt17:
 225             self.skipTest('Requires GCrypt 1.7 or later.')
 226         capture_file = os.path.join(config.capture_dir, 'tls12-chacha20poly1305.pcap')
 227         key_file = os.path.join(config.key_dir, 'tls12-chacha20poly1305.keys')
 228         ciphers=[
 229             'ECDHE-ECDSA-CHACHA20-POLY1305',
 230             'ECDHE-RSA-CHACHA20-POLY1305',
 231             'DHE-RSA-CHACHA20-POLY1305',
 232             'RSA-PSK-CHACHA20-POLY1305',
 233             'DHE-PSK-CHACHA20-POLY1305',
 234             'ECDHE-PSK-CHACHA20-POLY1305',
 235             'PSK-CHACHA20-POLY1305',
 236         ]
 237         stream = 0
 238         for cipher in ciphers:
 239             self.runProcess((config.cmd_tshark,
 240                     '-r', capture_file,
 241                     '-o', 'tls.keylog_file: {}'.format(key_file),
 242                     '-q',
 243                     '-z', 'follow,tls,ascii,{}'.format(stream),
 244                 ),
 245                 env=config.test_env)
 246             stream += 1
 247             self.assertTrue(self.grepOutput('Cipher is {}'.format(cipher)))
 248
 249     def test_tls13_chacha20poly1305(self):
 250         '''TLS 1.3 with ChaCha20-Poly1305'''
 251         if not config.have_libgcrypt17:
 252             self.skipTest('Requires GCrypt 1.7 or later.')
 253         capture_file = os.path.join(config.capture_dir, 'tls13-20-chacha20poly1305.pcap')
 254         key_file = os.path.join(config.key_dir, 'tls13-20-chacha20poly1305.keys')
 255         self.runProcess((config.cmd_tshark,
 256                 '-r', capture_file,
 257                 '-o', 'tls.keylog_file: {}'.format(key_file),
 258                 '-q',
 259                 '-z', 'follow,tls,ascii,0',
 260             ),
 261             env=config.test_env)
 262         self.assertTrue(self.grepOutput('TLS13-CHACHA20-POLY1305-SHA256'))
 263
 264     def test_tls13_rfc8446(self):
 265         '''TLS 1.3 (normal session, then early data followed by normal data).'''
 266         if not config.have_libgcrypt16:
 267             self.skipTest('Requires GCrypt 1.6 or later.')
 268         capture_file = os.path.join(config.capture_dir, 'tls13-rfc8446.pcap')
 269         key_file = os.path.join(config.key_dir, 'tls13-rfc8446.keys')
 270         proc = self.runProcess((config.cmd_tshark,
 271                 '-r', capture_file,
 272                 '-otls.keylog_file:{}'.format(key_file),
 273                 '-Y', 'http',
 274                 '-Tfields',
 275                 '-e', 'frame.number',
 276                 '-e', 'http.request.uri',
 277                 '-e', 'http.file_data',
 278                 '-E', 'separator=|',
 279             ),
 280             env=config.test_env)
 281         self.assertEqual([
 282             r'5|/first|',
 283             r'6||Request for /first, version TLSv1.3, Early data: no\n',
 284             r'8|/early|',
 285             r'10||Request for /early, version TLSv1.3, Early data: yes\n',
 286             r'12|/second|',
 287             r'13||Request for /second, version TLSv1.3, Early data: yes\n',
 288         ], proc.stdout_str.splitlines())
 289
 290     def test_tls13_rfc8446_noearly(self):
 291         '''TLS 1.3 (with undecryptable early data).'''
 292         if not config.have_libgcrypt16:
 293             self.skipTest('Requires GCrypt 1.6 or later.')
 294         capture_file = os.path.join(config.capture_dir, 'tls13-rfc8446.pcap')
 295         key_file = os.path.join(config.key_dir, 'tls13-rfc8446-noearly.keys')
 296         proc = self.runProcess((config.cmd_tshark,
 297                 '-r', capture_file,
 298                 '-otls.keylog_file:{}'.format(key_file),
 299                 '-Y', 'http',
 300                 '-Tfields',
 301                 '-e', 'frame.number',
 302                 '-e', 'http.request.uri',
 303                 '-e', 'http.file_data',
 304                 '-E', 'separator=|',
 305             ),
 306             env=config.test_env)
 307         self.assertEqual([
 308             r'5|/first|',
 309             r'6||Request for /first, version TLSv1.3, Early data: no\n',
 310             r'10||Request for /early, version TLSv1.3, Early data: yes\n',
 311             r'12|/second|',
 312             r'13||Request for /second, version TLSv1.3, Early data: yes\n',
 313         ], proc.stdout_str.splitlines())
 314
 315
 316 class case_decrypt_zigbee(subprocesstest.SubprocessTestCase):
 317     def test_zigbee(self):
 318         '''ZigBee'''
 319         # https://bugs.wireshark.org/bugzilla/show_bug.cgi?id=7022
 320         capture_file = os.path.join(config.capture_dir, 'sample_control4_2012-03-24.pcap')
 321         self.runProcess((config.cmd_tshark,
 322                 '-r', capture_file,
 323                 '-Tfields',
 324                 '-e', 'data.data',
 325                 '-Y', 'zbee_aps',
 326             ),
 327             env=config.test_env)
 328         self.assertTrue(self.grepOutput('30:67:63:63:38:65:20:63:34:2e:64:6d:2e:74:76:20'))
 329
 330 class case_decrypt_ansi_c1222(subprocesstest.SubprocessTestCase):
 331     def test_ansi_c1222(self):
 332         '''ANSI C12.22'''
 333         # https://bugs.wireshark.org/bugzilla/show_bug.cgi?id=9196
 334         capture_file = os.path.join(config.capture_dir, 'c1222_std_example8.pcap')
 335         self.runProcess((config.cmd_tshark,
 336                 '-r', capture_file,
 337                 '-o', 'c1222.decrypt: TRUE',
 338                 '-o', 'c1222.baseoid: 2.16.124.113620.1.22.0',
 339                 '-Tfields',
 340                 '-e', 'c1222.data',
 341             ),
 342             env=config.test_env)
 343         self.assertTrue(self.grepOutput('00:10:4d:41:4e:55:46:41:43:54:55:52:45:52:20:53:4e:20:92'))
 344
 345 class case_decrypt_dvb_ci(subprocesstest.SubprocessTestCase):
 346     def test_dvb_ci(self):
 347         '''DVB-CI'''
 348         # simplified version of the sample capture in
 349         # https://bugs.wireshark.org/bugzilla/show_bug.cgi?id=6700
 350         capture_file = os.path.join(config.capture_dir, 'dvb-ci_UV1_0000.pcap')
 351         self.runProcess((config.cmd_tshark,
 352                 '-r', capture_file,
 353                 '-o', 'dvb-ci.sek: 00000000000000000000000000000000',
 354                 '-o', 'dvb-ci.siv: 00000000000000000000000000000000',
 355                 '-Tfields',
 356                 '-e', 'dvb-ci.cc.sac.padding',
 357             ),
 358             env=config.test_env)
 359         self.assertTrue(self.grepOutput('80:00:00:00:00:00:00:00:00:00:00:00'))
 360
 361 class case_decrypt_ipsec(subprocesstest.SubprocessTestCase):
 362     def test_ipsec_esp(self):
 363         '''IPsec ESP'''
 364         # https://bugs.wireshark.org/bugzilla/show_bug.cgi?id=12671
 365         capture_file = os.path.join(config.capture_dir, 'esp-bug-12671.pcapng.gz')
 366         self.runProcess((config.cmd_tshark,
 367                 '-r', capture_file,
 368                 '-o', 'esp.enable_encryption_decode: TRUE',
 369                 '-Tfields',
 370                 '-e', 'data.data',
 371             ),
 372             env=config.test_env)
 373         self.assertTrue(self.grepOutput('08:09:0a:0b:0c:0d:0e:0f:10:11:12:13:14:15:16:17'))
 374
 375 class case_decrypt_ike_isakmp(subprocesstest.SubprocessTestCase):
 376     def test_ikev1_certs(self):
 377         '''IKEv1 (ISAKMP) with certificates'''
 378         # https://bugs.wireshark.org/bugzilla/show_bug.cgi?id=7951
 379         capture_file = os.path.join(config.capture_dir, 'ikev1-certs.pcap')
 380         self.runProcess((config.cmd_tshark,
 381                 '-r', capture_file,
 382                 '-Tfields',
 383                 '-e', 'x509sat.printableString',
 384             ),
 385             env=config.test_env)
 386         self.assertTrue(self.grepOutput('OpenSwan'))
 387
 388     def test_ikev1_simultaneous(self):
 389         '''IKEv1 (ISAKMP) simultaneous exchanges'''
 390         # https://bugs.wireshark.org/bugzilla/show_bug.cgi?id=12610
 391         capture_file = os.path.join(config.capture_dir, 'ikev1-bug-12610.pcapng.gz')
 392         self.runProcess((config.cmd_tshark,
 393                 '-r', capture_file,
 394                 '-Tfields',
 395                 '-e', 'isakmp.hash',
 396             ),
 397             env=config.test_env)
 398         self.assertTrue(self.grepOutput('b5:25:21:f7:74:96:74:02:c9:f6:ce:e9:5f:d1:7e:5b'))
 399
 400     def test_ikev1_unencrypted(self):
 401         '''IKEv1 (ISAKMP) unencrypted phase 1'''
 402         # https://bugs.wireshark.org/bugzilla/show_bug.cgi?id=12620
 403         capture_file = os.path.join(config.capture_dir, 'ikev1-bug-12620.pcapng.gz')
 404         self.runProcess((config.cmd_tshark,
 405                 '-r', capture_file,
 406                 '-Tfields',
 407                 '-e', 'isakmp.hash',
 408             ),
 409             env=config.test_env)
 410         self.assertTrue(self.grepOutput('40:04:3b:64:0f:43:73:25:0d:5a:c3:a1:fb:63:15:3c'))
 411
 412     def test_ikev2_3des_sha160(self):
 413         '''IKEv2 decryption test (3DES-CBC/SHA1_160)'''
 414         capture_file = os.path.join(config.capture_dir, 'ikev2-decrypt-3des-sha1_160.pcap')
 415         self.runProcess((config.cmd_tshark,
 416                 '-r', capture_file,
 417                 '-Tfields',
 418                 '-e', 'isakmp.auth.data',
 419             ),
 420             env=config.test_env)
 421         self.assertTrue(self.grepOutput('02:f7:a0:d5:f1:fd:c8:ea:81:03:98:18:c6:5b:b9:bd:09:af:9b:89:17:31:9b:88:7f:f9:ba:30:46:c3:44:c7'))
 422
 423     def test_ikev2_aes128_ccm12(self):
 424         '''IKEv2 decryption test (AES-128-CCM-12) - with CBC-MAC verification'''
 425         capture_file = os.path.join(config.capture_dir, 'ikev2-decrypt-aes128ccm12.pcap')
 426         self.runProcess((config.cmd_tshark,
 427                 '-r', capture_file,
 428                 '-Tfields',
 429                 '-e', 'isakmp.auth.data',
 430             ),
 431             env=config.test_env)
 432         self.assertTrue(self.grepOutput('c2:10:43:94:29:9e:1f:fe:79:08:ea:72:0a:d5:d1:37:17:a0:d4:54:e4:fa:0a:21:28:ea:68:94:11:f4:79:c4'))
 433
 434     def test_ikev2_aes128_ccm12_2(self):
 435         '''IKEv2 decryption test (AES-128-CCM-12 using CTR mode, without checksum)'''
 436         capture_file = os.path.join(config.capture_dir, 'ikev2-decrypt-aes128ccm12-2.pcap')
 437         self.runProcess((config.cmd_tshark,
 438                 '-r', capture_file,
 439                 '-Tfields',
 440                 '-e', 'isakmp.auth.data',
 441             ),
 442             env=config.test_env)
 443         self.assertTrue(self.grepOutput('aa:a2:81:c8:7b:4a:19:04:6c:57:27:1d:55:74:88:ca:41:3b:57:22:8c:b9:51:f5:fa:96:40:99:2a:02:85:b9'))
 444
 445     def test_ikev2_aes192ctr_sha512(self):
 446         '''IKEv2 decryption test (AES-192-CTR/SHA2-512)'''
 447         capture_file = os.path.join(config.capture_dir, 'ikev2-decrypt-aes192ctr.pcap')
 448         self.runProcess((config.cmd_tshark,
 449                 '-r', capture_file,
 450                 '-Tfields',
 451                 '-e', 'isakmp.auth.data',
 452             ),
 453             env=config.test_env)
 454         self.assertTrue(self.grepOutput('3e:c2:3d:cf:93:48:48:56:38:40:7c:75:45:47:ae:b3:08:52:90:08:2c:49:f5:83:fd:ba:e5:92:63:a2:0b:4a'))
 455
 456     def test_ikev2_aes256cbc_sha256(self):
 457         '''IKEv2 decryption test (AES-256-CBC/SHA2-256)'''
 458         capture_file = os.path.join(config.capture_dir, 'ikev2-decrypt-aes256cbc.pcapng')
 459         self.runProcess((config.cmd_tshark,
 460                 '-r', capture_file,
 461                 '-Tfields',
 462                 '-e', 'isakmp.auth.data',
 463             ),
 464             env=config.test_env)
 465         self.assertTrue(self.grepOutput('e1:a8:d5:50:06:42:01:a7:ec:02:4a:85:75:8d:06:73:c6:1c:5c:51:0a:c1:3b:cd:22:5d:63:27:f5:0d:a3:d3'))
 466
 467     def test_ikev2_aes256ccm16(self):
 468         '''IKEv2 decryption test (AES-256-CCM-16)'''
 469         capture_file = os.path.join(config.capture_dir, 'ikev2-decrypt-aes256ccm16.pcapng')
 470         self.runProcess((config.cmd_tshark,
 471                 '-r', capture_file,
 472                 '-Tfields',
 473                 '-e', 'isakmp.auth.data',
 474             ),
 475             env=config.test_env)
 476         self.assertTrue(self.grepOutput('fa:2e:74:bd:c0:1e:30:fb:0b:3d:dc:97:23:c9:44:90:95:96:9d:a5:1f:69:e5:60:20:9d:2c:2b:79:40:21:0a'))
 477
 478     def test_ikev2_aes256gcm16(self):
 479         '''IKEv2 decryption test (AES-256-GCM-16)'''
 480         capture_file = os.path.join(config.capture_dir, 'ikev2-decrypt-aes256gcm16.pcap')
 481         self.runProcess((config.cmd_tshark,
 482                 '-r', capture_file,
 483                 '-Tfields',
 484                 '-e', 'isakmp.auth.data',
 485             ),
 486             env=config.test_env)
 487         self.assertTrue(self.grepOutput('9a:b7:1f:14:ab:55:3c:ad:87:3a:1a:a7:0b:99:df:15:5d:ee:77:cd:cf:36:94:b3:b7:52:7a:cb:b9:71:2d:ed'))
 488
 489     def test_ikev2_aes256gcm8(self):
 490         '''IKEv2 decryption test (AES-256-GCM-8)'''
 491         capture_file = os.path.join(config.capture_dir, 'ikev2-decrypt-aes256gcm8.pcap')
 492         self.runProcess((config.cmd_tshark,
 493                 '-r', capture_file,
 494                 '-Tfields',
 495                 '-e', 'isakmp.auth.data',
 496             ),
 497             env=config.test_env)
 498         self.assertTrue(self.grepOutput('4a:66:d8:22:d0:af:bc:22:ad:9a:92:a2:cf:42:87:c9:20:ad:8a:c3:b0:69:a4:a7:e7:5f:e0:a5:d4:99:f9:14'))
 499
 500 class case_decrypt_http2(subprocesstest.SubprocessTestCase):
 501     def test_http2(self):
 502         '''HTTP2 (HPACK)'''
 503         if not config.have_nghttp2:
 504             self.skipTest('Requires nghttp2.')
 505         capture_file = os.path.join(config.capture_dir, 'packet-h2-14_headers.pcapng')
 506         self.runProcess((config.cmd_tshark,
 507                 '-r', capture_file,
 508                 '-Tfields',
 509                 '-e', 'http2.header.value',
 510                 '-d', 'tcp.port==3000,http2',
 511             ),
 512             env=config.test_env)
 513         test_passed = self.grepOutput('nghttp2')
 514         if not test_passed:
 515             self.log_fd.write(u'\n\n-- Verbose output --\n\n')
 516             self.runProcess((config.cmd_tshark,
 517                     '-r', capture_file,
 518                     '-V',
 519                     '-d', 'tcp.port==3000,http2',
 520                 ),
 521                 env=config.test_env)
 522         self.assertTrue(test_passed)
 523
 524 class case_decrypt_kerberos(subprocesstest.SubprocessTestCase):
 525     def test_kerberos(self):
 526         '''Kerberos'''
 527         # Files are from krb-816.zip on the SampleCaptures page.
 528         if not config.have_kerberos:
 529             self.skipTest('Requires kerberos.')
 530         capture_file = os.path.join(config.capture_dir, 'krb-816.pcap.gz')
 531         keytab_file = os.path.join(config.key_dir, 'krb-816.keytab')
 532         self.runProcess((config.cmd_tshark,
 533                 '-r', capture_file,
 534                 '-o', 'kerberos.decrypt: TRUE',
 535                 '-o', 'kerberos.file: {}'.format(keytab_file),
 536                 '-Tfields',
 537                 '-e', 'kerberos.keyvalue',
 538             ),
 539             env=config.test_env)
 540         # keyvalue: ccda7d48219f73c3b28311c4ba7242b3
 541         self.assertTrue(self.grepOutput('cc:da:7d:48:21:9f:73:c3:b2:83:11:c4:ba:72:42:b3'))
 542
 543 class case_decrypt_wireguard(subprocesstest.SubprocessTestCase):
 544     # The "foo_alt" keys are similar as "foo" except that some bits are changed.
 545     # The crypto library should be able to handle this and internally the
 546     # dissector uses MSB to recognize whether a private key is set.
 547     key_Spriv_i = 'AKeZaHwBxjiKLFnkY2unvEdOTtg4AL+M9dQXfopFVFk='
 548     key_Spriv_i_alt = 'B6eZaHwBxjiKLFnkY2unvEdOTtg4AL+M9dQXfopFVJk='
 549     key_Spub_i = 'Igge9KzRytKNwrgkzDE/8hrLu6Ly0OqVdvOPWhA5KR4='
 550     key_Spriv_r = 'cFIxTUyBs1Qil414hBwEgvasEax8CKJ5IS5ZougplWs='
 551     key_Spub_r = 'YDCttCs9e1J52/g9vEnwJJa+2x6RqaayAYMpSVQfGEY='
 552     key_Epriv_i0 = 'sLGLJSOQfyz7JNJ5ZDzFf3Uz1rkiCMMjbWerNYcPFFU='
 553     key_Epriv_i0_alt = 't7GLJSOQfyz7JNJ5ZDzFf3Uz1rkiCMMjbWerNYcPFJU='
 554     key_Epriv_r0 = 'QC4/FZKhFf0b/eXEcCecmZNt6V6PXmRa4EWG1PIYTU4='
 555     key_Epriv_i1 = 'ULv83D+y3vA0t2mgmTmWz++lpVsrP7i4wNaUEK2oX0E='
 556     key_Epriv_r1 = 'sBv1dhsm63cbvWMv/XML+bvynBp9PTdY9Vvptu3HQlg='
 557     # Ephemeral keys and PSK for wireguard-psk.pcap
 558     key_Epriv_i2 = 'iCv2VTi/BC/q0egU931KXrrQ4TSwXaezMgrhh7uCbXs='
 559     key_Epriv_r2 = '8G1N3LnEqYC7+NW/b6mqceVUIGBMAZSm+IpwG1U0j0w='
 560     key_psk2 = '//////////////////////////////////////////8='
 561     key_Epriv_i3 = '+MHo9sfkjPsjCx7lbVhRLDvMxYvTirOQFDSdzAW6kUQ='
 562     key_Epriv_r3 = '0G6t5j1B/We65MXVEBIGuRGYadwB2ITdvJovtAuATmc='
 563     key_psk3 = 'iIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIg='
 564     # dummy key that should not work with anything.
 565     key_dummy = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx='
 566
 567     def runOne(self, args, keylog=None, pcap_file='wireguard-ping-tcp.pcap'):
 568         if not config.have_libgcrypt17:
 569             self.skipTest('Requires Gcrypt 1.7 or later')
 570         capture_file = os.path.join(config.capture_dir, pcap_file)
 571         if keylog:
 572             keylog_file = self.filename_from_id('wireguard.keys')
 573             args += ['-owg.keylog_file:%s' % keylog_file]
 574             with open(keylog_file, 'w') as f:
 575                 f.write("\n".join(keylog))
 576         proc = self.runProcess([config.cmd_tshark, '-r', capture_file] + args,
 577                                env=config.test_env)
 578         lines = proc.stdout_str.splitlines()
 579         return lines
 580
 581     def test_mac1_public(self):
 582         """Check that MAC1 identification using public keys work."""
 583         lines = self.runOne([
 584             '-ouat:wg_keys:"Public","%s"' % self.key_Spub_i,
 585             '-ouat:wg_keys:"Public","%s"' % self.key_Spub_r,
 586             '-Y', 'wg.receiver_pubkey',
 587             '-Tfields',
 588             '-e', 'frame.number',
 589             '-e', 'wg.receiver_pubkey',
 590             '-e', 'wg.receiver_pubkey.known_privkey',
 591         ])
 592         self.assertEqual(4, len(lines))
 593         self.assertIn('1\t%s\t0' % self.key_Spub_r, lines)
 594         self.assertIn('2\t%s\t0' % self.key_Spub_i, lines)
 595         self.assertIn('13\t%s\t0' % self.key_Spub_r, lines)
 596         self.assertIn('14\t%s\t0' % self.key_Spub_i, lines)
 597
 598     def test_mac1_private(self):
 599         """Check that MAC1 identification using private keys work."""
 600         lines = self.runOne([
 601             '-ouat:wg_keys:"Private","%s"' % self.key_Spriv_i,
 602             '-ouat:wg_keys:"Private","%s"' % self.key_Spriv_r,
 603             '-Y', 'wg.receiver_pubkey',
 604             '-Tfields',
 605             '-e', 'frame.number',
 606             '-e', 'wg.receiver_pubkey',
 607             '-e', 'wg.receiver_pubkey.known_privkey',
 608         ])
 609         self.assertEqual(4, len(lines))
 610         self.assertIn('1\t%s\t1' % self.key_Spub_r, lines)
 611         self.assertIn('2\t%s\t1' % self.key_Spub_i, lines)
 612         self.assertIn('13\t%s\t1' % self.key_Spub_r, lines)
 613         self.assertIn('14\t%s\t1' % self.key_Spub_i, lines)
 614
 615     def test_decrypt_initiation_sprivr(self):
 616         """Check for partial decryption using Spriv_r."""
 617         lines = self.runOne([
 618             '-ouat:wg_keys:"Private","%s"' % self.key_Spriv_r,
 619             '-Y', 'wg.type==1',
 620             '-Tfields',
 621             '-e', 'frame.number',
 622             '-e', 'wg.static',
 623             '-e', 'wg.static.known_pubkey',
 624             '-e', 'wg.static.known_privkey',
 625             '-e', 'wg.timestamp.nanoseconds',
 626         ])
 627         # static pubkey is unknown because Spub_i is not added to wg_keys.
 628         self.assertIn('1\t%s\t0\t0\t%s' % (self.key_Spub_i, '356537872'), lines)
 629         self.assertIn('13\t%s\t0\t0\t%s' % (self.key_Spub_i, '490514356'), lines)
 630
 631     def test_decrypt_initiation_ephemeral_only(self):
 632         """Check for partial decryption using Epriv_i."""
 633         lines = self.runOne([
 634             '-ouat:wg_keys:"Public","%s"' % self.key_Spub_r,
 635             '-Y', 'wg.type==1',
 636             '-Tfields',
 637             '-e', 'frame.number',
 638             '-e', 'wg.ephemeral.known_privkey',
 639             '-e', 'wg.static',
 640             '-e', 'wg.timestamp.nanoseconds',
 641         ], keylog=[
 642             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_i0,
 643             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_i1,
 644         ])
 645         # The current implementation tries to write as much decrypted data as
 646         # possible, even if the full handshake cannot be derived.
 647         self.assertIn('1\t1\t%s\t%s' % (self.key_Spub_i, ''), lines)
 648         self.assertIn('13\t1\t%s\t%s' % (self.key_Spub_i, ''), lines)
 649
 650     def test_decrypt_full_initiator(self):
 651         """
 652         Check for full handshake decryption using Spriv_r + Epriv_i.
 653         The public key Spub_r is provided via the key log as well.
 654         """
 655         lines = self.runOne([
 656             '-Tfields',
 657             '-e', 'frame.number',
 658             '-e', 'wg.ephemeral.known_privkey',
 659             '-e', 'wg.static',
 660             '-e', 'wg.timestamp.nanoseconds',
 661             '-e', 'wg.handshake_ok',
 662             '-e', 'icmp.type',
 663             '-e', 'tcp.dstport',
 664         ], keylog=[
 665             '  REMOTE_STATIC_PUBLIC_KEY = %s' % self.key_Spub_r,
 666             '  LOCAL_STATIC_PRIVATE_KEY = %s' % self.key_Spriv_i_alt,
 667             '  LOCAL_EPHEMERAL_PRIVATE_KEY = %s' % self.key_Epriv_i0_alt,
 668             '  LOCAL_EPHEMERAL_PRIVATE_KEY = %s' % self.key_Epriv_i1,
 669         ])
 670         self.assertIn('1\t1\t%s\t%s\t\t\t' % (self.key_Spub_i, '356537872'), lines)
 671         self.assertIn('2\t0\t\t\t1\t\t', lines)
 672         self.assertIn('3\t\t\t\t\t8\t', lines)
 673         self.assertIn('4\t\t\t\t\t0\t', lines)
 674         self.assertIn('13\t1\t%s\t%s\t\t\t' % (self.key_Spub_i, '490514356'), lines)
 675         self.assertIn('14\t0\t\t\t1\t\t', lines)
 676         self.assertIn('17\t\t\t\t\t\t443', lines)
 677         self.assertIn('18\t\t\t\t\t\t49472', lines)
 678
 679     def test_decrypt_full_responder(self):
 680         """Check for full handshake decryption using responder secrets."""
 681         lines = self.runOne([
 682             '-Tfields',
 683             '-e', 'frame.number',
 684             '-e', 'wg.ephemeral.known_privkey',
 685             '-e', 'wg.static',
 686             '-e', 'wg.timestamp.nanoseconds',
 687             '-e', 'wg.handshake_ok',
 688             '-e', 'icmp.type',
 689             '-e', 'tcp.dstport',
 690         ], keylog=[
 691             'REMOTE_STATIC_PUBLIC_KEY=%s' % self.key_Spub_i,
 692             'LOCAL_STATIC_PRIVATE_KEY=%s' % self.key_Spriv_r,
 693             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_r0,
 694             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_r1,
 695         ])
 696         self.assertIn('1\t0\t%s\t%s\t\t\t' % (self.key_Spub_i, '356537872'), lines)
 697         self.assertIn('2\t1\t\t\t1\t\t', lines)
 698         self.assertIn('3\t\t\t\t\t8\t', lines)
 699         self.assertIn('4\t\t\t\t\t0\t', lines)
 700         self.assertIn('13\t0\t%s\t%s\t\t\t' % (self.key_Spub_i, '490514356'), lines)
 701         self.assertIn('14\t1\t\t\t1\t\t', lines)
 702         self.assertIn('17\t\t\t\t\t\t443', lines)
 703         self.assertIn('18\t\t\t\t\t\t49472', lines)
 704
 705     def test_decrypt_psk_initiator(self):
 706         """Check whether PSKs enable decryption for initiation keys."""
 707         lines = self.runOne([
 708             '-Tfields',
 709             '-e', 'frame.number',
 710             '-e', 'wg.handshake_ok',
 711         ], keylog=[
 712             'REMOTE_STATIC_PUBLIC_KEY = %s' % self.key_Spub_r,
 713             'LOCAL_STATIC_PRIVATE_KEY = %s' % self.key_Spriv_i,
 714             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_i2,
 715             'PRESHARED_KEY=%s' % self.key_psk2,
 716             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_r3,
 717             'PRESHARED_KEY=%s' % self.key_psk3,
 718         ], pcap_file='wireguard-psk.pcap')
 719         self.assertIn('2\t1', lines)
 720         self.assertIn('4\t1', lines)
 721
 722     def test_decrypt_psk_responder(self):
 723         """Check whether PSKs enable decryption for responder keys."""
 724         lines = self.runOne([
 725             '-Tfields',
 726             '-e', 'frame.number',
 727             '-e', 'wg.handshake_ok',
 728         ], keylog=[
 729             'REMOTE_STATIC_PUBLIC_KEY=%s' % self.key_Spub_i,
 730             'LOCAL_STATIC_PRIVATE_KEY=%s' % self.key_Spriv_r,
 731             # Epriv_r2 needs psk2. This tests handling of duplicate ephemeral
 732             # keys with multiple PSKs. It should not have adverse effects.
 733             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_r2,
 734             'PRESHARED_KEY=%s' % self.key_dummy,
 735             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_r2,
 736             'PRESHARED_KEY=%s' % self.key_psk2,
 737             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_i3,
 738             'PRESHARED_KEY=%s' % self.key_psk3,
 739             # Epriv_i3 needs psk3, this tests that additional keys again have no
 740             # bad side-effects.
 741             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_i3,
 742             'PRESHARED_KEY=%s' % self.key_dummy,
 743         ], pcap_file='wireguard-psk.pcap')
 744         self.assertIn('2\t1', lines)
 745         self.assertIn('4\t1', lines)
 746
 747     def test_decrypt_psk_wrong_orderl(self):
 748         """Check that the wrong order of lines indeed fail decryption."""
 749         lines = self.runOne([
 750             '-Tfields',
 751             '-e', 'frame.number',
 752             '-e', 'wg.handshake_ok',
 753         ], keylog=[
 754             'REMOTE_STATIC_PUBLIC_KEY=%s' % self.key_Spub_i,
 755             'LOCAL_STATIC_PRIVATE_KEY=%s' % self.key_Spriv_r,
 756             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_r2,
 757             'LOCAL_EPHEMERAL_PRIVATE_KEY=%s' % self.key_Epriv_i3,
 758             'PRESHARED_KEY=%s' % self.key_psk2, # note: swapped with previous line
 759             'PRESHARED_KEY=%s' % self.key_psk3,
 760         ], pcap_file='wireguard-psk.pcap')
 761         self.assertIn('2\t0', lines)
 762         self.assertIn('4\t0', lines)