| ## This file is part of Scapy |
| ## Copyright (C) 2007, 2008, 2009 Arnaud Ebalard |
| ## 2015, 2016, 2017 Maxence Tury |
| ## This program is published under a GPLv2 license |
| |
| """ |
| TLS client automaton. This makes for a primitive TLS stack. |
| Obviously you need rights for network access. |
| |
| We support versions SSLv2 to TLS 1.2, along with many features. |
| There is no session resumption mechanism for now. |
| |
| In order to run a client to tcp/50000 with one cipher suite of your choice: |
| > from scapy.all import * |
| > ch = TLSClientHello(ciphers=<int code of the cipher suite>) |
| > t = TLSClientAutomaton(dport=50000, client_hello=ch) |
| > t.run() |
| """ |
| |
| from __future__ import print_function |
| import socket |
| |
| from scapy.pton_ntop import inet_pton |
| from scapy.utils import randstring |
| from scapy.automaton import ATMT |
| from scapy.layers.tls.automaton import _TLSAutomaton |
| from scapy.layers.tls.basefields import _tls_version, _tls_version_options |
| from scapy.layers.tls.session import tlsSession |
| from scapy.layers.tls.extensions import (TLS_Ext_SupportedGroups, |
| TLS_Ext_SupportedVersions, |
| TLS_Ext_SignatureAlgorithms, |
| TLS_Ext_ServerName, ServerName) |
| from scapy.layers.tls.handshake import * |
| from scapy.layers.tls.handshake_sslv2 import * |
| from scapy.layers.tls.keyexchange_tls13 import (TLS_Ext_KeyShare_CH, |
| KeyShareEntry) |
| from scapy.layers.tls.record import (TLS, TLSAlert, TLSChangeCipherSpec, |
| TLSApplicationData) |
| from scapy.modules import six |
| |
| |
| class TLSClientAutomaton(_TLSAutomaton): |
| """ |
| A simple TLS test client automaton. Try to overload some states or |
| conditions and see what happens on the other side. |
| |
| Rather than with an interruption, the best way to stop this client is by |
| typing 'quit'. This won't be a message sent to the server. |
| |
| _'mycert' and 'mykey' may be provided as filenames. They will be used in |
| the handshake, should the server ask for client authentication. |
| _'server_name' does not need to be set. |
| _'client_hello' may hold a TLSClientHello or SSLv2ClientHello to be sent |
| to the server. This is particularly useful for extensions tweaking. |
| _'version' is a quicker way to advertise a protocol version ("sslv2", |
| "tls1", "tls12", etc.) It may be overriden by the previous 'client_hello'. |
| _'data' is a list of raw data to be sent to the server once the handshake |
| has been completed. Both 'stop_server' and 'quit' will work this way. |
| """ |
| |
| def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, |
| mycert=None, mykey=None, |
| client_hello=None, version=None, |
| data=None, |
| **kargs): |
| |
| super(TLSClientAutomaton, self).parse_args(mycert=mycert, |
| mykey=mykey, |
| **kargs) |
| tmp = socket.getaddrinfo(server, dport) |
| self.remote_name = None |
| try: |
| if ':' in server: |
| inet_pton(socket.AF_INET6, server) |
| else: |
| inet_pton(socket.AF_INET, server) |
| except: |
| self.remote_name = socket.getfqdn(server) |
| if self.remote_name != server: |
| tmp = socket.getaddrinfo(self.remote_name, dport) |
| |
| if server_name: |
| self.remote_name = server_name |
| self.remote_family = tmp[0][0] |
| self.remote_ip = tmp[0][4][0] |
| self.remote_port = dport |
| self.local_ip = None |
| self.local_port = None |
| self.socket = None |
| |
| self.client_hello = client_hello |
| self.advertised_tls_version = None |
| if version: |
| v = _tls_version_options.get(version, None) |
| if not v: |
| self.vprint("Unrecognized TLS version option.") |
| else: |
| self.advertised_tls_version = v |
| |
| self.linebreak = False |
| if isinstance(data, bytes): |
| self.data_to_send = [data] |
| elif isinstance(data, six.string_types): |
| self.data_to_send = [raw(data)] |
| elif isinstance(data, list): |
| self.data_to_send = list(raw(d) for d in reversed(data)) |
| else: |
| self.data_to_send = [] |
| |
| |
| def vprint_sessioninfo(self): |
| if self.verbose: |
| s = self.cur_session |
| v = _tls_version[s.tls_version] |
| self.vprint("Version : %s" % v) |
| cs = s.wcs.ciphersuite.name |
| self.vprint("Cipher suite : %s" % cs) |
| if s.tls_version >= 0x0304: |
| ms = s.tls13_master_secret |
| else: |
| ms = s.master_secret |
| self.vprint("Master secret : %s" % repr_hex(ms)) |
| if s.server_certs: |
| self.vprint("Server certificate chain: %r" % s.server_certs) |
| self.vprint() |
| |
| |
| @ATMT.state(initial=True) |
| def INITIAL(self): |
| self.vprint("Starting TLS client automaton.") |
| raise self.INIT_TLS_SESSION() |
| |
| @ATMT.state() |
| def INIT_TLS_SESSION(self): |
| self.cur_session = tlsSession(connection_end="client") |
| self.cur_session.client_certs = self.mycert |
| self.cur_session.client_key = self.mykey |
| v = self.advertised_tls_version |
| if v: |
| self.cur_session.advertised_tls_version = v |
| else: |
| default_version = self.cur_session.advertised_tls_version |
| self.advertised_tls_version = default_version |
| raise self.CONNECT() |
| |
| @ATMT.state() |
| def CONNECT(self): |
| s = socket.socket(self.remote_family, socket.SOCK_STREAM) |
| self.vprint() |
| self.vprint("Trying to connect on %s:%d" % (self.remote_ip, |
| self.remote_port)) |
| s.connect((self.remote_ip, self.remote_port)) |
| self.socket = s |
| self.local_ip, self.local_port = self.socket.getsockname()[:2] |
| self.vprint() |
| if self.cur_session.advertised_tls_version in [0x0200, 0x0002]: |
| raise self.SSLv2_PREPARE_CLIENTHELLO() |
| elif self.cur_session.advertised_tls_version >= 0x0304: |
| raise self.TLS13_START() |
| else: |
| raise self.PREPARE_CLIENTFLIGHT1() |
| |
| ########################### TLS handshake ################################# |
| |
| @ATMT.state() |
| def PREPARE_CLIENTFLIGHT1(self): |
| self.add_record() |
| |
| @ATMT.condition(PREPARE_CLIENTFLIGHT1) |
| def should_add_ClientHello(self): |
| self.add_msg(self.client_hello or TLSClientHello()) |
| raise self.ADDED_CLIENTHELLO() |
| |
| @ATMT.state() |
| def ADDED_CLIENTHELLO(self): |
| pass |
| |
| @ATMT.condition(ADDED_CLIENTHELLO) |
| def should_send_ClientFlight1(self): |
| self.flush_records() |
| raise self.SENT_CLIENTFLIGHT1() |
| |
| @ATMT.state() |
| def SENT_CLIENTFLIGHT1(self): |
| raise self.WAITING_SERVERFLIGHT1() |
| |
| @ATMT.state() |
| def WAITING_SERVERFLIGHT1(self): |
| self.get_next_msg() |
| raise self.RECEIVED_SERVERFLIGHT1() |
| |
| @ATMT.state() |
| def RECEIVED_SERVERFLIGHT1(self): |
| pass |
| |
| @ATMT.condition(RECEIVED_SERVERFLIGHT1, prio=1) |
| def should_handle_ServerHello(self): |
| """ |
| XXX We should check the ServerHello attributes for discrepancies with |
| our own ClientHello. |
| """ |
| self.raise_on_packet(TLSServerHello, |
| self.HANDLED_SERVERHELLO) |
| |
| @ATMT.state() |
| def HANDLED_SERVERHELLO(self): |
| pass |
| |
| @ATMT.condition(RECEIVED_SERVERFLIGHT1, prio=2) |
| def missing_ServerHello(self): |
| raise self.MISSING_SERVERHELLO() |
| |
| @ATMT.state() |
| def MISSING_SERVERHELLO(self): |
| self.vprint("Missing TLS ServerHello message!") |
| raise self.CLOSE_NOTIFY() |
| |
| @ATMT.condition(HANDLED_SERVERHELLO, prio=1) |
| def should_handle_ServerCertificate(self): |
| if not self.cur_session.prcs.key_exchange.anonymous: |
| self.raise_on_packet(TLSCertificate, |
| self.HANDLED_SERVERCERTIFICATE) |
| raise self.HANDLED_SERVERCERTIFICATE() |
| |
| @ATMT.state() |
| def HANDLED_SERVERCERTIFICATE(self): |
| pass |
| |
| @ATMT.condition(HANDLED_SERVERHELLO, prio=2) |
| def missing_ServerCertificate(self): |
| raise self.MISSING_SERVERCERTIFICATE() |
| |
| @ATMT.state() |
| def MISSING_SERVERCERTIFICATE(self): |
| self.vprint("Missing TLS Certificate message!") |
| raise self.CLOSE_NOTIFY() |
| |
| @ATMT.state() |
| def HANDLED_CERTIFICATEREQUEST(self): |
| self.vprint("Server asked for a certificate...") |
| if not self.mykey or not self.mycert: |
| self.vprint("No client certificate to send!") |
| self.vprint("Will try and send an empty Certificate message...") |
| |
| @ATMT.condition(HANDLED_SERVERCERTIFICATE, prio=1) |
| def should_handle_ServerKeyExchange_from_ServerCertificate(self): |
| """ |
| XXX We should check the ServerKeyExchange attributes for discrepancies |
| with our own ClientHello, along with the ServerHello and Certificate. |
| """ |
| self.raise_on_packet(TLSServerKeyExchange, |
| self.HANDLED_SERVERKEYEXCHANGE) |
| |
| @ATMT.state(final=True) |
| def MISSING_SERVERKEYEXCHANGE(self): |
| pass |
| |
| @ATMT.condition(HANDLED_SERVERCERTIFICATE, prio=2) |
| def missing_ServerKeyExchange(self): |
| if not self.cur_session.prcs.key_exchange.no_ske: |
| raise self.MISSING_SERVERKEYEXCHANGE() |
| |
| @ATMT.state() |
| def HANDLED_SERVERKEYEXCHANGE(self): |
| pass |
| |
| def should_handle_CertificateRequest(self): |
| """ |
| XXX We should check the CertificateRequest attributes for discrepancies |
| with the cipher suite, etc. |
| """ |
| self.raise_on_packet(TLSCertificateRequest, |
| self.HANDLED_CERTIFICATEREQUEST) |
| |
| @ATMT.condition(HANDLED_SERVERKEYEXCHANGE, prio=2) |
| def should_handle_CertificateRequest_from_ServerKeyExchange(self): |
| self.should_handle_CertificateRequest() |
| |
| @ATMT.condition(HANDLED_SERVERCERTIFICATE, prio=3) |
| def should_handle_CertificateRequest_from_ServerCertificate(self): |
| self.should_handle_CertificateRequest() |
| |
| def should_handle_ServerHelloDone(self): |
| self.raise_on_packet(TLSServerHelloDone, |
| self.HANDLED_SERVERHELLODONE) |
| |
| @ATMT.condition(HANDLED_SERVERKEYEXCHANGE, prio=1) |
| def should_handle_ServerHelloDone_from_ServerKeyExchange(self): |
| return self.should_handle_ServerHelloDone() |
| |
| @ATMT.condition(HANDLED_CERTIFICATEREQUEST, prio=4) |
| def should_handle_ServerHelloDone_from_CertificateRequest(self): |
| return self.should_handle_ServerHelloDone() |
| |
| @ATMT.condition(HANDLED_SERVERCERTIFICATE, prio=4) |
| def should_handle_ServerHelloDone_from_ServerCertificate(self): |
| return self.should_handle_ServerHelloDone() |
| |
| @ATMT.state() |
| def HANDLED_SERVERHELLODONE(self): |
| raise self.PREPARE_CLIENTFLIGHT2() |
| |
| @ATMT.state() |
| def PREPARE_CLIENTFLIGHT2(self): |
| self.add_record() |
| |
| @ATMT.condition(PREPARE_CLIENTFLIGHT2, prio=1) |
| def should_add_ClientCertificate(self): |
| """ |
| If the server sent a CertificateRequest, we send a Certificate message. |
| If no certificate is available, an empty Certificate message is sent: |
| - this is a SHOULD in RFC 4346 (Section 7.4.6) |
| - this is a MUST in RFC 5246 (Section 7.4.6) |
| |
| XXX We may want to add a complete chain. |
| """ |
| hs_msg = [type(m) for m in self.cur_session.handshake_messages_parsed] |
| if not TLSCertificateRequest in hs_msg: |
| return |
| certs = [] |
| if self.mycert: |
| certs = [self.mycert] |
| self.add_msg(TLSCertificate(certs=certs)) |
| raise self.ADDED_CLIENTCERTIFICATE() |
| |
| @ATMT.state() |
| def ADDED_CLIENTCERTIFICATE(self): |
| pass |
| |
| def should_add_ClientKeyExchange(self): |
| self.add_msg(TLSClientKeyExchange()) |
| raise self.ADDED_CLIENTKEYEXCHANGE() |
| |
| @ATMT.condition(PREPARE_CLIENTFLIGHT2, prio=2) |
| def should_add_ClientKeyExchange_from_ClientFlight2(self): |
| return self.should_add_ClientKeyExchange() |
| |
| @ATMT.condition(ADDED_CLIENTCERTIFICATE) |
| def should_add_ClientKeyExchange_from_ClientCertificate(self): |
| return self.should_add_ClientKeyExchange() |
| |
| @ATMT.state() |
| def ADDED_CLIENTKEYEXCHANGE(self): |
| pass |
| |
| @ATMT.condition(ADDED_CLIENTKEYEXCHANGE, prio=1) |
| def should_add_ClientVerify(self): |
| """ |
| XXX Section 7.4.7.1 of RFC 5246 states that the CertificateVerify |
| message is only sent following a client certificate that has signing |
| capability (i.e. not those containing fixed DH params). |
| We should verify that before adding the message. We should also handle |
| the case when the Certificate message was empty. |
| """ |
| hs_msg = [type(m) for m in self.cur_session.handshake_messages_parsed] |
| if (not TLSCertificateRequest in hs_msg or |
| self.mycert is None or |
| self.mykey is None): |
| return |
| self.add_msg(TLSCertificateVerify()) |
| raise self.ADDED_CERTIFICATEVERIFY() |
| |
| @ATMT.state() |
| def ADDED_CERTIFICATEVERIFY(self): |
| pass |
| |
| @ATMT.condition(ADDED_CERTIFICATEVERIFY) |
| def should_add_ChangeCipherSpec_from_CertificateVerify(self): |
| self.add_record() |
| self.add_msg(TLSChangeCipherSpec()) |
| raise self.ADDED_CHANGECIPHERSPEC() |
| |
| @ATMT.condition(ADDED_CLIENTKEYEXCHANGE, prio=2) |
| def should_add_ChangeCipherSpec_from_ClientKeyExchange(self): |
| self.add_record() |
| self.add_msg(TLSChangeCipherSpec()) |
| raise self.ADDED_CHANGECIPHERSPEC() |
| |
| @ATMT.state() |
| def ADDED_CHANGECIPHERSPEC(self): |
| pass |
| |
| @ATMT.condition(ADDED_CHANGECIPHERSPEC) |
| def should_add_ClientFinished(self): |
| self.add_record() |
| self.add_msg(TLSFinished()) |
| raise self.ADDED_CLIENTFINISHED() |
| |
| @ATMT.state() |
| def ADDED_CLIENTFINISHED(self): |
| pass |
| |
| @ATMT.condition(ADDED_CLIENTFINISHED) |
| def should_send_ClientFlight2(self): |
| self.flush_records() |
| raise self.SENT_CLIENTFLIGHT2() |
| |
| @ATMT.state() |
| def SENT_CLIENTFLIGHT2(self): |
| raise self.WAITING_SERVERFLIGHT2() |
| |
| @ATMT.state() |
| def WAITING_SERVERFLIGHT2(self): |
| self.get_next_msg() |
| raise self.RECEIVED_SERVERFLIGHT2() |
| |
| @ATMT.state() |
| def RECEIVED_SERVERFLIGHT2(self): |
| pass |
| |
| @ATMT.condition(RECEIVED_SERVERFLIGHT2) |
| def should_handle_ChangeCipherSpec(self): |
| self.raise_on_packet(TLSChangeCipherSpec, |
| self.HANDLED_CHANGECIPHERSPEC) |
| |
| @ATMT.state() |
| def HANDLED_CHANGECIPHERSPEC(self): |
| pass |
| |
| @ATMT.condition(HANDLED_CHANGECIPHERSPEC) |
| def should_handle_Finished(self): |
| self.raise_on_packet(TLSFinished, |
| self.HANDLED_SERVERFINISHED) |
| |
| @ATMT.state() |
| def HANDLED_SERVERFINISHED(self): |
| self.vprint("TLS handshake completed!") |
| self.vprint_sessioninfo() |
| self.vprint("You may send data or use 'quit'.") |
| |
| ####################### end of TLS handshake ############################## |
| |
| @ATMT.condition(HANDLED_SERVERFINISHED) |
| def should_wait_ClientData(self): |
| raise self.WAIT_CLIENTDATA() |
| |
| @ATMT.state() |
| def WAIT_CLIENTDATA(self): |
| pass |
| |
| @ATMT.condition(WAIT_CLIENTDATA, prio=1) |
| def add_ClientData(self): |
| """ |
| The user may type in: |
| GET / HTTP/1.1\r\nHost: testserver.com\r\n\r\n |
| Special characters are handled so that it becomes a valid HTTP request. |
| """ |
| if not self.data_to_send: |
| data = six.moves.input().replace('\\r', '\r').replace('\\n', '\n').encode() |
| else: |
| data = self.data_to_send.pop() |
| if data == b"quit": |
| return |
| if self.linebreak: |
| data += b"\n" |
| self.add_record() |
| self.add_msg(TLSApplicationData(data=data)) |
| raise self.ADDED_CLIENTDATA() |
| |
| @ATMT.condition(WAIT_CLIENTDATA, prio=2) |
| def no_more_ClientData(self): |
| raise self.CLOSE_NOTIFY() |
| |
| @ATMT.state() |
| def ADDED_CLIENTDATA(self): |
| pass |
| |
| @ATMT.condition(ADDED_CLIENTDATA) |
| def should_send_ClientData(self): |
| self.flush_records() |
| raise self.SENT_CLIENTDATA() |
| |
| @ATMT.state() |
| def SENT_CLIENTDATA(self): |
| raise self.WAITING_SERVERDATA() |
| |
| @ATMT.state() |
| def WAITING_SERVERDATA(self): |
| self.get_next_msg(0.3, 1) |
| raise self.RECEIVED_SERVERDATA() |
| |
| @ATMT.state() |
| def RECEIVED_SERVERDATA(self): |
| pass |
| |
| @ATMT.condition(RECEIVED_SERVERDATA, prio=1) |
| def should_handle_ServerData(self): |
| if not self.buffer_in: |
| raise self.WAIT_CLIENTDATA() |
| p = self.buffer_in[0] |
| if isinstance(p, TLSApplicationData): |
| print("> Received: %r" % p.data) |
| elif isinstance(p, TLSAlert): |
| print("> Received: %r" % p) |
| raise self.CLOSE_NOTIFY() |
| else: |
| print("> Received: %r" % p) |
| self.buffer_in = self.buffer_in[1:] |
| raise self.HANDLED_SERVERDATA() |
| |
| @ATMT.state() |
| def HANDLED_SERVERDATA(self): |
| raise self.WAIT_CLIENTDATA() |
| |
| @ATMT.state() |
| def CLOSE_NOTIFY(self): |
| self.vprint() |
| self.vprint("Trying to send a TLSAlert to the server...") |
| |
| @ATMT.condition(CLOSE_NOTIFY) |
| def close_session(self): |
| self.add_record() |
| self.add_msg(TLSAlert(level=1, descr=0)) |
| try: |
| self.flush_records() |
| except: |
| self.vprint("Could not send termination Alert, maybe the server stopped?") |
| raise self.FINAL() |
| |
| ########################## SSLv2 handshake ################################ |
| |
| @ATMT.state() |
| def SSLv2_PREPARE_CLIENTHELLO(self): |
| pass |
| |
| @ATMT.condition(SSLv2_PREPARE_CLIENTHELLO) |
| def sslv2_should_add_ClientHello(self): |
| self.add_record(is_sslv2=True) |
| p = self.client_hello or SSLv2ClientHello(challenge=randstring(16)) |
| self.add_msg(p) |
| raise self.SSLv2_ADDED_CLIENTHELLO() |
| |
| @ATMT.state() |
| def SSLv2_ADDED_CLIENTHELLO(self): |
| pass |
| |
| @ATMT.condition(SSLv2_ADDED_CLIENTHELLO) |
| def sslv2_should_send_ClientHello(self): |
| self.flush_records() |
| raise self.SSLv2_SENT_CLIENTHELLO() |
| |
| @ATMT.state() |
| def SSLv2_SENT_CLIENTHELLO(self): |
| raise self.SSLv2_WAITING_SERVERHELLO() |
| |
| @ATMT.state() |
| def SSLv2_WAITING_SERVERHELLO(self): |
| self.get_next_msg() |
| raise self.SSLv2_RECEIVED_SERVERHELLO() |
| |
| @ATMT.state() |
| def SSLv2_RECEIVED_SERVERHELLO(self): |
| pass |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERHELLO, prio=1) |
| def sslv2_should_handle_ServerHello(self): |
| self.raise_on_packet(SSLv2ServerHello, |
| self.SSLv2_HANDLED_SERVERHELLO) |
| |
| @ATMT.state() |
| def SSLv2_HANDLED_SERVERHELLO(self): |
| pass |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERHELLO, prio=2) |
| def sslv2_missing_ServerHello(self): |
| raise self.SSLv2_MISSING_SERVERHELLO() |
| |
| @ATMT.state() |
| def SSLv2_MISSING_SERVERHELLO(self): |
| self.vprint("Missing SSLv2 ServerHello message!") |
| raise self.SSLv2_CLOSE_NOTIFY() |
| |
| @ATMT.condition(SSLv2_HANDLED_SERVERHELLO) |
| def sslv2_should_add_ClientMasterKey(self): |
| self.add_record(is_sslv2=True) |
| self.add_msg(SSLv2ClientMasterKey()) |
| raise self.SSLv2_ADDED_CLIENTMASTERKEY() |
| |
| @ATMT.state() |
| def SSLv2_ADDED_CLIENTMASTERKEY(self): |
| pass |
| |
| @ATMT.condition(SSLv2_ADDED_CLIENTMASTERKEY) |
| def sslv2_should_send_ClientMasterKey(self): |
| self.flush_records() |
| raise self.SSLv2_SENT_CLIENTMASTERKEY() |
| |
| @ATMT.state() |
| def SSLv2_SENT_CLIENTMASTERKEY(self): |
| raise self.SSLv2_WAITING_SERVERVERIFY() |
| |
| @ATMT.state() |
| def SSLv2_WAITING_SERVERVERIFY(self): |
| # We give the server 0.5 second to send his ServerVerify. |
| # Else we assume that he's waiting for our ClientFinished. |
| self.get_next_msg(0.5, 0) |
| raise self.SSLv2_RECEIVED_SERVERVERIFY() |
| |
| @ATMT.state() |
| def SSLv2_RECEIVED_SERVERVERIFY(self): |
| pass |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERVERIFY, prio=1) |
| def sslv2_should_handle_ServerVerify(self): |
| self.raise_on_packet(SSLv2ServerVerify, |
| self.SSLv2_HANDLED_SERVERVERIFY, |
| get_next_msg=False) |
| |
| @ATMT.state() |
| def SSLv2_HANDLED_SERVERVERIFY(self): |
| pass |
| |
| def sslv2_should_add_ClientFinished(self): |
| hs_msg = [type(m) for m in self.cur_session.handshake_messages_parsed] |
| if SSLv2ClientFinished in hs_msg: |
| return |
| self.add_record(is_sslv2=True) |
| self.add_msg(SSLv2ClientFinished()) |
| raise self.SSLv2_ADDED_CLIENTFINISHED() |
| |
| @ATMT.condition(SSLv2_HANDLED_SERVERVERIFY, prio=1) |
| def sslv2_should_add_ClientFinished_from_ServerVerify(self): |
| return self.sslv2_should_add_ClientFinished() |
| |
| @ATMT.condition(SSLv2_HANDLED_SERVERVERIFY, prio=2) |
| def sslv2_should_wait_ServerFinished_from_ServerVerify(self): |
| raise self.SSLv2_WAITING_SERVERFINISHED() |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERVERIFY, prio=2) |
| def sslv2_should_add_ClientFinished_from_NoServerVerify(self): |
| return self.sslv2_should_add_ClientFinished() |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERVERIFY, prio=3) |
| def sslv2_missing_ServerVerify(self): |
| raise self.SSLv2_MISSING_SERVERVERIFY() |
| |
| @ATMT.state(final=True) |
| def SSLv2_MISSING_SERVERVERIFY(self): |
| self.vprint("Missing SSLv2 ServerVerify message!") |
| raise self.SSLv2_CLOSE_NOTIFY() |
| |
| @ATMT.state() |
| def SSLv2_ADDED_CLIENTFINISHED(self): |
| pass |
| |
| @ATMT.condition(SSLv2_ADDED_CLIENTFINISHED) |
| def sslv2_should_send_ClientFinished(self): |
| self.flush_records() |
| raise self.SSLv2_SENT_CLIENTFINISHED() |
| |
| @ATMT.state() |
| def SSLv2_SENT_CLIENTFINISHED(self): |
| hs_msg = [type(m) for m in self.cur_session.handshake_messages_parsed] |
| if SSLv2ServerVerify in hs_msg: |
| raise self.SSLv2_WAITING_SERVERFINISHED() |
| else: |
| self.get_next_msg() |
| raise self.SSLv2_RECEIVED_SERVERVERIFY() |
| |
| @ATMT.state() |
| def SSLv2_WAITING_SERVERFINISHED(self): |
| self.get_next_msg() |
| raise self.SSLv2_RECEIVED_SERVERFINISHED() |
| |
| @ATMT.state() |
| def SSLv2_RECEIVED_SERVERFINISHED(self): |
| pass |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERFINISHED, prio=1) |
| def sslv2_should_handle_ServerFinished(self): |
| self.raise_on_packet(SSLv2ServerFinished, |
| self.SSLv2_HANDLED_SERVERFINISHED) |
| |
| ####################### SSLv2 client authentication ####################### |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERFINISHED, prio=2) |
| def sslv2_should_handle_RequestCertificate(self): |
| self.raise_on_packet(SSLv2RequestCertificate, |
| self.SSLv2_HANDLED_REQUESTCERTIFICATE) |
| |
| @ATMT.state() |
| def SSLv2_HANDLED_REQUESTCERTIFICATE(self): |
| self.vprint("Server asked for a certificate...") |
| if not self.mykey or not self.mycert: |
| self.vprint("No client certificate to send!") |
| raise self.SSLv2_CLOSE_NOTIFY() |
| |
| @ATMT.condition(SSLv2_HANDLED_REQUESTCERTIFICATE) |
| def sslv2_should_add_ClientCertificate(self): |
| self.add_record(is_sslv2=True) |
| self.add_msg(SSLv2ClientCertificate(certdata=self.mycert)) |
| raise self.SSLv2_ADDED_CLIENTCERTIFICATE() |
| |
| @ATMT.state() |
| def SSLv2_ADDED_CLIENTCERTIFICATE(self): |
| pass |
| |
| @ATMT.condition(SSLv2_ADDED_CLIENTCERTIFICATE) |
| def sslv2_should_send_ClientCertificate(self): |
| self.flush_records() |
| raise self.SSLv2_SENT_CLIENTCERTIFICATE() |
| |
| @ATMT.state() |
| def SSLv2_SENT_CLIENTCERTIFICATE(self): |
| raise self.SSLv2_WAITING_SERVERFINISHED() |
| |
| ################### end of SSLv2 client authentication #################### |
| |
| @ATMT.state() |
| def SSLv2_HANDLED_SERVERFINISHED(self): |
| self.vprint("SSLv2 handshake completed!") |
| self.vprint_sessioninfo() |
| self.vprint("You may send data or use 'quit'.") |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERFINISHED, prio=3) |
| def sslv2_missing_ServerFinished(self): |
| raise self.SSLv2_MISSING_SERVERFINISHED() |
| |
| @ATMT.state() |
| def SSLv2_MISSING_SERVERFINISHED(self): |
| self.vprint("Missing SSLv2 ServerFinished message!") |
| raise self.SSLv2_CLOSE_NOTIFY() |
| |
| ######################## end of SSLv2 handshake ########################### |
| |
| @ATMT.condition(SSLv2_HANDLED_SERVERFINISHED) |
| def sslv2_should_wait_ClientData(self): |
| raise self.SSLv2_WAITING_CLIENTDATA() |
| |
| @ATMT.state() |
| def SSLv2_WAITING_CLIENTDATA(self): |
| pass |
| |
| @ATMT.condition(SSLv2_WAITING_CLIENTDATA, prio=1) |
| def sslv2_add_ClientData(self): |
| if not self.data_to_send: |
| data = six.moves.input().replace('\\r', '\r').replace('\\n', '\n').encode() |
| else: |
| data = self.data_to_send.pop() |
| self.vprint("> Read from list: %s" % data) |
| if data == "quit": |
| return |
| if self.linebreak: |
| data += "\n" |
| self.add_record(is_sslv2=True) |
| self.add_msg(Raw(data)) |
| raise self.SSLv2_ADDED_CLIENTDATA() |
| |
| @ATMT.condition(SSLv2_WAITING_CLIENTDATA, prio=2) |
| def sslv2_no_more_ClientData(self): |
| raise self.SSLv2_CLOSE_NOTIFY() |
| |
| @ATMT.state() |
| def SSLv2_ADDED_CLIENTDATA(self): |
| pass |
| |
| @ATMT.condition(SSLv2_ADDED_CLIENTDATA) |
| def sslv2_should_send_ClientData(self): |
| self.flush_records() |
| raise self.SSLv2_SENT_CLIENTDATA() |
| |
| @ATMT.state() |
| def SSLv2_SENT_CLIENTDATA(self): |
| raise self.SSLv2_WAITING_SERVERDATA() |
| |
| @ATMT.state() |
| def SSLv2_WAITING_SERVERDATA(self): |
| self.get_next_msg(0.3, 1) |
| raise self.SSLv2_RECEIVED_SERVERDATA() |
| |
| @ATMT.state() |
| def SSLv2_RECEIVED_SERVERDATA(self): |
| pass |
| |
| @ATMT.condition(SSLv2_RECEIVED_SERVERDATA) |
| def sslv2_should_handle_ServerData(self): |
| if not self.buffer_in: |
| raise self.SSLv2_WAITING_CLIENTDATA() |
| p = self.buffer_in[0] |
| print("> Received: %r" % p.load) |
| if p.load.startswith(b"goodbye"): |
| raise self.SSLv2_CLOSE_NOTIFY() |
| self.buffer_in = self.buffer_in[1:] |
| raise self.SSLv2_HANDLED_SERVERDATA() |
| |
| @ATMT.state() |
| def SSLv2_HANDLED_SERVERDATA(self): |
| raise self.SSLv2_WAITING_CLIENTDATA() |
| |
| @ATMT.state() |
| def SSLv2_CLOSE_NOTIFY(self): |
| """ |
| There is no proper way to end an SSLv2 session. |
| We try and send a 'goodbye' message as a substitute. |
| """ |
| self.vprint() |
| self.vprint("Trying to send a 'goodbye' to the server...") |
| |
| @ATMT.condition(SSLv2_CLOSE_NOTIFY) |
| def sslv2_close_session(self): |
| self.add_record() |
| self.add_msg(Raw('goodbye')) |
| try: |
| self.flush_records() |
| except: |
| self.vprint("Could not send our goodbye. The server probably stopped.") |
| self.socket.close() |
| raise self.FINAL() |
| |
| ######################### TLS 1.3 handshake ############################### |
| |
| @ATMT.state() |
| def TLS13_START(self): |
| pass |
| |
| @ATMT.condition(TLS13_START) |
| def tls13_should_add_ClientHello(self): |
| # we have to use the legacy, plaintext TLS record here |
| self.add_record(is_tls13=False) |
| if self.client_hello: |
| p = self.client_hello |
| else: |
| # When trying to connect to a public TLS 1.3 server, |
| # you will most likely need to provide an SNI extension. |
| #sn = ServerName(servername="<put server name here>") |
| ext = [TLS_Ext_SupportedGroups(groups=["secp256r1"]), |
| #TLS_Ext_ServerName(servernames=[sn]), |
| TLS_Ext_KeyShare_CH(client_shares=[KeyShareEntry(group=23)]), |
| TLS_Ext_SupportedVersions(versions=["TLS 1.3-d18"]), |
| TLS_Ext_SignatureAlgorithms(sig_algs=["sha256+rsapss", |
| "sha256+rsa"]) ] |
| p = TLSClientHello(ciphers=0x1301, ext=ext) |
| self.add_msg(p) |
| raise self.TLS13_ADDED_CLIENTHELLO() |
| |
| @ATMT.state() |
| def TLS13_ADDED_CLIENTHELLO(self): |
| pass |
| |
| @ATMT.condition(TLS13_ADDED_CLIENTHELLO) |
| def tls13_should_send_ClientHello(self): |
| self.flush_records() |
| raise self.TLS13_SENT_CLIENTHELLO() |
| |
| @ATMT.state() |
| def TLS13_SENT_CLIENTHELLO(self): |
| raise self.TLS13_WAITING_SERVERHELLO() |
| |
| @ATMT.state() |
| def TLS13_WAITING_SERVERHELLO(self): |
| self.get_next_msg() |
| |
| @ATMT.condition(TLS13_WAITING_SERVERHELLO) |
| def tls13_should_handle_ServerHello(self): |
| self.raise_on_packet(TLS13ServerHello, |
| self.TLS13_WAITING_ENCRYPTEDEXTENSIONS) |
| |
| @ATMT.state() |
| def TLS13_WAITING_ENCRYPTEDEXTENSIONS(self): |
| self.get_next_msg() |
| |
| @ATMT.condition(TLS13_WAITING_ENCRYPTEDEXTENSIONS) |
| def tls13_should_handle_EncryptedExtensions(self): |
| self.raise_on_packet(TLSEncryptedExtensions, |
| self.TLS13_WAITING_CERTIFICATE) |
| |
| @ATMT.state() |
| def TLS13_WAITING_CERTIFICATE(self): |
| self.get_next_msg() |
| |
| @ATMT.condition(TLS13_WAITING_CERTIFICATE, prio=1) |
| def tls13_should_handle_Certificate(self): |
| self.raise_on_packet(TLS13Certificate, |
| self.TLS13_WAITING_CERTIFICATEVERIFY) |
| |
| @ATMT.condition(TLS13_WAITING_CERTIFICATE, prio=2) |
| def tls13_should_handle_CertificateRequest(self): |
| hs_msg = [type(m) for m in self.cur_session.handshake_messages_parsed] |
| if TLSCertificateRequest in hs_msg: |
| self.vprint("TLSCertificateRequest already received!") |
| self.raise_on_packet(TLSCertificateRequest, |
| self.TLS13_WAITING_CERTIFICATE) |
| |
| @ATMT.condition(TLS13_WAITING_CERTIFICATE, prio=3) |
| def tls13_should_handle_ServerFinished_from_EncryptedExtensions(self): |
| self.raise_on_packet(TLSFinished, |
| self.TLS13_CONNECTED) |
| |
| @ATMT.condition(TLS13_WAITING_CERTIFICATE, prio=4) |
| def tls13_missing_Certificate(self): |
| self.vprint("Missing TLS 1.3 message after EncryptedExtensions!") |
| raise self.FINAL() |
| |
| @ATMT.state() |
| def TLS13_WAITING_CERTIFICATEVERIFY(self): |
| self.get_next_msg() |
| |
| @ATMT.condition(TLS13_WAITING_CERTIFICATEVERIFY) |
| def tls13_should_handle_CertificateVerify(self): |
| self.raise_on_packet(TLSCertificateVerify, |
| self.TLS13_WAITING_SERVERFINISHED) |
| |
| @ATMT.state() |
| def TLS13_WAITING_SERVERFINISHED(self): |
| self.get_next_msg() |
| |
| @ATMT.condition(TLS13_WAITING_SERVERFINISHED) |
| def tls13_should_handle_ServerFinished_from_CertificateVerify(self): |
| self.raise_on_packet(TLSFinished, |
| self.TLS13_PREPARE_CLIENTFLIGHT2) |
| |
| @ATMT.state() |
| def TLS13_PREPARE_CLIENTFLIGHT2(self): |
| self.add_record(is_tls13=True) |
| #raise self.FINAL() |
| |
| @ATMT.condition(TLS13_PREPARE_CLIENTFLIGHT2) |
| def tls13_should_add_ClientFinished(self): |
| self.add_msg(TLSFinished()) |
| raise self.TLS13_ADDED_CLIENTFINISHED() |
| |
| @ATMT.state() |
| def TLS13_ADDED_CLIENTFINISHED(self): |
| pass |
| |
| @ATMT.condition(TLS13_ADDED_CLIENTFINISHED) |
| def tls13_should_send_ClientFlight2(self): |
| self.flush_records() |
| raise self.TLS13_SENT_CLIENTFLIGHT2() |
| |
| @ATMT.state() |
| def TLS13_SENT_CLIENTFLIGHT2(self): |
| raise self.HANDLED_SERVERFINISHED() |
| |
| @ATMT.state(final=True) |
| def FINAL(self): |
| # We might call shutdown, but it may happen that the server |
| # did not wait for us to shutdown after answering our data query. |
| #self.socket.shutdown(1) |
| self.vprint("Closing client socket...") |
| self.socket.close() |
| self.vprint("Ending TLS client automaton.") |
| |
