Add a global flag to strictly handle response headers

According to pep 3333 response headers are supposed to be
bytestrings. In Python 2 that means a str, and in Python 3, bytes.
Some wsgi servers, notably mod_wsgi, check that the values of
headers are correct, and make an error if they are not.

In at least some testing situations wsgi-intercept should
replicate this behavior so that an application can know that
it is not sending bad headers.

This is something that needs to be done in wsgi_intercept itself,
and not external tests, because many http client libraries will
transform response header values to native strings before providing
them to the caller.

However, many existing tools (for example selector) do not obey
this constraint, so we cannot simply turn this check on by default.
Instead we need to be able to declare that for a given context
we want it to be done. The easiest way to do this is to set a module
level global, as done in wsgi_intercept/tests/__init__.py.

In practical terms what this code does, is disallow

* in py2, sending responses with unicode values
* in py3, sending responses with str values

if wsgi_intercept.STRICT_RESPONSE_HEADERS == True.

Because this change introduced some six into
wsgi_intercept/__init__.py, additional changes to use six were made.

Note that since Python 2.7 io.BytesIO has been available so we no
longer need to do a conditional import. Interestingly, some testing
suggests that six.BytesIO in 2.7 is _not_ the same as io.BytesIO.

Fixes: #43

wsgi_intercept/__init__.py

@@ -123,25 +123,25 @@ from __future__ import print_function
 
 import sys
 import traceback
+from io import BytesIO
+
+# Don't use six here because it is unquote_to_bytes that we want in
+# Python 3.
+try:
+    from urllib.parse import unquote_to_bytes as url_unquote
+except ImportError:
+    from urllib import unquote as url_unquote
+
+import six
+from six.moves.http_client import HTTPConnection, HTTPSConnection
 
 
 __version__ = '1.3.2'
 
 
-try:
-    from http.client import HTTPConnection, HTTPSConnection
-except ImportError:
-    from httplib import HTTPConnection, HTTPSConnection
-
-try:
-    from io import BytesIO
-except ImportError:
-    from StringIO import StringIO as BytesIO
-
-try:
-    from urllib.parse import unquote_to_bytes as url_unquote
-except ImportError:
-    from urllib import unquote as url_unquote
+# Set this to True to cause response headers from the intercepted
+# app to be confirmed as bytestrings, behaving as some wsgi servers.
+STRICT_RESPONSE_HEADERS = False
 
 
 debuglevel = 0
@@ -210,7 +210,7 @@ def make_environ(inp, host, port, script_name):
     environ = {}
 
     method_line = inp.readline()
-    if sys.version_info[0] > 2:
+    if six.PY3:
         method_line = method_line.decode('ISO-8859-1')
 
     content_type = None
@@ -290,7 +290,7 @@ def make_environ(inp, host, port, script_name):
     # do to be like a server. Later various libraries will be forced
     # to decode and then reencode to get the UTF-8 that everyone
     # wants.
