Support GSSAPI channel bindings

README.rst

@@ -241,3 +241,21 @@ If you are having difficulty we suggest you configure logging. Issues with the
 underlying GSSAPI libraries will be made apparent. Additionally, copious debug
 information is made available which may assist in troubleshooting if you
 increase your log level all the way up to debug.
+
+Channel Bindings
+----------------
+
+Optional simplified support for channel bindings is available, but limited to
+the 'tls-server-end-point' bindings type (manual construction of different
+channel bindings can be achieved using the raw API). When requesting this kind
+of bindings python-cryptography must be available as request-gssapi will try
+to import its x509 module to process the peer certificate.
+
+.. code-block:: python
+
+    >>> import requests
+    >>> from requests_gssapi import HTTPSPNEGOAuth
+    >>> gssapi_auth = HTTPSPNEGOAuth(channel_bindings='tls-server-end-point')
+    >>> r = requests.get("https://windows.example.org/wsman", auth=gssapi_auth)
+    ...
+

src/requests_gssapi/gssapi_.py

@@ -107,6 +107,10 @@ class HTTPSPNEGOAuth(AuthBase):
     `sanitize_mutual_error_response` controls whether we should clean up
     server responses.  See the `SanitizedResponse` class.
 
+    `channel_bindings` can be used to pass channel bindings to GSSAPI.
+    The only accepted value is 'tls-server-end-point' which uses the TLS
+    server's certificate for the channel bindings. Default is `None`.
+
     """
 
     def __init__(
@@ -118,6 +122,7 @@ def __init__(
         creds=None,
         mech=SPNEGO,
         sanitize_mutual_error_response=True,
+        channel_bindings=None,
     ):
         self.context = {}
         self.pos = None
@@ -128,6 +133,9 @@ def __init__(
         self.creds = creds
         self.mech = mech if mech else SPNEGO
         self.sanitize_mutual_error_response = sanitize_mutual_error_response
+        if channel_bindings not in (None, "tls-server-end-point"):
+            raise ValueError("channel_bindings must be None or 'tls-server-end-point'")
+        self.channel_bindings = channel_bindings
 
     def generate_request_header(self, response, host, is_preemptive=False):
         """
@@ -144,6 +152,33 @@ def generate_request_header(self, response, host, is_preemptive=False):
         if self.mutual_authentication != DISABLED:
             gssflags.append(gssapi.RequirementFlag.mutual_authentication)
 
+        gss_cb = None
+        if self.channel_bindings == "tls-server-end-point":
+            if is_preemptive:
+                raise SPNEGOExchangeError(
+                    "channel_bindings were requested, but are unavailable for opportunistic authentication"
+                )
+            # The 'connection' attribute on raw is a public urllib3 API
+            # and can be None if the connection has been released.
+            elif getattr(response.raw, "connection", None) and getattr(response.raw.connection, "sock", None):
+                # Defer import so it's not a hard dependency.
+                from cryptography import x509
+
+                sock = response.raw.connection.sock
+
+                der_cert = sock.getpeercert(binary_form=True)
+                cert = x509.load_der_x509_certificate(der_cert)
+                hash = cert.signature_hash_algorithm
+                cert_hash = cert.fingerprint(hash)
+
+                app_data = b"tls-server-end-point:" + cert_hash
+                gss_cb = gssapi.raw.ChannelBindings(application_data=app_data)
+                log.debug("generate_request_header(): Successfully retrieved channel bindings")
+            else:
+                raise SPNEGOExchangeError(
+                    "channel_bindings were requested, but a socket could not be retrieved from the response"
+                )
+
         try:
             gss_stage = "initiating context"
             name = self.target_name
@@ -153,7 +188,12 @@ def generate_request_header(self, response, host, is_preemptive=False):
 
                 name = gssapi.Name(name, gssapi.NameType.hostbased_service)
             self.context[host] = gssapi.SecurityContext(
-                usage="initiate", flags=gssflags, name=name, creds=self.creds, mech=self.mech
+                usage="initiate",
+                flags=gssflags,
+                name=name,
+                creds=self.creds,
+                mech=self.mech,
+                channel_bindings=gss_cb,
             )
 
             gss_stage = "stepping context"

tests/test_requests_gssapi.py

