Fix several problems in auth_certificate_callback().

Add utility to print out SSLUsage flags.
Set the auth_certificate_callback in the client to verify the server,
it was a big problem that had been failing to verify the server cert.
Add new utility to verify a server based on the auth_certificate_callback().


git-svn-id: svn://10.0.0.236/trunk@260027 18797224-902f-48f8-a5cc-f745e15eee43
This commit is contained in:
jdennis%redhat.com
2010-03-24 16:53:07 +00:00
parent 9fbac0c95e
commit 514d140fd0
3 changed files with 467 additions and 31 deletions

View File

@@ -66,6 +66,70 @@ nickname = hostname.split('.')[0]
port = 1234
# -----------------------------------------------------------------------------
# Utility Functions
# -----------------------------------------------------------------------------
def cert_usage_str(cert_usage):
usages = []
if cert_usage & nss.certificateUsageSSLClient:
cert_usage &= ~nss.certificateUsageSSLClient
usages.append('SSLClient')
if cert_usage & nss.certificateUsageSSLServer:
cert_usage &= ~nss.certificateUsageSSLServer
usages.append('SSLServer')
if cert_usage & nss.certificateUsageSSLServerWithStepUp:
cert_usage &= ~nss.certificateUsageSSLServerWithStepUp
usages.append('SSLServerWithStepUp')
if cert_usage & nss.certificateUsageSSLCA:
cert_usage &= ~nss.certificateUsageSSLCA
usages.append('SSLCA')
if cert_usage & nss.certificateUsageEmailSigner:
cert_usage &= ~nss.certificateUsageEmailSigner
usages.append('EmailSigner')
if cert_usage & nss.certificateUsageEmailRecipient:
cert_usage &= ~nss.certificateUsageEmailRecipient
usages.append('EmailRecipient')
if cert_usage & nss.certificateUsageObjectSigner:
cert_usage &= ~nss.certificateUsageObjectSigner
usages.append('ObjectSigner')
if cert_usage & nss.certificateUsageUserCertImport:
cert_usage &= ~nss.certificateUsageUserCertImport
usages.append('UserCertImport')
if cert_usage & nss.certificateUsageVerifyCA:
cert_usage &= ~nss.certificateUsageVerifyCA
usages.append('VerifyCA')
if cert_usage & nss.certificateUsageProtectedObjectSigner:
cert_usage &= ~nss.certificateUsageProtectedObjectSigner
usages.append('ProtectedObjectSigner')
if cert_usage & nss.certificateUsageStatusResponder:
cert_usage &= ~nss.certificateUsageStatusResponder
usages.append('StatusResponder')
if cert_usage & nss.certificateUsageAnyCA:
cert_usage &= ~nss.certificateUsageAnyCA
usages.append('AnyCA')
usages.sort()
usage_str = ','.join(usages)
if cert_usage:
usage_str += ' (plus unknown flags %#x)' % cert_usage
return usage_str
# -----------------------------------------------------------------------------
# Callback Functions
# -----------------------------------------------------------------------------
@@ -78,30 +142,47 @@ def handshake_callback(sock):
print "handshake complete, peer = %s" % (sock.get_peer_name())
def auth_certificate_callback(sock, check_sig, is_server, certdb):
validity = False
print "auth_certificate_callback: check_sig=%s is_server=%s" % (check_sig, is_server)
cert_is_valid = False
cert = sock.get_peer_certificate()
pin_args = sock.get_pkcs11_pin_arg()
if pin_args is None:
pin_args = ()
print "client cert:\n%s" % cert
print "cert:\n%s" % cert
# Define how the cert is being used based upon the is_server flag.
# This may seem backwards, but isn't.
# Define how the cert is being used based upon the is_server flag. This may
# seem backwards, but isn't. If we're a server we're trying to validate a
# client cert. If we're a client we're trying to validate a server cert.
if is_server:
cert_usage = nss.certificateUsageSSLClient
intended_usage = nss.certificateUsageSSLClient
else:
cert_usage = nss.certificateUsageSSLServer
intended_usage = nss.certificateUsageSSLServer
valid_usage = cert.verify_now(certdb, check_sig, cert_usage, *pin_args)
try:
# If the cert fails validation it will raise an exception, the errno attribute
# will be set to the error code matching the reason why the validation failed
# and the strerror attribute will contain a string describing the reason.
approved_usage = cert.verify_now(certdb, check_sig, intended_usage, *pin_args)
except Exception, e:
print e.strerror
cert_is_valid = False
print "Returning cert_is_valid = %s" % cert_is_valid
return cert_is_valid
if valid_usage & cert_usage:
validity = True
print "approved_usage = %s" % cert_usage_str(approved_usage)
# Is the intended usage a proper subset of the approved usage
if approved_usage & intended_usage:
cert_is_valid = True
else:
validity = False
cert_is_valid = False
# If this is a server, we're finished
if is_server or not validity:
return validity
if is_server or not cert_is_valid:
print "Returning cert_is_valid = %s" % cert_is_valid
return cert_is_valid
# Certificate is OK. Since this is the client side of an SSL
# connection, we need to verify that the name field in the cert
@@ -109,9 +190,18 @@ def auth_certificate_callback(sock, check_sig, is_server, certdb):
# man-in-the-middle attacks.
hostname = sock.get_hostname()
validity = cert.verify_hostname(hostname)
print "verifying socket hostname (%s) matches cert subject (%s)" % (hostname, cert.subject)
try:
# If the cert fails validation it will raise an exception
cert_is_valid = cert.verify_hostname(hostname)
except Exception, e:
print e.strerror
cert_is_valid = False
print "Returning cert_is_valid = %s" % cert_is_valid
return cert_is_valid
return validity
print "Returning cert_is_valid = %s" % cert_is_valid
return cert_is_valid
def client_auth_data_callback(ca_names, chosen_nickname, password, certdb):
cert = None
@@ -158,7 +248,12 @@ def Client():
sock.set_handshake_callback(handshake_callback)
# Provide a callback to supply our client certificate info
sock.set_client_auth_data_callback(client_auth_data_callback, nickname, password, nss.get_default_certdb())
sock.set_client_auth_data_callback(client_auth_data_callback, nickname,
password, nss.get_default_certdb())
# Provide a callback to verify the servers certificate
sock.set_auth_certificate_callback(auth_certificate_callback,
nss.get_default_certdb())
else:
sock = io.Socket()
@@ -166,12 +261,17 @@ def Client():
sock.connect(net_addr)
# Talk to the server
sock.send("Hello")
buf = sock.recv(1024)
if not buf:
print "client lost connection"
try:
sock.send("Hello")
buf = sock.recv(1024)
if not buf:
print "client lost connection"
return
print "client received: %s" % (buf)
except Exception, e:
print e.strerror
sock.shutdown()
return
print "client received: %s" % (buf)
# End of (simple) protocol session?
if buf == 'Goodbye':
@@ -233,17 +333,21 @@ def Server():
print "client connect from: %s" % (client_addr)
while True:
# Handle the client connection
buf = client_sock.recv(1024)
if not buf:
print "server lost lost connection to %s" % (client_addr)
try:
# Handle the client connection
buf = client_sock.recv(1024)
if not buf:
print "server lost lost connection to %s" % (client_addr)
break
print "server received: %s" % (buf)
client_sock.send("Goodbye")
client_sock.shutdown(io.PR_SHUTDOWN_RCV)
break
except Exception, e:
print e.strerror
break
print "server received: %s" % (buf)
client_sock.send("Goodbye")
client_sock.shutdown(io.PR_SHUTDOWN_RCV)
break
break
sock.shutdown()
@@ -338,7 +442,11 @@ if not (client or server):
sys.exit(1)
# Perform basic configuration and setup
nss.nss_init(certdir)
if certdir is None:
nss.nss_init_nodb()
else:
nss.nss_init(certdir)
ssl.set_domestic_policy()
nss.set_password_callback(password_callback)