Tests requests with custom certificates

makecerts.sh

#!/bin/sh

set -e

ca_conf='
[ ca ]
default_ca = ca_default

[ ca_default ]
dir = ./
certs = $dir
new_certs_dir = $dir
database = ca-db-index
serial = ca-db-serial
RANDFILE = ca-db-rand
certificate = ca-cert.pem
private_key = ca-key.pem
default_days = 3650
default_crl_days = 3650
default_md = sha256
preserve = no
policy = generic_policy

[ generic_policy ]
countryName = optional
stateOrProvinceName = optional
localityName = optional
organizationName = optional
organizationalUnitName = optional
commonName = supplied
emailAddress = optional'

make_ca(){
    echo "$ca_conf" > ca.conf
    touch ca-db-index
    echo 01 > ca-db-serial

    openssl req -nodes -x509 -days 3650 \
        -newkey rsa:1024 -keyout ca-key.pem \
        -out ca-cert.pem \
        -subj "/C=US/ST=New York/L=New York/O=remram.fr/CN=ca.remram.fr"
}

make_cert(){
    openssl req -nodes -new \
        -newkey rsa:1024 -keyout server-key.pem \
        -out server.csr \
        -subj "/C=US/ST=New York/L=New York/O=remram.fr/CN=localhost"
    openssl ca -batch -config ca.conf \
        -days 3650 \
        -in server.csr -out server-cert.pem
}

rm -Rf certs; mkdir certs && cd certs

mkdir ca1 && cd ca1
make_ca
cp ca-cert.pem ../ca1.pem
make_cert
cp server-cert.pem ../cert1.pem
cp server-key.pem ../key1.pem
cd ..

mkdir ca2 && cd ca2
make_ca 2
cp ca-cert.pem ../ca2.pem
cd ..

server.py

import BaseHTTPServer
import logging
import threading
import ssl


class RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
    server_version = "ssl_test.py"

    def do_GET(self):
        logging.info("HTTP request: %s", self.path)
        self.send_response(200)
        self.send_header('Content-type', 'text/plain')
        self.end_headers()
        self.wfile.write("%s\n" % self.path)


class ServerThread(object):
    """An HTTPS server running in a background thread.
    """
    def __init__(self, certfile, keyfile):
        self._certfile = certfile
        self._keyfile = keyfile
        self._thread = threading.Thread(target=self.run)
        self._thread.setDaemon(True)
        self._running = False
        self._condition = threading.Condition()

    def run(self):
        try:
            # http://www.piware.de/2011/01/creating-an-https-server-in-python/
            self._httpd = BaseHTTPServer.HTTPServer(
                    ('127.0.0.1', 4443),
                    RequestHandler)
            self._httpd.socket = ssl.wrap_socket(
                    self._httpd.socket,
                    certfile=self._certfile,
                    keyfile=self._keyfile,
                    server_side=True)
            logging.info("Server started")

            with self._condition:
                # Signals start() to return
                self._running = True
                self._condition.notifyAll()

            self._httpd.serve_forever()
            logging.info("Server finishing")
        except:
            logging.critical("Server failed")
            raise
        finally:
            with self._condition:
                # Signals stop() to return
                self._running = False
                self._condition.notifyAll()

    def start(self):
        logging.info("Starting HTTPS server")
        with self._condition:
            self._thread.start()
            # Waits for the server to be running
            self._condition.wait()
            if not self._running:
                raise RuntimeError("Failed to start server")

    def stop(self):
        logging.info("Stopping HTTPS server")
        with self._condition:
            self._httpd.shutdown()
            self._condition.wait()

ssl_test.py

import logging
import requests
from requests.exceptions import SSLError
import unittest

from server import ServerThread


class TestHTTPS(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        """Sets up a web server with a certificate from the custom CA.
        """
        cls.__http_server = ServerThread('certs/cert1.pem', 'certs/key1.pem')
        cls.__http_server.start()

    @classmethod
    def tearDownClass(cls):
        """Stops the web server.
        """
        cls.__http_server.stop()

    def test_default_valid(self):
        """Using default CA, access a valid HTTPS server."""
        requests.get('https://www.google.com/',
                     verify=True, timeout=3.0)

    def test_default_invalid(self):
        """Using default CA, access an invalid HTTPS server."""
        with self.assertRaisesRegexp(SSLError, "certificate verify failed"):
            requests.get('https://localhost:4443/default_invalid',
                         verify=True, timeout=3.0)

    def test_custom_default(self):
        """Using custom CA, connect to an originally valid certificate."""
        with self.assertRaisesRegexp(SSLError, "certificate verify failed"):
            requests.get('https://www.google.com/',
                         verify='certs/ca1.pem', timeout=3.0)

    def test_custom_valid(self):
        """Using custom CA, connect to a valid certificate (for that CA)."""
        requests.get('https://localhost:4443/custom_valid',
                     verify='certs/ca1.pem', timeout=3.0).raise_for_status()

    def test_custom_invalid(self):
        """Using custom CA, connect to an invalid certificate (for both CAs)."""
        with self.assertRaisesRegexp(SSLError, "certificate verify failed|block type is not "):
            requests.get('https://localhost:4443/custom_invalid',
                          verify='certs/ca2.pem', timeout=3.0)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    unittest.main()