Support GSSAPI channel bindings

This change adds support for GSSAPI channel bindings, which helps to protect
against man-in-the-middle relay attacks by tying the authentication to the
underlying secure channel.

A `channel_bindings` parameter is added to `HTTPSPNEGOAuth`. When set to 'tls-
server-end-point', the server's TLS certificate is retrieved from the socket,
hashed, and used to create the GSSAPI channel bindings.

This feature requires the `cryptography` library as an optional dependency. If
it's not available, channel bindings cannot be used and a warning is logged.

Co-authored-by: Gemini <[email protected]>
Signed-off-by: Simo Sorce <[email protected]>

Author: simo5

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,37 @@ 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:
+                log.warning("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):
+                try:
+                    # 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.ChannelBindings(application_data=app_data)
+                    log.debug("generate_request_header(): Successfully retrieved channel bindings")
+                except ImportError:
+                    log.warning("Could not import cryptography, python-cryptography is required for this feature.")
+                except Exception:
+                    log.warning(
+                        "Failed to get channel bindings from socket",
+                        exc_info=True,
+                    )
+            else:
+                log.warning("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 +192,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")