-    if sys.version_info[0] > 2:
+    if six.PY3:
         path_info = path_info.decode('latin-1')
 
     environ.update({
@@ -411,7 +411,8 @@ class wsgi_fake_socket:
 
         def start_response(status, headers, exc_info=None):
             # construct the HTTP request.
-            self.output.write(b"HTTP/1.0 " + status.encode('utf-8') + b"\n")
+            self.output.write(
+                b"HTTP/1.0 " + status.encode('ISO-8859-1') + b"\n")
             # Keep the reference of the headers list to write them only
             # when the whole application have been processed
             self.headers = headers
@@ -436,12 +437,18 @@ class wsgi_fake_socket:
         # send the headers
 
         for k, v in self.headers:
+            original_header = k
             try:
-                k = k.encode('utf-8')
+                k = k.encode('ISO-8859-1')
             except AttributeError:
                 pass
             try:
-                v = v.encode('utf-8')
+                if STRICT_RESPONSE_HEADERS:
+                    if not isinstance(v, six.binary_type):
+                        raise TypeError(
+                            'Header %s has value %s which is not a bytestring.'
+                            % (original_header, v))
+                v = v.encode('ISO-8859-1')
             except AttributeError:
                 pass
             self.output.write(k + b': ' + v + b"\n")

wsgi_intercept/tests/__init__.py

@@ -1,4 +1,9 @@
 import os
+import wsgi_intercept
+
+# Ensure that our test apps are sending strict headers.
+wsgi_intercept.STRICT_RESPONSE_HEADERS = True
+
 
 if os.environ.get('USER') == 'cdent':
     import warnings

wsgi_intercept/tests/test_response_headers.py

@@ -0,0 +1,85 @@
+"""Test response header validations.
+
+Response headers are supposed to be bytestrings and some servers,
+notably will experience an error if they are given headers with
+the wrong form. Since wsgi-intercept is standing in as a server,
+it should behave like one on this front. At the moment it does
+not. There are tests for how it delivers request headers, but
+not the other way round. Let's write some tests to fix that.
+"""
+
+import py.test
+import requests
+
+import wsgi_intercept
+from wsgi_intercept.interceptor import RequestsInterceptor
+
+
+class HeaderApp(object):
+    """A simple app that returns whatever headers we give it."""
+
+    def __init__(self, headers):
+        self.headers = headers
+
+    def __call__(self, environ, start_response):
+
+        headers = []
+        for header in self.headers:
+            headers.append((header, self.headers[header]))
+        start_response('200 OK', headers)
+        return ['']
+
+
+def app(headers):
+    return HeaderApp(headers)
+
+
+def test_header_app():
+    """Make sure the header apps returns headers.
+
+    Many libraries normalize headers to strings so we're not
+    going to get exact matches.
+    """
+    header_value = b'alpha'
+    header_value_str = 'alpha'
+
+    def header_app():
+        return app({'request-id': header_value})
+
+    with RequestsInterceptor(header_app) as url:
+        response = requests.get(url)
+
+    assert response.headers['request-id'] == header_value_str
+
+
+def test_encoding_violation():
+    """If the header is unicode we expect boom."""
+    header_value = u'alpha'
+
+    def header_app():
+        return app({'request-id': header_value})
+
+    # save original
+    strict_response_headers = wsgi_intercept.STRICT_RESPONSE_HEADERS
+
+    # With STRICT_RESPONSE_HEADERS True, response headers must be
+    # bytestrings.
+    with RequestsInterceptor(header_app) as url:
+        wsgi_intercept.STRICT_RESPONSE_HEADERS = True
+
+        with py.test.raises(TypeError) as error:
+            response = requests.get(url)
+
+        assert (str(error.value) ==
+            'Header request-id has value alpha which is not a bytestring.')
+
+        # When False, other types of strings are okay.
+        wsgi_intercept.STRICT_RESPONSE_HEADERS = False
+
+        response = requests.get(url)
+
+        assert response.headers['request-id'] == header_value
+
+    # reset back to saved original
+    wsgi_intercept.STRICT_RESPONSE_HEADERS = \
+        strict_response_headers

wsgi_intercept/tests/wsgi_app.py

@@ -13,20 +13,20 @@ except ImportError:
 def simple_app(environ, start_response):
     """Simplest possible application object"""
     status = '200 OK'
-    response_headers = [('Content-type', 'text/plain')]
+    response_headers = [('Content-type', b'text/plain')]
     start_response(status, response_headers)
     return [b'WSGI intercept successful!\n']
 
 
 def more_interesting_app(environ, start_response):
-    start_response('200 OK', [('Content-type', 'text/plain')])
+    start_response('200 OK', [('Content-type', b'text/plain')])
     return [pformat(environ).encode('utf-8')]
 
 
 def post_status_headers_app(environ, start_response):
     headers = []
     start_response('200 OK', headers)
-    headers.append(('Content-type', 'text/plain'))
+    headers.append(('Content-type', b'text/plain'))
     return [b'WSGI intercept successful!\n']