| # Wrapper module for _ssl, providing some additional facilities |
| # implemented in Python. Written by Bill Janssen. |
| |
| """\ |
| This module provides some more Pythonic support for SSL. |
| |
| Object types: |
| |
| sslsocket -- subtype of socket.socket which does SSL over the socket |
| |
| Exceptions: |
| |
| sslerror -- exception raised for I/O errors |
| |
| Functions: |
| |
| cert_time_to_seconds -- convert time string used for certificate |
| notBefore and notAfter functions to integer |
| seconds past the Epoch (the time values |
| returned from time.time()) |
| |
| fetch_server_certificate (HOST, PORT) -- fetch the certificate provided |
| by the server running on HOST at port PORT. No |
| validation of the certificate is performed. |
| |
| Integer constants: |
| |
| SSL_ERROR_ZERO_RETURN |
| SSL_ERROR_WANT_READ |
| SSL_ERROR_WANT_WRITE |
| SSL_ERROR_WANT_X509_LOOKUP |
| SSL_ERROR_SYSCALL |
| SSL_ERROR_SSL |
| SSL_ERROR_WANT_CONNECT |
| |
| SSL_ERROR_EOF |
| SSL_ERROR_INVALID_ERROR_CODE |
| |
| The following group define certificate requirements that one side is |
| allowing/requiring from the other side: |
| |
| CERT_NONE - no certificates from the other side are required (or will |
| be looked at if provided) |
| CERT_OPTIONAL - certificates are not required, but if provided will be |
| validated, and if validation fails, the connection will |
| also fail |
| CERT_REQUIRED - certificates are required, and will be validated, and |
| if validation fails, the connection will also fail |
| |
| The following constants identify various SSL protocol variants: |
| |
| PROTOCOL_SSLv2 |
| PROTOCOL_SSLv3 |
| PROTOCOL_SSLv23 |
| PROTOCOL_TLSv1 |
| """ |
| |
| import os, sys |
| |
| import _ssl # if we can't import it, let the error propagate |
| from socket import socket |
| from _ssl import sslerror |
| from _ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED |
| from _ssl import PROTOCOL_SSLv2, PROTOCOL_SSLv3, PROTOCOL_SSLv23, PROTOCOL_TLSv1 |
| |
| # Root certs: |
| # |
| # The "ca_certs" argument to sslsocket() expects a file containing one or more |
| # certificates that are roots of various certificate signing chains. This file |
| # contains the certificates in PEM format (RFC ) where each certificate is |
| # encoded in base64 encoding and surrounded with a header and footer: |
| # -----BEGIN CERTIFICATE----- |
| # ... (CA certificate in base64 encoding) ... |
| # -----END CERTIFICATE----- |
| # The various certificates in the file are just concatenated together: |
| # -----BEGIN CERTIFICATE----- |
| # ... (CA certificate in base64 encoding) ... |
| # -----END CERTIFICATE----- |
| # -----BEGIN CERTIFICATE----- |
| # ... (a second CA certificate in base64 encoding) ... |
| # -----END CERTIFICATE----- |
| # |
| # Some "standard" root certificates are available at |
| # |
| # http://www.thawte.com/roots/ (for Thawte roots) |
| # http://www.verisign.com/support/roots.html (for Verisign) |
| |
| class sslsocket (socket): |
| |
| def __init__(self, sock, keyfile=None, certfile=None, |
| server_side=False, cert_reqs=CERT_NONE, |
| ssl_version=PROTOCOL_SSLv23, ca_certs=None): |
| socket.__init__(self, _sock=sock._sock) |
| if certfile and not keyfile: |
| keyfile = certfile |
| if server_side: |
| self._sslobj = _ssl.sslwrap(self._sock, 1, keyfile, certfile, |
| cert_reqs, ssl_version, ca_certs) |
| else: |
| # see if it's connected |
| try: |
| socket.getpeername(self) |
| except: |
| # no, no connection yet |
| self._sslobj = None |
| else: |
| # yes, create the SSL object |
| self._sslobj = _ssl.sslwrap(self._sock, 0, keyfile, certfile, |
| cert_reqs, ssl_version, ca_certs) |
| self.keyfile = keyfile |
| self.certfile = certfile |
| self.cert_reqs = cert_reqs |
| self.ssl_version = ssl_version |
| self.ca_certs = ca_certs |
| |
| def read(self, len=1024): |
| return self._sslobj.read(len) |
| |
| def write(self, data): |
| return self._sslobj.write(data) |
| |
| def getpeercert(self): |
| return self._sslobj.peer_certificate() |
| |
| def send (self, data, flags=0): |
| if flags != 0: |
| raise ValueError( |
| "non-zero flags not allowed in calls to send() on %s" % |
| self.__class__) |
| return self._sslobj.write(data) |
| |
| def send_to (self, data, addr, flags=0): |
| raise ValueError("send_to not allowed on instances of %s" % |
| self.__class__) |
| |
| def sendall (self, data, flags=0): |
| if flags != 0: |
| raise ValueError( |
| "non-zero flags not allowed in calls to sendall() on %s" % |
| self.__class__) |
| return self._sslobj.write(data) |
| |
| def recv (self, buflen=1024, flags=0): |
| if flags != 0: |
| raise ValueError( |
| "non-zero flags not allowed in calls to sendall() on %s" % |
| self.__class__) |
| return self._sslobj.read(data, buflen) |
| |
| def recv_from (self, addr, buflen=1024, flags=0): |
| raise ValueError("recv_from not allowed on instances of %s" % |
| self.__class__) |
| |
| def shutdown(self): |
| if self._sslobj: |
| self._sslobj.shutdown() |
| self._sslobj = None |
| else: |
| socket.shutdown(self) |
| |
| def close(self): |
| if self._sslobj: |
| self.shutdown() |
| else: |
| socket.close(self) |
| |
| def connect(self, addr): |
| # Here we assume that the socket is client-side, and not |
| # connected at the time of the call. We connect it, then wrap it. |
| if self._sslobj or (self.getsockname()[1] != 0): |
| raise ValueError("attempt to connect already-connected sslsocket!") |
| socket.connect(self, addr) |
| self._sslobj = _ssl.sslwrap(self._sock, 0, self.keyfile, self.certfile, |
| self.cert_reqs, self.ssl_version, |
| self.ca_certs) |
| |
| def accept(self): |
| raise ValueError("accept() not supported on an sslsocket") |
| |
| |
| # some utility functions |
| |
| def cert_time_to_seconds(cert_time): |
| import time |
| return time.mktime(time.strptime(cert_time, "%b %d %H:%M:%S %Y GMT")) |
| |
| # a replacement for the old socket.ssl function |
| |
| def sslwrap_simple (sock, keyfile=None, certfile=None): |
| |
| return _ssl.sslwrap(sock._sock, 0, keyfile, certfile, CERT_NONE, |
| PROTOCOL_SSLv23, None) |
| |
| # fetch the certificate that the server is providing in PEM form |
| |
| def fetch_server_certificate (host, port): |
| |
| import re, tempfile, os |
| |
| def subproc(cmd): |
| from subprocess import Popen, PIPE, STDOUT |
| proc = Popen(cmd, stdout=PIPE, stderr=STDOUT, shell=True) |
| status = proc.wait() |
| output = proc.stdout.read() |
| return status, output |
| |
| def strip_to_x509_cert(certfile_contents, outfile=None): |
| m = re.search(r"^([-]+BEGIN CERTIFICATE[-]+[\r]*\n" |
| r".*[\r]*^[-]+END CERTIFICATE[-]+)$", |
| certfile_contents, re.MULTILINE | re.DOTALL) |
| if not m: |
| return None |
| else: |
| tn = tempfile.mktemp() |
| fp = open(tn, "w") |
| fp.write(m.group(1) + "\n") |
| fp.close() |
| try: |
| tn2 = (outfile or tempfile.mktemp()) |
| status, output = subproc(r'openssl x509 -in "%s" -out "%s"' % |
| (tn, tn2)) |
| if status != 0: |
| raise OperationError(status, tsig, output) |
| fp = open(tn2, 'rb') |
| data = fp.read() |
| fp.close() |
| os.unlink(tn2) |
| return data |
| finally: |
| os.unlink(tn) |
| |
| if sys.platform.startswith("win"): |
| tfile = tempfile.mktemp() |
| fp = open(tfile, "w") |
| fp.write("quit\n") |
| fp.close() |
| try: |
| status, output = subproc( |
| 'openssl s_client -connect "%s:%s" -showcerts < "%s"' % |
| (host, port, tfile)) |
| finally: |
| os.unlink(tfile) |
| else: |
| status, output = subproc( |
| 'openssl s_client -connect "%s:%s" -showcerts < /dev/null' % |
| (host, port)) |
| if status != 0: |
| raise OSError(status) |
| certtext = strip_to_x509_cert(output) |
| if not certtext: |
| raise ValueError("Invalid response received from server at %s:%s" % |
| (host, port)) |
| return certtext |