diff --git a/.config/ci/check_commits.sh b/.config/ci/check_commits.sh index f5efc667c48..6f052e8787e 100755 --- a/.config/ci/check_commits.sh +++ b/.config/ci/check_commits.sh @@ -6,7 +6,7 @@ # We copy Wireshark's contributing guide, thanks to them for the idea ! # This script is inspired by https://gitlab.com/wireshark/wireshark/-/blob/master/.gitlab-ci.yml -commits=$(git rev-list --no-merges --after="2026-01-00T00:00:00" --max-count=$((PR_FETCH_DEPTH - 1)) HEAD) +commits=$(git rev-list --no-merges --after="2026-01-00T00:00:00" master..HEAD) if [ -z "$commits" ]; then echo "No commit to check in PR. OK." exit 0 diff --git a/.github/workflows/unittests.yml b/.github/workflows/unittests.yml index d36b4008bc7..59129326829 100644 --- a/.github/workflows/unittests.yml +++ b/.github/workflows/unittests.yml @@ -29,6 +29,10 @@ jobs: uses: actions/checkout@v6 with: fetch-depth: ${{ env.PR_FETCH_DEPTH }} + fetch-tags: true + if: github.event_name == 'pull_request' && !contains(github.event.pull_request.labels.*.name, 'skip-ai-check') + - name: Checkout master tag + run: git fetch origin master:master if: github.event_name == 'pull_request' && !contains(github.event.pull_request.labels.*.name, 'skip-ai-check') - name: AI trailer reminder run: bash ./.config/ci/check_commits.sh diff --git a/scapy/asn1/mib.py b/scapy/asn1/mib.py index ed25c6e95b6..738099d80c3 100644 --- a/scapy/asn1/mib.py +++ b/scapy/asn1/mib.py @@ -275,6 +275,9 @@ def load_mib(filenames): "2.16.840.1.101.3.4.2.10": "sha3-512", "2.16.840.1.101.3.4.2.11": "shake128", "2.16.840.1.101.3.4.2.12": "shake256", + "2.16.840.1.101.3.4.3.17": "ml-dsa-44", + "2.16.840.1.101.3.4.3.18": "ml-dsa-65", + "2.16.840.1.101.3.4.3.19": "ml-dsa-87", } # thawte # diff --git a/scapy/automaton.py b/scapy/automaton.py index f26370b102f..9c4ac93d35e 100644 --- a/scapy/automaton.py +++ b/scapy/automaton.py @@ -1260,7 +1260,7 @@ def __iter__(self): def __del__(self): # type: () -> None - self.destroy() + self.destroy(force=True) def _run_condition(self, cond, *args, **kargs): # type: (_StateWrapper, Any, Any) -> None @@ -1289,12 +1289,27 @@ def _do_start(self, *args, **kargs): target=self._do_control, args=(ready,) + (args), kwargs=kargs, - name="scapy.automaton _do_start" + name="scapy.automaton _do_control" ) _t.daemon = True _t.start() ready.wait() + def _transfer_error(self, ex: Exception, exc_info: Any) -> None: + """ + This transfers the error to the main thread. + """ + self.debug(3, "Transferring exception from tid=%i:\n%s" % ( + self.threadid or 0, + "".join(traceback.format_exception(*exc_info)) + )) + m = Message( + type=_ATMT_Command.EXCEPTION, + exception=ex, + exc_info=exc_info, + ) + self.cmdout.send(m) + def _do_control(self, ready, *args, **kargs): # type: (threading.Event, Any, Any) -> None with self.started: @@ -1306,20 +1321,26 @@ def _do_control(self, ready, *args, **kargs): a = args + self.init_args[len(args):] k = self.init_kargs.copy() k.update(kargs) - self.parse_args(*a, **k) - - # Start the automaton - self.state = self.initial_states[0](self) - self.send_sock = self.sock or self.send_sock_class(**self.socket_kargs) - if self.recv_conditions: - # Only start a receiving socket if we have at least one recv_conditions - self.listen_sock = self.sock or self.recv_sock_class(**self.socket_kargs) # noqa: E501 + + # Initialize the automaton self.packets = PacketList(name="session[%s]" % self.__class__.__name__) + try: + self.parse_args(*a, **k) + self.state = self.initial_states[0](self) + self.send_sock = self.sock or self.send_sock_class(**self.socket_kargs) + if self.recv_conditions: + # Only start a receiving socket if we have at least one + # recv_conditions + self.listen_sock = self.sock or self.recv_sock_class(**self.socket_kargs) # noqa: E501 + except Exception as e: + self._transfer_error(e, sys.exc_info()) + ready.set() + return + # Main loop of the control thread singlestep = True iterator = self._do_iter() self.debug(3, "Starting control thread [tid=%i]" % self.threadid) - # Sync threads ready.set() try: while True: @@ -1360,10 +1381,7 @@ def _do_control(self, ready, *args, **kargs): result=self.final_state_output) self.cmdout.send(c) except Exception as e: - exc_info = sys.exc_info() - self.debug(3, "Transferring exception from tid=%i:\n%s" % (self.threadid, "".join(traceback.format_exception(*exc_info)))) # noqa: E501 - m = Message(type=_ATMT_Command.EXCEPTION, exception=e, exc_info=exc_info) # noqa: E501 - self.cmdout.send(m) + self._transfer_error(e, sys.exc_info()) self.debug(3, "Stopping control thread (tid=%i)" % self.threadid) self.threadid = None if self.listen_sock: @@ -1569,15 +1587,17 @@ def _flush_inout(self): for cmd in [self.cmdin, self.cmdout]: cmd.clear() - def destroy(self): - # type: () -> None + def destroy(self, force=False): + # type: (bool) -> None """ Destroys a stopped Automaton: this cleanups all opened file descriptors. + The automaton will not be able to be restarted. + Required on PyPy for instance where the garbage collector behaves differently. """ if not hasattr(self, "started"): return # was never started. - if self.isrunning(): + if not force and self.isrunning(): raise ValueError("Can't close running Automaton ! Call stop() beforehand") # Close command pipes self.cmdin.close() diff --git a/scapy/layers/tls/automaton.py b/scapy/layers/tls/automaton.py index 9395388de91..647ad5280b4 100644 --- a/scapy/layers/tls/automaton.py +++ b/scapy/layers/tls/automaton.py @@ -93,6 +93,9 @@ def parse_args(self, mycert=None, mykey=None, **kargs): else: self.mykey = None + # Output of the automaton (final status) + self.final_reason = None + def get_next_msg(self, socket_timeout=2, retry=2): """ The purpose of the function is to make next message(s) available in diff --git a/scapy/layers/tls/automaton_cli.py b/scapy/layers/tls/automaton_cli.py index ff0e9125ff9..5cdaf42142a 100644 --- a/scapy/layers/tls/automaton_cli.py +++ b/scapy/layers/tls/automaton_cli.py @@ -47,6 +47,7 @@ from scapy.error import warning from scapy.layers.tls.automaton import _TLSAutomaton from scapy.layers.tls.basefields import _tls_version, _tls_version_options +from scapy.layers.tls.cert import CertList, CertTree from scapy.layers.tls.session import tlsSession from scapy.layers.tls.extensions import ( ServerName, @@ -97,6 +98,9 @@ class TLSClientAutomaton(_TLSAutomaton): :param server: the server IP or hostname. defaults to 127.0.0.1 :param dport: the server port. defaults to 4433 :param server_name: the SNI to use. It does not need to be set + :param cafile: optional CA certificate bundle used to authenticate the server. + By default, the system trust store is used (if verify_server is set to True). + :param verify_server: whether to verify the server certificate. False by default. :param mycert: :param mykey: may be provided as filenames. They will be used in the (or post) handshake, should the server ask for client authentication. @@ -116,6 +120,7 @@ class TLSClientAutomaton(_TLSAutomaton): """ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, + cafile=None, verify_server=False, mycert=None, mykey=None, client_hello=None, version=None, resumption_master_secret=None, @@ -137,6 +142,14 @@ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, self.remote_ip = tmp[0][4][0] self.remote_port = dport self.server_name = server_name + self.expected_server_name = server_name or server + self.verify_server = verify_server or bool(cafile) + if self.verify_server and cafile: + self.server_trust_store = CertList(cafile) + elif self.verify_server: + self.server_trust_store = CertList.load_system_store() + else: + self.server_trust_store = None self.local_ip = None self.local_port = None self.socket = None @@ -400,9 +413,30 @@ def should_handle_ServerCertificate(self): self.HANDLED_SERVERCERTIFICATE) raise self.HANDLED_SERVERCERTIFICATE() + def _verify_server_cert(self): + if self.verify_server: + try: + CertTree( + self.cur_session.server_certs, + self.server_trust_store, + ).verify( + self.cur_session.server_certs[0], + hostname=self.server_name or self.remote_ip, + ) + except ValueError: + raise self.INVALID_SERVER_CERTIFICATE() + @ATMT.state() def HANDLED_SERVERCERTIFICATE(self): - pass + self._verify_server_cert() + + @ATMT.state() + def INVALID_SERVER_CERTIFICATE(self): + self.vprint("Server certificate verification failed!") + self.add_record() + self.add_msg(TLSAlert(level=2, descr=46)) + self.flush_records() + raise self.FINAL(reason=self.INVALID_SERVER_CERTIFICATE) @ATMT.condition(HANDLED_SERVERHELLO, prio=2) def missing_ServerCertificate(self): @@ -429,9 +463,9 @@ def should_handle_ServerKeyExchange_from_ServerCertificate(self): self.raise_on_packet(TLSServerKeyExchange, self.HANDLED_SERVERKEYEXCHANGE) - @ATMT.state(final=True) + @ATMT.state() def MISSING_SERVERKEYEXCHANGE(self): - pass + raise self.FINAL(reason=self.MISSING_SERVERKEYEXCHANGE) @ATMT.condition(HANDLED_SERVERCERTIFICATE, prio=2) def missing_ServerKeyExchange(self): @@ -798,7 +832,7 @@ def close_session(self): self.flush_records() except Exception: self.vprint("Could not send termination Alert, maybe the server stopped?") # noqa: E501 - raise self.FINAL() + raise self.FINAL(reason=self.CLOSE_NOTIFY) # SSLv2 handshake # @@ -842,7 +876,10 @@ def sslv2_should_handle_ServerHello(self): @ATMT.state() def SSLv2_HANDLED_SERVERHELLO(self): - pass + try: + self._verify_server_cert() + except self.INVALID_SERVER_CERTIFICATE: + raise self.SSLv2_CLOSE_NOTIFY() @ATMT.condition(SSLv2_RECEIVED_SERVERHELLO, prio=2) def sslv2_missing_ServerHello(self): @@ -922,7 +959,7 @@ def sslv2_should_add_ClientFinished_from_NoServerVerify(self): def sslv2_missing_ServerVerify(self): raise self.SSLv2_MISSING_SERVERVERIFY() - @ATMT.state(final=True) + @ATMT.state() def SSLv2_MISSING_SERVERVERIFY(self): self.vprint("Missing SSLv2 ServerVerify message!") raise self.SSLv2_CLOSE_NOTIFY() @@ -1092,7 +1129,7 @@ def sslv2_close_session(self): except Exception: self.vprint("Could not send our goodbye. The server probably stopped.") # noqa: E501 self.socket.close() - raise self.FINAL() + raise self.FINAL(reason=self.SSLv2_CLOSE_NOTIFY) # TLS 1.3 handshake # @@ -1347,7 +1384,7 @@ def tls13_should_handle_Certificate(self): @ATMT.state() def TLS13_HANDLED_CERTIFICATE(self): - pass + self._verify_server_cert() @ATMT.condition(TLS13_HANDLED_CERTIFICATE, prio=1) def tls13_should_handle_CertificateVerify(self): @@ -1370,7 +1407,7 @@ def TLS13_INVALID_CERTIFICATE_VERIFY(self): self.add_record() self.add_msg(TLSAlert(level=2, descr=51)) self.flush_records() - raise self.FINAL() + raise self.FINAL(reason=self.TLS13_INVALID_CERTIFICATE_VERIFY) @ATMT.condition(TLS13_HANDLED_CERTIFICATE_VERIFY, prio=1) def tls13_should_handle_finished(self): @@ -1466,7 +1503,7 @@ def TLS13_SENT_CLIENTFLIGHT2(self): @ATMT.state() def SOCKET_CLOSED(self): - raise self.FINAL() + raise self.FINAL(reason=self.SOCKET_CLOSED) @ATMT.state(stop=True) def STOP(self): @@ -1477,10 +1514,12 @@ def STOP(self): raise self.CLOSE_NOTIFY() @ATMT.state(final=True) - def FINAL(self): + def FINAL(self, reason=None): # 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() + if reason is not None: + self.final_reason = reason self.vprint("Ending TLS client automaton.") diff --git a/scapy/layers/tls/automaton_srv.py b/scapy/layers/tls/automaton_srv.py index 769d6c4b45b..384d4a5bc8c 100644 --- a/scapy/layers/tls/automaton_srv.py +++ b/scapy/layers/tls/automaton_srv.py @@ -30,7 +30,7 @@ from scapy.automaton import ATMT from scapy.error import warning from scapy.layers.tls.automaton import _TLSAutomaton -from scapy.layers.tls.cert import PrivKeyRSA, PrivKeyECDSA, PrivKeyEdDSA +from scapy.layers.tls.cert import PrivKeyRSA, PrivKeyECDSA, PrivKeyEdDSA, CertList from scapy.layers.tls.basefields import _tls_version from scapy.layers.tls.session import tlsSession from scapy.layers.tls.crypto.groups import _tls_named_groups @@ -86,7 +86,8 @@ class TLSServerAutomaton(_TLSAutomaton): """ A simple TLS test server automaton. Try to overload some states or - conditions and see what happens on the other side. + conditions and see what happens on the other side. This server + can only serve a SINGLE CLIENT at a time. Because of socket and automaton limitations, for now, the best way to interrupt the server is by sending him 'stop_server'. Interruptions with @@ -96,15 +97,17 @@ class TLSServerAutomaton(_TLSAutomaton): message in a SSLv2 version, he will close the client session with a similar message, and start waiting for new client connections. - _'mycert' and 'mykey' may be provided as filenames. They are needed for any - server authenticated handshake. - _'preferred_ciphersuite' allows the automaton to choose a cipher suite when - offered in the ClientHello. If absent, another one will be chosen. - _'client_auth' means the client has to provide a certificate. - _'is_echo_server' means that everything received will be sent back. - _'max_client_idle_time' is the maximum silence duration from the client. - Once this limit has been reached, the client (if still here) is dropped, - and we wait for a new connection. + :param mycert: the Cert or path to certificate + :param mykey: the PrivKey or path to private key + :param preferred_ciphersuite: a ciphersuite to prefer in the client hello + :param client_auth: whether the client has to provide a certificate + + Server behavior: + + :param is_echo_server: means that everything received will be sent back + :param max_client_idle_time: is the maximum silence duration from the client. + Once this limit has been reached, the client (if still here) is dropped, + and we wait for a new connection. """ def parse_args(self, server="127.0.0.1", sport=4433, @@ -232,6 +235,26 @@ def http_sessioninfo(self): answer = (header + body) % len(body) return answer + def stop(self, wait=True): + # We shutdown the server socket so that WAITING_CLIENT exits. + try: + self.serversocket.shutdown(socket.SHUT_RDWR) + self.serversocket.close() + except OSError: + pass + self.serversocket.close() + return super(TLSServerAutomaton, self).stop(wait=wait) + + def forcestop(self, wait=True): + # We shutdown the server socket so that WAITING_CLIENT exits. + try: + self.serversocket.shutdown(socket.SHUT_RDWR) + self.serversocket.close() + except OSError: + pass + self.serversocket.close() + return super(TLSServerAutomaton, self).forcestop(wait=wait) + @ATMT.state(initial=True) def INITIAL(self): self.vprint("Starting TLS server automaton.") @@ -257,7 +280,7 @@ def BIND(self): self.vprint(m) self.vprint("Maybe some server is already listening there?") self.vprint() - raise self.FINAL() + raise self.FINAL(reason=self.BIND) raise self.WAITING_CLIENT() @ATMT.state() @@ -272,7 +295,11 @@ def WAITING_CLIENT(self): self.vprint() self.vprint("Waiting for a new client on %s:%d" % (self.local_ip, self.local_port)) - self.socket, addr = self.serversocket.accept() + try: + self.socket, addr = self.serversocket.accept() + except OSError: + self.vprint("Wait was cancelled: server socket was closed.") + raise self.FINAL() if not isinstance(addr, tuple): addr = self.socket.getpeername() if len(addr) > 2: @@ -291,7 +318,7 @@ def INIT_TLS_SESSION(self): every server_key with both server_rsa_key and server_ecdsa_key. """ self.cur_session = tlsSession(connection_end="server") - self.cur_session.server_certs = [self.mycert] + self.cur_session.server_certs = CertList([self.mycert]) self.cur_session.server_key = self.mykey if isinstance(self.mykey, PrivKeyRSA): self.cur_session.server_rsa_key = self.mykey @@ -1209,7 +1236,7 @@ def close_session_final(self): # We might call shutdown, but unit tests with s_client fail with this # self.socket.shutdown(1) self.socket.close() - raise self.FINAL() + raise self.FINAL(reason=self.CLOSE_NOTIFY_FINAL) # SSLv2 handshake # @@ -1511,10 +1538,12 @@ def sslv2_close_session_final(self): except Exception: self.vprint("Could not send our goodbye. The client probably left.") # noqa: E501 self.socket.close() - raise self.FINAL() + raise self.FINAL(reason=self.SSLv2_CLOSE_NOTIFY_FINAL) @ATMT.state(stop=True, final=True) - def FINAL(self): + def FINAL(self, reason=None): self.vprint("Closing server socket...") self.serversocket.close() + if reason is not None: + self.final_reason = reason self.vprint("Ending TLS server automaton.") diff --git a/scapy/layers/tls/cert.py b/scapy/layers/tls/cert.py index 5b1dc565bd8..cb9a1cb5864 100644 --- a/scapy/layers/tls/cert.py +++ b/scapy/layers/tls/cert.py @@ -92,15 +92,19 @@ """ import base64 +import calendar import enum import os +import socket +import ssl import time import warnings from scapy.config import conf, crypto_validator -from scapy.compat import Self -from scapy.error import warning +from scapy.compat import Self, plain_str +from scapy.error import log_runtime, warning from scapy.utils import binrepr +from scapy.pton_ntop import inet_ntop from scapy.asn1.asn1 import ( ASN1_BIT_STRING, ASN1_NULL, @@ -134,6 +138,8 @@ X509_AttributeValue, X509_Cert, X509_CRL, + X509_DNSName, + X509_IPAddress, X509_SubjectPublicKeyInfo, ) from scapy.layers.tls.crypto.hash import _get_hash @@ -943,6 +949,22 @@ def _get_csr_sig_hashname(csr): return hash_by_oid[sigAlg.algorithm.val] +def _match_dns_name(pattern, hostname): + """ + Whether a certificate DNS name matches a hostname, RFC 6125 sect 6.4.3. + + :param pattern: a dNSName from the certificate + :param hostname: the name the client asked for + :return: True if they match + """ + pattern = pattern.lower().rstrip(".") + hostname = hostname.lower().rstrip(".") + if pattern[:2] == "*.": + return pattern[2:] == hostname.split(".", 1)[-1] + else: + return pattern == hostname + + class Cert(metaclass=_CertMaker): """ Wrapper for the X509_Cert from layers/x509.py. @@ -985,6 +1007,7 @@ def import_from_asn1pkt(self, cert): self.notAfter_str_simple = time.strftime("%x", self.notAfter) self.pubkey = PubKey(bytes(tbsCert.subjectPublicKeyInfo)) + self.subjectAltName = None if tbsCert.extensions: for extn in tbsCert.extensions: @@ -998,6 +1021,8 @@ def import_from_asn1pkt(self, cert): self.extKeyUsage = extn.extnValue.get_extendedKeyUsage() elif extn.extnID.oidname == "authorityKeyIdentifier": self.authorityKeyID = extn.extnValue.keyIdentifier.val + elif extn.extnID.oidname == "subjectAltName": + self.subjectAltName = extn.extnValue.subjectAltName self.signatureValue = bytes(cert.signatureValue) self.signatureLen = len(self.signatureValue) @@ -1097,6 +1122,63 @@ def remainingDays(self, now=None): diff = (nft - now) / (24.0 * 3600) return diff + def isValidAt(self, now=None): + """ + Whether the current time falls inside the certificate's validity period. + + The comparison is made in UTC, which is how notBefore and notAfter are + stored. (:func:`remainingDays` compares in local time and so is off by + the local UTC offset.) + + :param now: (optional) a UTC time tuple to compare against, defaulting + to the current time + :return: True if the certificate is neither expired nor not yet valid + """ + if now is None: + now = time.gmtime() + now = calendar.timegm(now) + return ( + calendar.timegm(self.notBefore) <= now <= + calendar.timegm(self.notAfter) + ) + + def matchesHostname(self, hostname): + """ + Whether this certificate was issued to the given host. + + Names come from the subjectAltName extension. RFC 6125 sect 6.4.4 says + the Common Name is only consulted when there is no subjectAltName at + all, and that is what happens here. + + :param hostname: the DNS name or IP address the client asked for + :return: True if the certificate names that host + """ + if not hostname: + return False + hostname = plain_str(hostname) + + if self.subjectAltName is not None: + # Check subjectAltName + for generalName in self.subjectAltName: + name = generalName.generalName + if isinstance(name, X509_DNSName): + if _match_dns_name(plain_str(name.dNSName.val), hostname): + return True + elif isinstance(name, X509_IPAddress): + raw = name.iPAddress.val + if len(raw) == 4: + if inet_ntop(socket.AF_INET, raw) == hostname: + return True + elif len(raw) == 16: + if inet_ntop(socket.AF_INET6, raw) == hostname: + return True + return False + else: + # Check Common Name + if "commonName" in self.subject: + return _match_dns_name(self.subject["commonName"], hostname) + return False + def isRevoked(self, crl_list): """ Given a list of trusted CRL (their signature has already been @@ -1497,6 +1579,22 @@ def __init__( super(CertList, self).__init__(certList) + @classmethod + def load_system_store(cls): + """ + Return a CertList containing the default trusted system store. + """ + context = ssl.create_default_context() + certs = [] + for der in context.get_ca_certs(binary_form=True): + try: + certs.append(Cert(der)) + except Exception as ex: + log_runtime.error("Failed loading cert.") + log_runtime.error(der2pem(der)) + raise ex + return cls(certs) + def findCertBySid(self, sid): """ Find a certificate in the list by SubjectIDentifier. @@ -1662,13 +1760,46 @@ def _rec_getchain(chain, curtree): else: return None - def verify(self, cert): + def verify(self, cert, hostname=None, now=None, allow_expired=False): """ - Verify that a certificate is properly signed. + Verify that a certificate is properly signed, current, and the right one. + + Raises ValueError when the certificate fails any of the checks. + + :param cert: the certificate to verify + :param hostname: (optional) the DNS name or IP address the peer was + expected to be. Without it the identity of the peer is not checked, + so any certificate the store can chain is accepted. + :param now: (optional) a UTC time tuple to check validity against, + defaulting to the current time + :param allow_expired: (optional) accept a chain that is outside its + validity period. Useful when checking a signature made while the + certificate was still valid, or against a stored capture, where + the dates say nothing about whether the signature was genuine. """ # Check that we can find a chain to this certificate - if not self.getchain(cert): + chain = self.getchain(cert) + if not chain: raise ValueError("Certificate verification failed !") + # Nothing in the chain may have expired or be in the future: an issuer + # that is out of date does not vouch for anything below it. A chain can + # also hold a CSR, which has no validity period to check. + for c in chain: + if not isinstance(c, Cert): + continue + if not allow_expired and not c.isValidAt(now): + raise ValueError( + "Certificate %s is outside its validity period " + "(%s to %s) !" % ( + c.subject_str, c.notBefore_str, c.notAfter_str + ) + ) + if hostname is not None and not cert.matchesHostname(hostname): + raise ValueError( + "Certificate %s was not issued to %s !" % ( + cert.subject_str, plain_str(hostname) + ) + ) def show(self, ret: bool = False): """ @@ -1895,6 +2026,7 @@ def verify( eContentType: Optional[ASN1_OID] = None, eContent: Optional[bytes] = None, no_verify_cert: bool = False, + allow_expired: bool = False, ): """ Verify a CMS message against the list of trusted certificates, @@ -1904,6 +2036,10 @@ def verify( :param eContentType: if provided, verifies that the content type is valid :param eContent: in PKCS 7.1, provide the content to verify :param no_verify_cert: do not check the remote certificate (unsafe) + :param allow_expired: accept a signer certificate that is outside its + validity period. A CMS signature is often checked long after it was + made, so an expired signer does not by itself mean the signature is + not genuine. """ if contentInfo.contentType.oidname != "id-signedData": raise ValueError("ContentInfo isn't signed !") @@ -1933,7 +2069,7 @@ def verify( # Verify certificate signature if not no_verify_cert: - certTree.verify(cert) + certTree.verify(cert, allow_expired=allow_expired) # Verify the message hash if signerInfo.signedAttrs: diff --git a/scapy/layers/tls/handshake.py b/scapy/layers/tls/handshake.py index abde901a246..847ae10d894 100644 --- a/scapy/layers/tls/handshake.py +++ b/scapy/layers/tls/handshake.py @@ -39,7 +39,7 @@ from scapy.packet import Packet, Raw, Padding from scapy.utils import randstring, repr_hex from scapy.layers.x509 import OCSP_Response -from scapy.layers.tls.cert import Cert +from scapy.layers.tls.cert import Cert, CertList from scapy.layers.tls.basefields import (_tls_version, _TLSVersionField, _TLSClientVersionField) from scapy.layers.tls.extensions import (_ExtensionsLenField, _ExtensionsField, @@ -969,9 +969,9 @@ def post_dissection_tls_session_update(self, msg_str): self.tls_session_update(msg_str) connection_end = self.tls_session.connection_end if connection_end == "client": - self.tls_session.server_certs = [x[1] for x in self.certs] + self.tls_session.server_certs = CertList([x[1] for x in self.certs]) else: - self.tls_session.client_certs = [x[1] for x in self.certs] + self.tls_session.client_certs = CertList([x[1] for x in self.certs]) class _ASN1CertAndExt(_GenericTLSSessionInheritance): diff --git a/scapy/layers/tls/handshake_sslv2.py b/scapy/layers/tls/handshake_sslv2.py index 1004280ec54..48dfb6ad732 100644 --- a/scapy/layers/tls/handshake_sslv2.py +++ b/scapy/layers/tls/handshake_sslv2.py @@ -15,7 +15,7 @@ ShortEnumField, StrLenField, XStrField, XStrLenField from scapy.packet import Padding -from scapy.layers.tls.cert import Cert +from scapy.layers.tls.cert import Cert, CertList from scapy.layers.tls.basefields import _tls_version, _TLSVersionField from scapy.layers.tls.handshake import _CipherSuitesField from scapy.layers.tls.keyexchange import _TLSSignatureField, _TLSSignature @@ -196,7 +196,7 @@ def tls_session_update(self, msg_str): s.sslv2_connection_id = self.connection_id s.tls_version = self.version if self.cert is not None: - s.server_certs = [self.cert] + s.server_certs = CertList([self.cert]) ############################################################################### diff --git a/scapy/layers/tls/session.py b/scapy/layers/tls/session.py index 26813e49298..16baea88909 100644 --- a/scapy/layers/tls/session.py +++ b/scapy/layers/tls/session.py @@ -22,6 +22,7 @@ from scapy.sessions import TCPSession from scapy.utils import repr_hex, strxor from scapy.layers.inet import TCP +from scapy.layers.tls.cert import CertList from scapy.layers.tls.crypto.compression import Comp_NULL from scapy.layers.tls.crypto.hkdf import TLS13_HKDF from scapy.layers.tls.crypto.prf import PRF @@ -417,7 +418,7 @@ def __init__(self, # Either we act as server and it has to be provided, or it is expected # to be sent by the server through a Certificate message. # The server certificate should be self.server_certs[0]. - self.server_certs = [] + self.server_certs = CertList([]) # The server private key, as a PrivKey instance, when acting as server. # XXX It would be nice to be able to provide both an RSA and an ECDSA diff --git a/scapy/layers/x509.py b/scapy/layers/x509.py index 4070605894f..b283d8274c7 100644 --- a/scapy/layers/x509.py +++ b/scapy/layers/x509.py @@ -942,10 +942,14 @@ class X509_AlgorithmIdentifier(ASN1_Packet): # RFC8410: # "For all of the OIDs, the parameters MUST be absent." + + # RFC9881: + # "The parameters of these signature algorithms MUST be absent" ASN1F_omit("parameters", None), lambda pkt: ( pkt.algorithm.val[:16] == "1.2.840.10045.4." or - pkt.algorithm.val in ["1.3.101.112", "1.3.101.113"] + pkt.algorithm.val in ["1.3.101.112", "1.3.101.113"] or + pkt.algorithm.val[:21] == "2.16.840.1.101.3.4.3." ) ), # RFC5480 diff --git a/test/configs/windows.utsc b/test/configs/windows.utsc index a38f065e8ca..22973c4c77e 100644 --- a/test/configs/windows.utsc +++ b/test/configs/windows.utsc @@ -30,7 +30,6 @@ "broken_windows", "ipv6", "linux", - "native_tls13", "mock_read_routes_bsd", "open_ssl_client", "osx", diff --git a/test/scapy/layers/kerberos.uts b/test/scapy/layers/kerberos.uts index da71abb8fe3..4e776d697ad 100644 --- a/test/scapy/layers/kerberos.uts +++ b/test/scapy/layers/kerberos.uts @@ -277,8 +277,11 @@ ca = Cert(bytes.fromhex('3082036930820251a00302010202106b671318bb858b8e437e4229b # Build CMS engine to verify the authpack cms = CMS_Engine(CertList([ca])) -# Verify signature -authpack = cms.verify(signedauthpack, ASN1_OID('id-pkinit-authData')) +# Verify signature. +# allow_expired: this is a real captured MIT Kerberos exchange, and its client +# certificate was valid 2025-09-20 to 2026-09-20. The signature it carries is +# genuine; only the clock has moved on. +authpack = cms.verify(signedauthpack, ASN1_OID('id-pkinit-authData'), allow_expired=True) assert isinstance(authpack, KRB_AuthPack) = PKINIT - Resign AuthPack and re-verify signature @@ -295,7 +298,7 @@ signed = cms.sign( key, ) -authpack = cms.verify(signed, ASN1_OID('id-pkinit-authData')) +authpack = cms.verify(signed, ASN1_OID('id-pkinit-authData'), allow_expired=True) assert isinstance(authpack, KRB_AuthPack) = PKINIT - Parse AS-REP with CMS structures (MIT Kerberos) diff --git a/test/scapy/layers/tls/cert.uts b/test/scapy/layers/tls/cert.uts index b8dc78bf394..c5f368a15ea 100644 --- a/test/scapy/layers/tls/cert.uts +++ b/test/scapy/layers/tls/cert.uts @@ -652,21 +652,42 @@ assert repr_str == '/OU=Domain Control Validated/CN=*.tools.ietf.org [Not Self S = CertTree class : verify -CertTree([c1, c2]).verify(c0) +# c0 expired in 2016, so the validity check needs a date it was still current on. +in2016 = (2016, 1, 1, 0, 0, 0, 4, 1, 0) + +CertTree([c1, c2]).verify(c0, now=in2016) CertTree([c2]).verify(c1) try: - CertTree([c1]).verify(c0) + CertTree([c1]).verify(c0, now=in2016) assert False except ValueError: pass try: - CertTree([c2]).verify(c0) + CertTree([c2]).verify(c0, now=in2016) + assert False +except ValueError: + pass + += CertTree class : verify rejects an expired certificate + +# The chain is intact; only the date is wrong. +try: + CertTree([c1, c2]).verify(c0) assert False except ValueError: pass += CertTree class : verify checks the hostname when given one + +CertTree([c1, c2]).verify(c0, hostname="www.tools.ietf.org", now=in2016) + +try: + CertTree([c1, c2]).verify(c0, hostname="www.example.com", now=in2016) + assert False +except ValueError: + pass = Test GeneralizedTime diff --git a/test/scapy/layers/tls/tlsclientserver.uts b/test/scapy/layers/tls/tlsclientserver.uts index 64736edcfe5..58ffbe55a14 100644 --- a/test/scapy/layers/tls/tlsclientserver.uts +++ b/test/scapy/layers/tls/tlsclientserver.uts @@ -97,6 +97,7 @@ def run_tls_test_server(expected_data, q, curve=None, cookie=False, client_auth= res = check_output_for_data(out, err, expected_data) # Return data q.put(res) + print("run_tls_test_server: exit.") def wait_tls_test_server_online(): @@ -257,7 +258,8 @@ send_data = cipher_suite_code = version = None def run_tls_test_client(send_data=None, cipher_suite_code=None, version=None, client_auth=False, key_update=False, stop_server=True, - session_ticket_file_out=None, session_ticket_file_in=None): + session_ticket_file_out=None, session_ticket_file_in=None, + verify_server=False): print("Loading client...") mycert = scapy_path("/test/scapy/layers/tls/pki/cli_cert.pem") if client_auth else None mykey = scapy_path("/test/scapy/layers/tls/pki/cli_key.pem") if client_auth else None @@ -272,22 +274,26 @@ def run_tls_test_client(send_data=None, cipher_suite_code=None, version=None, if version == "0002": t = TLSClientAutomaton(data=commands, version="sslv2", debug=4, mycert=mycert, mykey=mykey, session_ticket_file_in=session_ticket_file_in, - session_ticket_file_out=session_ticket_file_out) + session_ticket_file_out=session_ticket_file_out, + verify_server=verify_server) elif version == "0304": ch = TLS13ClientHello(ciphers=int(cipher_suite_code, 16)) t = TLSClientAutomaton(client_hello=ch, data=commands, version="tls13", debug=4, mycert=mycert, mykey=mykey, session_ticket_file_in=session_ticket_file_in, - session_ticket_file_out=session_ticket_file_out) + session_ticket_file_out=session_ticket_file_out, + verify_server=verify_server) else: ch = TLSClientHello(version=int(version, 16), ciphers=int(cipher_suite_code, 16)) t = TLSClientAutomaton(client_hello=ch, data=commands, debug=4, mycert=mycert, mykey=mykey, session_ticket_file_in=session_ticket_file_in, - session_ticket_file_out=session_ticket_file_out) + session_ticket_file_out=session_ticket_file_out, + verify_server=verify_server) print("Running client...") t.run() + return t.final_reason def test_tls_client(suite, version, curve=None, cookie=False, client_auth=False, - key_update=False, sess_in_out=False, sigalgo="rsa"): + key_update=False, sess_in_out=False, sigalgo="rsa", verify_server=False): msg = ("TestC_%s_data" % suite).encode() # Run server q_ = Queue() @@ -305,24 +311,43 @@ def test_tls_client(suite, version, curve=None, cookie=False, client_auth=False, try: wait_tls_test_server_online() except Exception as ex: - atmtsrv.stop() + atmtsrv.forcestop() raise ex print("Thread synchronised") # Run client - if sess_in_out: - file_sess = scapy_path("/test/session") - run_tls_test_client(msg, suite, version, client_auth, key_update, session_ticket_file_out=file_sess, - stop_server=False) - run_tls_test_client(msg, suite, version, client_auth, key_update, session_ticket_file_in=file_sess, - stop_server=True) + try: + if sess_in_out: + file_sess = scapy_path("/test/session") + run_tls_test_client(msg, suite, version, client_auth, key_update, session_ticket_file_out=file_sess, + stop_server=False, verify_server=verify_server) + + final_reason = run_tls_test_client(msg, suite, version, client_auth, key_update, session_ticket_file_in=file_sess, + stop_server=True, verify_server=verify_server) + else: + final_reason = run_tls_test_client(msg, suite, version, client_auth, key_update, verify_server=verify_server) + except Exception as ex: + final_reason = Bunch(atmt_origfunc=ex) + # Check the final reason + if final_reason.atmt_origfunc not in [TLSClientAutomaton.CLOSE_NOTIFY.atmt_origfunc, TLSClientAutomaton.SSLv2_CLOSE_NOTIFY.atmt_origfunc]: + print("Client crashed:", final_reason.atmt_origfunc, "Stopping server.") + atmtsrv.stop(wait=False) + th_.join(timeout=5) else: - run_tls_test_client(msg, suite, version, client_auth, key_update) - # Wait for server - print("Client running, waiting...") - ret = q_.get(timeout=5) - if not ret: - raise RuntimeError("Test timed out") - atmtsrv.stop() + print("Client finished successfully") + final_reason = None + # Wait for server + print("Server still running, waiting...") + ret = q_.get(timeout=1) + if not ret: + raise RuntimeError("Test timed out") + atmtsrv.stop() + print("Server stopped. Destroying server...") + atmtsrv.destroy(force=True) + print("Server destroyed.") + # re-raise if client failed. + if final_reason: + raise ValueError(final_reason.atmt_origfunc) + # final assertion that server didn't fail. print(ret) assert ret[0] @@ -402,63 +427,14 @@ test_tls_client("1305", "0304", key_update=True) test_tls_client("1305", "0304", client_auth=True, sess_in_out=True) -= Reject a CertificateVerify made with a key unrelated to the client certificate += Testing TLS server and client with TLS 1.3 and TLS_AES_128_CCM_8_SHA256 and server verification ~ crypto_advanced -def run_tls13_client_auth_exchange(client_key): - marker = b"certificate proof accepted" - server_cert = scapy_path("/test/scapy/layers/tls/pki/srv_cert.pem") - server_key = scapy_path("/test/scapy/layers/tls/pki/srv_key.pem") - client_cert = scapy_path("/test/scapy/layers/tls/pki/cli_cert.pem") - probe = socket.socket() - probe.bind(("127.0.0.1", 0)) - port = probe.getsockname()[1] - probe.close() - server = TLSServerAutomaton( - server="127.0.0.1", - sport=port, - mycert=server_cert, - mykey=server_key, - client_auth=True, - debug=0, - ) - server_thread = threading.Thread(target=server.run, daemon=True) - with captured_output() as (out, _): - server_thread.start() - time.sleep(0.1) - client = TLSClientAutomaton( - server="127.0.0.1", - dport=port, - version="tls13", - mycert=client_cert, - mykey=client_key, - data=[marker, b"stop_server", b"quit"], - debug=0, - ) - client_thread = threading.Thread(target=client.run, daemon=True) - client_thread.start() - client_thread.join(timeout=2) - time.sleep(0.1) - if client_thread.is_alive(): - client.stop(wait=False) - if server_thread.is_alive(): - stopper = TLSClientAutomaton( - server="127.0.0.1", - dport=port, - version="tls13", - mycert=client_cert, - mykey=scapy_path("/test/scapy/layers/tls/pki/cli_key.pem"), - data=[b"stop_server", b"quit"], - debug=0, - ) - stopper.run() - server_thread.join(timeout=1) - return "> Received: %r" % marker in out.s - -server_key = scapy_path("/test/scapy/layers/tls/pki/srv_key.pem") -client_key = scapy_path("/test/scapy/layers/tls/pki/cli_key.pem") -assert not run_tls13_client_auth_exchange(server_key) -assert run_tls13_client_auth_exchange(client_key) +try: + test_tls_client("1305", "0304", verify_server=True) + assert False, "Client should have rejected the server" +except ValueError as ex: + assert "INVALID_SERVER_CERTIFICATE" in str(ex) = Clear session file @@ -473,118 +449,6 @@ except: + TLS client automaton tests against builtin ssl using Post Handshake Authentication ~ client post_handshake_auth -= TLS 1.3 client rejects an invalid server CertificateVerify -~ crypto_advanced - -import contextlib, io, ipaddress, socket -from datetime import datetime, timedelta, timezone -from cryptography import x509 as crypto_x509 -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID -from scapy.layers.tls.cert import PrivKey - -server_key = scapy_path("/test/scapy/layers/tls/pki/srv_key.pem") -with open(server_key, "rb") as cert_key_file: - cert_key = serialization.load_pem_private_key(cert_key_file.read(), None) - -with open(scapy_path("/test/scapy/layers/tls/pki/ca_key.pem"), "rb") as ca_key_file: - ca_key = serialization.load_pem_private_key(ca_key_file.read(), None) - -ca_name = crypto_x509.Name([ - crypto_x509.NameAttribute(NameOID.COMMON_NAME, "CertificateVerify test CA") -]) -cert_name = crypto_x509.Name([ - crypto_x509.NameAttribute(NameOID.COMMON_NAME, "127.0.0.1") -]) -now = datetime.now(timezone.utc) -ca_cert = (crypto_x509.CertificateBuilder() - .subject_name(ca_name) - .issuer_name(ca_name) - .public_key(ca_key.public_key()) - .serial_number(crypto_x509.random_serial_number()) - .not_valid_before(now - timedelta(days=1)) - .not_valid_after(now + timedelta(days=1)) - .add_extension(crypto_x509.BasicConstraints(ca=True, path_length=None), True) - .add_extension(crypto_x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), False) - .add_extension(crypto_x509.KeyUsage(False, False, False, False, False, True, True, None, None), True) - .sign(ca_key, hashes.SHA256())) -ca_certfile = get_temp_file() -with open(ca_certfile, "wb") as ca_cert_file: - ca_cert_file.write(ca_cert.public_bytes(serialization.Encoding.PEM)) - -trusted_server_cert = (crypto_x509.CertificateBuilder() - .subject_name(cert_name) - .issuer_name(ca_name) - .public_key(cert_key.public_key()) - .serial_number(crypto_x509.random_serial_number()) - .not_valid_before(now - timedelta(days=1)) - .not_valid_after(now + timedelta(days=1)) - .add_extension(crypto_x509.BasicConstraints(ca=False, path_length=None), True) - .add_extension(crypto_x509.SubjectKeyIdentifier.from_public_key(cert_key.public_key()), False) - .add_extension(crypto_x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), False) - .add_extension(crypto_x509.KeyUsage(True, False, True, False, False, False, False, None, None), True) - .add_extension(crypto_x509.SubjectAlternativeName([ - crypto_x509.IPAddress(ipaddress.ip_address("127.0.0.1")) - ]), False) - .add_extension(crypto_x509.ExtendedKeyUsage([ - ExtendedKeyUsageOID.SERVER_AUTH - ]), False) - .sign(ca_key, hashes.SHA256())) -trusted_server_certfile = get_temp_file() -with open(trusted_server_certfile, "wb") as cert_file: - cert_file.write(trusted_server_cert.public_bytes(serialization.Encoding.PEM)) - -def run_server_certificate_verify_case(signing_key): - marker = b"server proof accepted" - probe = socket.socket() - probe.bind(("127.0.0.1", 0)) - port = probe.getsockname()[1] - probe.close() - server = TLSServerAutomaton( - server="127.0.0.1", - sport=port, - mycert=trusted_server_certfile, - mykey=signing_key, - is_echo_server=True, - max_client_idle_time=0.3, - debug=0, - ) - server_thread = threading.Thread(target=server.run, daemon=True) - output = io.StringIO() - with contextlib.redirect_stdout(output): - server_thread.start() - for _ in range(100): - if server.serversocket is not None: - break - time.sleep(0.01) - client = TLSClientAutomaton( - server="127.0.0.1", - dport=port, - version="tls13", - cafile=ca_certfile, - data=[marker, b"stop_server", b"quit"], - debug=0, - ) - client.run() - server_thread.join(timeout=0.5) - if server_thread.is_alive(): - server.mykey = PrivKey(server_key) - stopper = TLSClientAutomaton( - server="127.0.0.1", - dport=port, - version="tls13", - cafile=ca_certfile, - data=[b"stop_server", b"quit"], - debug=0, - ) - stopper.run() - server_thread.join(timeout=1) - return ("> Received: %r" % marker) in output.getvalue() - -mismatched_key = scapy_path("/test/scapy/layers/tls/pki/cli_key.pem") -assert not run_server_certificate_verify_case(mismatched_key) -assert run_server_certificate_verify_case(server_key) - = Load native server util functions # Imports @@ -615,6 +479,7 @@ with open(certfile, "wb") as fd: with open(keyfile, "wb") as fd: fd.write(rsa_key) + # Define server REQS = [ @@ -714,15 +579,11 @@ def test_tls_client_native(post_handshake_auth=False, assert not server.is_alive() -# XXX: Ugh, Appveyor uses an ancient Windows 10 build that doesn't support TLS 1.3 natively. - = Testing TLS client against ssl.SSLContext server with TLS 1.3 and a post-handshake authentication -~ native_tls13 test_tls_client_native(post_handshake_auth=True) = Testing TLS client against ssl.SSLContext server with TLS 1.3 and a Hello-Retry request -~ native_tls13 test_tls_client_native(with_hello_retry=True) @@ -746,4 +607,4 @@ def _test_connection(): assert HTTPResponse in pkt assert b"" in pkt[HTTPResponse].load -retry_test(_test_connection) \ No newline at end of file +retry_test(_test_connection) diff --git a/test/scapy/layers/x509.uts b/test/scapy/layers/x509.uts index cc561089b34..d4f01469c6e 100644 --- a/test/scapy/layers/x509.uts +++ b/test/scapy/layers/x509.uts @@ -185,6 +185,9 @@ c = X509_Cert(cert_with_bmp_string) bmp_field_value = str(c.tbsCertificate.issuer[7].rdn[0].value.val, "utf-16be") assert bmp_field_value == '1246183514EP@websense.com' += Cert class: ml-dsa signature signature +ai = X509_AlgorithmIdentifier(b'0\x0b\x06\t`\x86H\x01e\x03\x04\x03\x13') +assert ai.algorithm.oidname == 'ml-dsa-87' ############ CRL class ###############################################