@@ -98,7 +98,12 @@ def test_generate_request_header(self):
             auth = requests_gssapi.HTTPKerberosAuth()
             self.assertEqual(auth.generate_request_header(response, host), b64_negotiate_response)
             fake_init.assert_called_with(
-                name=gssapi_sname("[email protected]"), creds=None, mech=SPNEGO, flags=gssflags, usage="initiate"
+                name=gssapi_sname("[email protected]"),
+                creds=None,
+                mech=SPNEGO,
+                flags=gssflags,
+                usage="initiate",
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -113,7 +118,12 @@ def test_generate_request_header_init_error(self):
                 requests_gssapi.exceptions.SPNEGOExchangeError, auth.generate_request_header, response, host
             )
             fake_init.assert_called_with(
-                name=gssapi_sname("[email protected]"), usage="initiate", flags=gssflags, creds=None, mech=SPNEGO
+                name=gssapi_sname("[email protected]"),
+                usage="initiate",
+                flags=gssflags,
+                creds=None,
+                mech=SPNEGO,
+                channel_bindings=None,
             )
 
     def test_generate_request_header_step_error(self):
@@ -127,7 +137,12 @@ def test_generate_request_header_step_error(self):
                 requests_gssapi.exceptions.SPNEGOExchangeError, auth.generate_request_header, response, host
             )
             fake_init.assert_called_with(
-                name=gssapi_sname("[email protected]"), usage="initiate", flags=gssflags, creds=None, mech=SPNEGO
+                name=gssapi_sname("[email protected]"),
+                usage="initiate",
+                flags=gssflags,
+                creds=None,
+                mech=SPNEGO,
+                channel_bindings=None,
             )
             fail_resp.assert_called_with(b"token")
 
@@ -162,7 +177,12 @@ def test_authenticate_user(self):
             connection.send.assert_called_with(request)
             raw.release_conn.assert_called_with()
             fake_init.assert_called_with(
-                name=gssapi_sname("[email protected]"), flags=gssflags, usage="initiate", creds=None, mech=SPNEGO
+                name=gssapi_sname("[email protected]"),
+                flags=gssflags,
+                usage="initiate",
+                creds=None,
+                mech=SPNEGO,
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -197,7 +217,12 @@ def test_handle_401(self):
             connection.send.assert_called_with(request)
             raw.release_conn.assert_called_with()
             fake_init.assert_called_with(
-                name=gssapi_sname("[email protected]"), creds=None, mech=SPNEGO, flags=gssflags, usage="initiate"
+                name=gssapi_sname("[email protected]"),
+                creds=None,
+                mech=SPNEGO,
+                flags=gssflags,
+                usage="initiate",
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -402,7 +427,12 @@ def test_handle_response_401(self):
             connection.send.assert_called_with(request)
             raw.release_conn.assert_called_with()
             fake_init.assert_called_with(
-                name=gssapi_sname("[email protected]"), usage="initiate", flags=gssflags, creds=None, mech=SPNEGO
+                name=gssapi_sname("[email protected]"),
+                usage="initiate",
+                flags=gssflags,
+                creds=None,
+                mech=SPNEGO,
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -443,7 +473,12 @@ def connection_send(self, *args, **kwargs):
             connection.send.assert_called_with(request)
             raw.release_conn.assert_called_with()
             fake_init.assert_called_with(
-                name=gssapi_sname("[email protected]"), usage="initiate", flags=gssflags, creds=None, mech=SPNEGO
+                name=gssapi_sname("[email protected]"),
+                usage="initiate",
+                flags=gssflags,
+                creds=None,
+                mech=SPNEGO,
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -456,7 +491,12 @@ def test_generate_request_header_custom_service(self):
             auth = requests_gssapi.HTTPKerberosAuth(service="barfoo")
             auth.generate_request_header(response, host),
             fake_init.assert_called_with(
-                name=gssapi_sname("[email protected]"), usage="initiate", flags=gssflags, creds=None, mech=SPNEGO
+                name=gssapi_sname("[email protected]"),
+                usage="initiate",
+                flags=gssflags,
+                creds=None,
+                mech=SPNEGO,
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -496,6 +536,7 @@ def test_delegation(self):
                 flags=gssdelegflags,
                 creds=None,
                 mech=SPNEGO,
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -522,6 +563,7 @@ def test_principal_override(self):
                 flags=gssflags,
                 creds=b"fake creds",
                 mech=SPNEGO,
+                channel_bindings=None,
             )
 
     def test_realm_override(self):
@@ -538,6 +580,7 @@ def test_realm_override(self):
                 flags=gssflags,
                 creds=None,
                 mech=SPNEGO,
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -569,6 +612,7 @@ def test_explicit_creds(self):
                 flags=gssflags,
                 creds=b"fake creds",
                 mech=SPNEGO,
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -589,6 +633,7 @@ def test_explicit_mech(self):
                 flags=gssflags,
                 creds=None,
                 mech=b"fake mech",
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")
 
@@ -606,6 +651,7 @@ def test_target_name(self):
                 flags=gssflags,
                 creds=None,
                 mech=SPNEGO,
+                channel_bindings=None,
             )
             fake_resp.assert_called_with(b"token")