Skip to content

tls: add optional server certificate verification #5144

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
KernelClint wants to merge 1 commit into
base: master
Choose a base branch
from
Open

tls: add optional server certificate verification #5144

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 82 additions & 3 deletions scapy/layers/tls/automaton_cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@

import socket
import binascii
import ssl
import struct
import time

Expand All @@ -47,6 +48,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 Cert, CertTree
from scapy.layers.tls.session import tlsSession
from scapy.layers.tls.extensions import (
ServerName,
Expand Down Expand Up @@ -86,6 +88,45 @@
)


def _load_trust_anchors(cafile):
"""
The certificates to trust: those in `cafile`, or the system trust store.

``ssl`` is used only to find and read the store; nothing about the
connection goes through it.

:param cafile: (optional) a PEM bundle to trust instead of the system store
:return: a list of Cert
"""
if not conf.crypto_valid:
return []
context = ssl.create_default_context(cafile=cafile)
return [Cert(der) for der in context.get_ca_certs(binary_form=True)]
Comment on lines +91 to +104

@gpotter2 gpotter2 Sep 8, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PR looks pretty good. If possible can you move this to CertTree ? behind like a load_system_store=False argument that only kicks in when the passed trusted certs list is empty. Thanks !



def _verify_server_certificate(certificates, trusted_certs, hostname):
"""
Whether the server's certificate chains to a trusted CA and names the host.

:param certificates: the chain the server sent, leaf first
:param trusted_certs: the CAs to trust
:param hostname: the name the client asked for
:return: True if the server is authenticated
"""
# An empty anchor list must fail closed. CertTree treats no roots as
# "trust any self-signed certificate in the list", and that list is the
# one the peer just sent.
if not certificates or not trusted_certs:
return False
try:
CertTree(list(certificates), trusted_certs).verify(
certificates[0], hostname=hostname
)
except Exception:
return False
Comment on lines +91 to +126

@gpotter2 gpotter2 Sep 3, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you use 'CertTree' from scapy.layers.tls instead? It should have a verify function, although a bit rudimentary.

@KernelClint KernelClint Sep 6, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can do that, and it drops ipaddress. ssl has to stay — it is the only way to find the system trust store. Say if you would rather require a cafile instead.

One thing first, because it changes what the option promises.

CertTree.verify() checks every signature properly; I tampered with one and it failed. But that is all it checks. It never looks at the date, and it never sees the hostname — verify(self, cert) has nowhere to put one. So a certificate issued to someone else passes, and so does one that expired ten days ago. The test in this PR asserts the first of those is rejected.

There is also a trap. Leave out rootCAs and CertTree trusts any self-signed certificate in the list you hand it — and here that list came from the peer. That's basically a non-check...

The other three clients with no_check_certificate — HTTP_Client, LDAP_Client, and Kerberos through HTTP_Client — all fall back to ssl.create_default_context(), which does check the hostname. This one cannot: there is no ssl socket to hand the job to.

So what would you prefer here? I'm thinking one of the following:

  1. CertTree, plus a name check against subjectAltName and an expiry check. No new imports. Wildcards and IP addresses work; punycode does not.
  2. CertTree alone, and I rename the option so it does not oversell itself.
  3. CertTree alone, name unchanged.

Note that 2 and 3 accept expired certificates, which the code they replace rejects... but maybe that's just the nature of Scapy? :-)

@gpotter2 gpotter2 Sep 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you (or agents) have the time, 1 with an updated verify() that checks the SAN and expiry would be great. Thanks

return True


class TLSClientAutomaton(_TLSAutomaton):
"""
A simple TLS test client automaton. Try to overload some states or
Expand All @@ -97,6 +138,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.
:param verify: whether to authenticate the server certificate. Defaults to True.
: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.
Expand All @@ -116,6 +160,7 @@ class TLSClientAutomaton(_TLSAutomaton):
"""

