Use the new insecure channel credential object

src/python/grpcio/grpc/__init__.py

@@ -1729,6 +1729,16 @@ def xds_server_credentials(fallback_credentials):
     # tODO: Is this really how we get at the underlying server credentials?
     return ServerCredentials(_cygrpc.xds_server_credentials(fallback_credentials._credentials))
 
+def insecure_server_credentials():
+    """Creates a credentials object directing the server to use no credentials.
+
+    This object cannot be used directly in a call to `add_secure_port`.
+    Instead, it should be used to construct other credentials objects, e.g.
+    with xds_server_credentials.
+    """
+    return ServerCredentials(_cygrpc.insecure_server_credentials())
+
+
 def ssl_server_certificate_configuration(private_key_certificate_chain_pairs,
                                          root_certificates=None):
     """Creates a ServerCertificateConfiguration for use with a Server.

src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi

@@ -356,6 +356,14 @@ cdef class LocalChannelCredentials(ChannelCredentials):
 def channel_credentials_local(grpc_local_connect_type local_connect_type):
   return LocalChannelCredentials(local_connect_type)
 
+cdef class InsecureChannelCredentials(ChannelCredentials):
+
+  cdef grpc_channel_credentials *c(self) except *:
+    return grpc_insecure_credentials_create()
+
+def channel_credentials_insecure():
+  return InsecureChannelCredentials()
+
 def server_credentials_local(grpc_local_connect_type local_connect_type):
   cdef ServerCredentials credentials = ServerCredentials()
   credentials.c_credentials = grpc_local_server_credentials_create(local_connect_type)
@@ -366,6 +374,11 @@ def xds_server_credentials(ServerCredentials fallback_credentials):
   credentials.c_credentials = grpc_xds_server_credentials_create(fallback_credentials.c_credentials)
   return credentials
 
+def insecure_server_credentials():
+  cdef ServerCredentials credentials = ServerCredentials()
+  credentials.c_credentials = grpc_insecure_server_credentials_create()
+  return credentials
+
 cdef class ALTSChannelCredentials(ChannelCredentials):
 
   def __cinit__(self, list service_accounts):

src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi

@@ -517,9 +517,13 @@ cdef extern from "grpc/grpc_security.h":
   grpc_channel_credentials *grpc_xds_credentials_create(
       grpc_channel_credentials *fallback_creds) nogil
 
+  grpc_channel_credentials *grpc_insecure_credentials_create() nogil
+
   grpc_server_credentials *grpc_xds_server_credentials_create(
       grpc_server_credentials *fallback_creds) nogil
 
+  grpc_server_credentials *grpc_insecure_server_credentials_create() nogil
+
   grpc_call_credentials *grpc_composite_call_credentials_create(
       grpc_call_credentials *creds1, grpc_call_credentials *creds2,
       void *reserved) nogil

src/python/grpcio/grpc/_simple_stubs.py

@@ -60,13 +60,6 @@ else:
 def _create_channel(target: str, options: Sequence[Tuple[str, str]],
                     channel_credentials: Optional[grpc.ChannelCredentials],
                     compression: Optional[grpc.Compression]) -> grpc.Channel:
-    if channel_credentials is grpc.experimental.insecure_channel_credentials():
-        _LOGGER.debug(f"Creating insecure channel with options '{options}' " +
-                      f"and compression '{compression}'")
-        return grpc.insecure_channel(target,
-                                     options=options,
-                                     compression=compression)
-    else:
         _LOGGER.debug(
             f"Creating secure channel with credentials '{channel_credentials}', "
             + f"options '{options}' and compression '{compression}'")

src/python/grpcio/grpc/experimental/__init__.py

@@ -22,6 +22,7 @@ import sys
 import warnings
 
 import grpc
+from grpc._cython import cygrpc as _cygrpc
 
 _EXPERIMENTAL_APIS_USED = set()
 
@@ -41,19 +42,13 @@ class UsageError(Exception):
     """Raised by the gRPC library to indicate usage not allowed by the API."""
 
 
-_insecure_channel_credentials_sentinel = object()
-_insecure_channel_credentials = grpc.ChannelCredentials(
-    _insecure_channel_credentials_sentinel)
+# It's important that there be a single insecure credentials object so that its
+# hash is deterministic and can be used for indexing in the simple stubs cache.
+_insecure_channel_credentials = grpc.ChannelCredentials(_cygrpc.channel_credentials_insecure())
 
 
 def insecure_channel_credentials():
     """Creates a ChannelCredentials for use with an insecure channel.
-
-    THIS IS AN EXPERIMENTAL API.
-
-    This is not for use with secure_channel function. Intead, this should be
-    used with grpc.unary_unary, grpc.unary_stream, grpc.stream_unary, or
-    grpc.stream_stream.
     """
     return _insecure_channel_credentials
 

src/python/grpcio_tests/tests/unit/_xds_credentials_test.py

@@ -19,6 +19,7 @@ import logging
 from concurrent import futures
 
 import grpc
+import grpc.experimental
 from tests.unit import test_common
 from tests.unit import resources
 
@@ -54,6 +55,25 @@ class XdsCredentialsTest(unittest.TestCase):
             self.assertEqual(response, request)
         server.stop(None)
 
+    def test_xds_creds_fallback_insecure(self):
+        # Since there is no xDS server, the fallback credentials will be used.
+        # In this case, insecure.
+        server = grpc.server(futures.ThreadPoolExecutor())
+        server.add_generic_rpc_handlers((_GenericHandler(),))
+        server_fallback_creds = grpc.insecure_server_credentials()
+        server_creds = grpc.xds_server_credentials(server_fallback_creds)
+        port = server.add_secure_port("localhost:0", server_creds)
+        server.start()
+        # TODO: Move out of experimental.
+        channel_fallback_creds = grpc.experimental.insecure_channel_credentials()
+        channel_creds = grpc.xds_channel_credentials(channel_fallback_creds)
+        server_address = "localhost:{}".format(port)
+        with grpc.secure_channel(server_address, channel_creds) as channel:
+            request = b"abc"
+            response = channel.unary_unary("/test/method")(request, wait_for_ready=True)
+            self.assertEqual(response, request)
+        server.stop(None)
+
 if __name__ == "__main__":
     logging.basicConfig()
     unittest.main()