PYTHON-3096 Finish implementation and tests for GSSAPI options

Author: blink1073

pymongo/asynchronous/auth.py

@@ -177,13 +177,20 @@ def _auth_key(nonce: str, username: str, password: str) -> str:
     return md5hash.hexdigest()
 
 
-def _canonicalize_hostname(hostname: str) -> str:
+def _canonicalize_hostname(hostname: str, option: str | bool) -> str:
     """Canonicalize hostname following MIT-krb5 behavior."""
     # https://github.com/krb5/krb5/blob/d406afa363554097ac48646a29249c04f498c88e/src/util/k5test.py#L505-L520
+    if option in [False, None]:
+        return hostname
+
     af, socktype, proto, canonname, sockaddr = socket.getaddrinfo(
         hostname, None, 0, 0, socket.IPPROTO_TCP, socket.AI_CANONNAME
     )[0]
 
+    # For forward just to resolve the cname as dns.lookup() will not return it.
+    if option == "forward":
+        return canonname.lower()
+
     try:
         name = socket.getnameinfo(sockaddr, socket.NI_NAMEREQD)
     except socket.gaierror:
@@ -205,9 +212,8 @@ async def _authenticate_gssapi(credentials: MongoCredential, conn: AsyncConnecti
         props = credentials.mechanism_properties
         # Starting here and continuing through the while loop below - establish
         # the security context. See RFC 4752, Section 3.1, first paragraph.
-        host = conn.address[0]
-        if props.canonicalize_host_name:
-            host = _canonicalize_hostname(host)
+        host = props.host_name or conn.address[0]
+        host = _canonicalize_hostname(host, props.canonicalize_host_name)
         service = props.service_name + "@" + host
         if props.service_realm is not None:
             service = service + "@" + props.service_realm

pymongo/synchronous/auth.py

@@ -174,13 +174,20 @@ def _auth_key(nonce: str, username: str, password: str) -> str:
     return md5hash.hexdigest()
 
 
-def _canonicalize_hostname(hostname: str) -> str:
+def _canonicalize_hostname(hostname: str, option: str | bool) -> str:
     """Canonicalize hostname following MIT-krb5 behavior."""
     # https://github.com/krb5/krb5/blob/d406afa363554097ac48646a29249c04f498c88e/src/util/k5test.py#L505-L520
+    if option in [False, None]:
+        return hostname
+
     af, socktype, proto, canonname, sockaddr = socket.getaddrinfo(
         hostname, None, 0, 0, socket.IPPROTO_TCP, socket.AI_CANONNAME
     )[0]
 
+    # For forward just to resolve the cname as dns.lookup() will not return it.
+    if option == "forward":
+        return canonname.lower()
+
     try:
         name = socket.getnameinfo(sockaddr, socket.NI_NAMEREQD)
     except socket.gaierror:
@@ -202,9 +209,8 @@ def _authenticate_gssapi(credentials: MongoCredential, conn: Connection) -> None
         props = credentials.mechanism_properties
         # Starting here and continuing through the while loop below - establish
         # the security context. See RFC 4752, Section 3.1, first paragraph.
-        host = conn.address[0]
-        if props.canonicalize_host_name:
-            host = _canonicalize_hostname(host)
+        host = props.host_name or conn.address[0]
+        host = _canonicalize_hostname(host, props.canonicalize_host_name)
         service = props.service_name + "@" + host
         if props.service_realm is not None:
             service = service + "@" + props.service_realm

test/asynchronous/test_auth.py

@@ -96,10 +96,10 @@ def setUpClass(cls):
         cls.service_realm_required = (
             GSSAPI_SERVICE_REALM is not None and GSSAPI_SERVICE_REALM not in GSSAPI_PRINCIPAL
         )
-        mech_properties = f"SERVICE_NAME:{GSSAPI_SERVICE_NAME}"
-        mech_properties += f",CANONICALIZE_HOST_NAME:{GSSAPI_CANONICALIZE}"
-        if GSSAPI_SERVICE_REALM is not None:
-            mech_properties += f",SERVICE_REALM:{GSSAPI_SERVICE_REALM}"
+        mech_properties = dict(
+            SERVICE_NAME=GSSAPI_SERVICE_NAME, CANONICALIZE_HOST_NAME=GSSAPI_CANONICALIZE
+        )
+        mech_properties["SERVICE_REALM"] = GSSAPI_SERVICE_REALM
         cls.mech_properties = mech_properties
 
     async def test_credentials_hashing(self):
@@ -268,6 +268,64 @@ async def test_gssapi_threaded(self):
                 thread.join()
                 self.assertTrue(thread.success)
 
+    async def test_gssapi_canonicalize_host_name(self):
+        # Use the equivalent named CANONICALIZE_HOST_NAME.
+        if self.mech_properties["CANONICALIZE_HOST_NAME"] == "true":
+            self.mech_properties["CANONICALIZE_HOST_NAME"] = "forwardAndReverse"
+        else:
+            self.mech_properties["CANONICALIZE_HOST_NAME"] = "none"
+        client = self.simple_client(
+            GSSAPI_HOST,
+            GSSAPI_PORT,
+            username=GSSAPI_PRINCIPAL,
+            password=GSSAPI_PASS,
+            authMechanism="GSSAPI",
+            authMechanismProperties=self.mech_properties,
+        )
+        await client.server_info()
+
+        if self.mech_properties["CANONICALIZE_HOST_NAME"] == "none":
+            return
+
+        # Test with "forward".
+        self.mech_properties["CANONICALIZE_HOST_NAME"] = "forward"
+        client = self.simple_client(
+            GSSAPI_HOST,
+            GSSAPI_PORT,
+            username=GSSAPI_PRINCIPAL,
+            password=GSSAPI_PASS,
+            authMechanism="GSSAPI",
+            authMechanismProperties=self.mech_properties,
+        )
+        await client.server_info()
+
+    async def test_gssapi_host_name(self):
+        props = self.mech_properties
+        props["SERVICE_HOST"] = "example.com"
+
+        # Authenticate with authMechanismProperties.
+        client = self.simple_client(
+            GSSAPI_HOST,
+            GSSAPI_PORT,
+            username=GSSAPI_PRINCIPAL,
+            password=GSSAPI_PASS,
+            authMechanism="GSSAPI",
+            authMechanismProperties=self.mech_properties,
+        )
+        with self.assertRaises(OperationFailure):
+            await client.server_info()
+
+        props["SERVICE_HOST"] = GSSAPI_HOST
+        client = self.simple_client(
+            GSSAPI_HOST,
+            GSSAPI_PORT,
+            username=GSSAPI_PRINCIPAL,
+            password=GSSAPI_PASS,
+            authMechanism="GSSAPI",
+            authMechanismProperties=self.mech_properties,
+        )
+        await client.server_info()
+
 
 class TestSASLPlain(AsyncPyMongoTestCase):
     @classmethod

test/test_auth.py

@@ -96,10 +96,10 @@ def setUpClass(cls):
         cls.service_realm_required = (
             GSSAPI_SERVICE_REALM is not None and GSSAPI_SERVICE_REALM not in GSSAPI_PRINCIPAL
         )
-        mech_properties = f"SERVICE_NAME:{GSSAPI_SERVICE_NAME}"
-        mech_properties += f",CANONICALIZE_HOST_NAME:{GSSAPI_CANONICALIZE}"
-        if GSSAPI_SERVICE_REALM is not None:
-            mech_properties += f",SERVICE_REALM:{GSSAPI_SERVICE_REALM}"
+        mech_properties = dict(
+            SERVICE_NAME=GSSAPI_SERVICE_NAME, CANONICALIZE_HOST_NAME=GSSAPI_CANONICALIZE
+        )
+        mech_properties["SERVICE_REALM"] = GSSAPI_SERVICE_REALM
         cls.mech_properties = mech_properties
 
     def test_credentials_hashing(self):
@@ -268,6 +268,64 @@ def test_gssapi_threaded(self):
                 thread.join()
                 self.assertTrue(thread.success)
 
+    def test_gssapi_canonicalize_host_name(self):
+        # Use the equivalent named CANONICALIZE_HOST_NAME.
+        if self.mech_properties["CANONICALIZE_HOST_NAME"] == "true":
+            self.mech_properties["CANONICALIZE_HOST_NAME"] = "forwardAndReverse"
+        else:
+            self.mech_properties["CANONICALIZE_HOST_NAME"] = "none"
+        client = self.simple_client(
+            GSSAPI_HOST,
+            GSSAPI_PORT,
+            username=GSSAPI_PRINCIPAL,
+            password=GSSAPI_PASS,
+            authMechanism="GSSAPI",
+            authMechanismProperties=self.mech_properties,
+        )
+        client.server_info()
+
+        if self.mech_properties["CANONICALIZE_HOST_NAME"] == "none":
+            return
+
+        # Test with "forward".
+        self.mech_properties["CANONICALIZE_HOST_NAME"] = "forward"
+        client = self.simple_client(
+            GSSAPI_HOST,
+            GSSAPI_PORT,
+            username=GSSAPI_PRINCIPAL,
+            password=GSSAPI_PASS,
+            authMechanism="GSSAPI",
+            authMechanismProperties=self.mech_properties,
+        )
+        client.server_info()
+
+    def test_gssapi_host_name(self):
+        props = self.mech_properties
+        props["SERVICE_HOST"] = "example.com"
+
+        # Authenticate with authMechanismProperties.
+        client = self.simple_client(
+            GSSAPI_HOST,
+            GSSAPI_PORT,
+            username=GSSAPI_PRINCIPAL,
+            password=GSSAPI_PASS,
+            authMechanism="GSSAPI",
+            authMechanismProperties=self.mech_properties,
+        )
+        with self.assertRaises(OperationFailure):
+            client.server_info()
+
+        props["SERVICE_HOST"] = GSSAPI_HOST
+        client = self.simple_client(
+            GSSAPI_HOST,
+            GSSAPI_PORT,
+            username=GSSAPI_PRINCIPAL,
+            password=GSSAPI_PASS,
+            authMechanism="GSSAPI",
+            authMechanismProperties=self.mech_properties,
+        )
+        client.server_info()
+
 
 class TestSASLPlain(PyMongoTestCase):
     @classmethod