def parse_args(self, server="127.0.0.1", dport=4433, server_name=None,
cafile=None, verify=True,
mycert=None, mykey=None,
client_hello=None, version=None,
resumption_master_secret=None,
Expand All @@ -137,6 +182,11 @@ 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
self.server_trust_anchors = (
_load_trust_anchors(cafile) if verify else []
)
self.local_ip = None
self.local_port = None
self.socket = None
Expand Down Expand Up @@ -402,7 +452,22 @@ def should_handle_ServerCertificate(self):

@ATMT.state()
def HANDLED_SERVERCERTIFICATE(self):
pass
if self.verify_server:
self.cur_session.server_cert_valid = _verify_server_certificate(
self.cur_session.server_certs,
self.server_trust_anchors,
self.expected_server_name,
)
if not self.cur_session.server_cert_valid:
raise self.INVALID_SERVER_CERTIFICATE()

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

@ATMT.condition(HANDLED_SERVERHELLO, prio=2)
def missing_ServerCertificate(self):
Expand Down Expand Up @@ -842,7 +907,14 @@ def sslv2_should_handle_ServerHello(self):

@ATMT.state()
def SSLv2_HANDLED_SERVERHELLO(self):
pass
if self.verify_server:
self.cur_session.server_cert_valid = _verify_server_certificate(
self.cur_session.server_certs,
self.server_trust_anchors,
self.expected_server_name,
)
if not self.cur_session.server_cert_valid:
raise self.SSLv2_CLOSE_NOTIFY()

@ATMT.condition(SSLv2_RECEIVED_SERVERHELLO, prio=2)
def sslv2_missing_ServerHello(self):
Expand Down Expand Up @@ -1341,7 +1413,14 @@ def tls13_should_handle_Certificate(self):

@ATMT.state()
def TLS13_HANDLED_CERTIFICATE(self):
pass
if self.verify_server:
self.cur_session.server_cert_valid = _verify_server_certificate(
self.cur_session.server_certs,
self.server_trust_anchors,
self.expected_server_name,
)
if not self.cur_session.server_cert_valid:
raise self.INVALID_SERVER_CERTIFICATE()

@ATMT.condition(TLS13_HANDLED_CERTIFICATE, prio=1)
def tls13_should_handle_CertificateVerify(self):
Expand Down
147 changes: 143 additions & 4 deletions scapy/layers/tls/cert.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,15 +92,18 @@
"""

import base64
import calendar
import enum
import os
import socket
import time
import warnings

from scapy.config import conf, crypto_validator
from scapy.compat import Self
from scapy.compat import Self, plain_str
from scapy.error import warning
from scapy.utils import binrepr
from scapy.pton_ntop import inet_ntop
from scapy.asn1.asn1 import (
ASN1_BIT_STRING,
ASN1_NULL,
Expand Down Expand Up @@ -133,6 +136,8 @@
X509_AttributeValue,
X509_Cert,
X509_CRL,
X509_DNSName,
X509_IPAddress,
X509_SubjectPublicKeyInfo,
)
from scapy.layers.tls.crypto.hash import _get_hash
Expand Down Expand Up @@ -940,6 +945,55 @@ def _get_csr_sig_hashname(csr):
return hash_by_oid[sigAlg.algorithm.val]


def _parse_subject_alt_name(extnValue):
"""
Collect the DNS names and IP addresses from a subjectAltName extension.

:param extnValue: the X509_ExtSubjectAltName packet
:return: a list of ("DNS", name) and ("IP", address) pairs
"""
names = []
for generalName in extnValue.subjectAltName or []:
name = generalName.generalName
if isinstance(name, X509_DNSName):
names.append(("DNS", plain_str(name.dNSName.val)))
elif isinstance(name, X509_IPAddress):
raw = name.iPAddress.val
if len(raw) == 4:
names.append(("IP", inet_ntop(socket.AF_INET, raw)))
elif len(raw) == 16:
names.append(("IP", inet_ntop(socket.AF_INET6, raw)))
return names


def _match_dns_name(pattern, hostname):
"""
Whether a certificate DNS name matches a hostname, RFC 6125 sect 6.4.3.

A wildcard is only honoured as the whole leftmost label, and only when the
name has at least two more labels after it, so ``*.example.com`` matches
``a.example.com`` but not ``example.com`` or ``a.b.example.com``, and ``*.com``
matches nothing.

: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 not pattern or not hostname:
return False
if not pattern.startswith("*."):
return pattern == hostname
suffix = pattern[1:]
if suffix.count(".") < 2:
# A wildcard directly under a public suffix would match too much.
return False
if not hostname.endswith(suffix):
return False
return "." not in hostname[:-len(suffix)]


class Cert(metaclass=_CertMaker):
"""
Wrapper for the X509_Cert from layers/x509.py.
Expand Down Expand Up @@ -983,6 +1037,11 @@ def import_from_asn1pkt(self, cert):

self.pubkey = PubKey(bytes(tbsCert.subjectPublicKeyInfo))

# The names this certificate is issued to, as ("DNS", name) or
# ("IP", address) pairs. Other GeneralName kinds are not used to
# identify a server, so they are not recorded here.
self.subjectAltName = []

if tbsCert.extensions:
for extn in tbsCert.extensions:
if extn.extnID.oidname == "basicConstraints":
Expand All @@ -995,6 +1054,10 @@ 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 = _parse_subject_alt_name(
extn.extnValue
)

self.signatureValue = bytes(cert.signatureValue)
self.signatureLen = len(self.signatureValue)
Expand Down Expand Up @@ -1094,6 +1157,53 @@ 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:
for kind, name in self.subjectAltName:
if kind == "IP":
if name == hostname:
return True
elif _match_dns_name(name, hostname):
return True
return False
for attr in self.subject_str.split("/"):
if attr.startswith("CN=") and _match_dns_name(attr[3:], hostname):
return True
return False

def isRevoked(self, crl_list):
"""
Given a list of trusted CRL (their signature has already been
Expand Down Expand Up @@ -1659,13 +1769,42 @@ def _rec_getchain(chain, curtree):
else:
return None

def verify(self, cert):
def verify(self, cert, hostname=None, now=None):
"""
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
"""
# 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 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):
"""
Expand Down
1 change: 1 addition & 0 deletions scapy/layers/tls/session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -418,6 +418,7 @@ def __init__(self,
# to be sent by the server through a Certificate message.
# The server certificate should be self.server_certs[0].
self.server_certs = []
self.server_cert_valid = None

# 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
Expand Down
Loading
Loading