[Python Example] Add token based authentication example (#35477)

Add token based authentication example

<!--

If you know who should review your pull request, please assign it to
that
person, otherwise the pull request would get assigned randomly.

If your pull request is for a specific language, please add the
appropriate
lang label.

-->

3 files changed, 203 insertions, 0 deletions

diff --git a/examples/python/auth/README.md b/examples/python/auth/README.md
index 2fd044b8a3..fc66bb335e 100644
--- a/examples/python/auth/README.md
+++ b/examples/python/auth/README.md
@@ -110,3 +110,19 @@ call_credentials = grpc.composite_call_credentials(
     call_credentials_bar)
 stub.FooRpc(request, credentials=call_credentials)
 ```
+
+## Token-based authentication
+
+Instead of `AuthMetadataPlugin`, you can also use token-based authentication
+mechanisms using OAuth2 tokens or other customized tokens.
+
+OAuth2 tokens can be obtained using libraries like [google-auth](https://google-auth.readthedocs.io/en/master/user-guide.html):
+
+```Python
+import google.auth
+
+google_credentials, unused_project_id = google.auth.default()
+call_credentials = grpc.access_token_call_credentials(google_credentials.token)
+```
+
+After obtaining the token, the rest of the flow is documented in [token_based_auth_client.py](https://github.com/grpc/grpc/tree/master/examples/python/auth/token_based_auth_client.py) and [token_based_auth_server.py](https://github.com/grpc/grpc/tree/master/examples/python/auth/token_based_auth_server.py).
diff --git a/examples/python/auth/token_based_auth_client.py b/examples/python/auth/token_based_auth_client.py
new file mode 100644
index 0000000000..521d55df4c
--- /dev/null
+++ b/examples/python/auth/token_based_auth_client.py
@@ -0,0 +1,82 @@
+# Copyright 2024 The gRPC Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Client of the Python example of token based authentication mechanism."""
+
+import argparse
+import contextlib
+import logging
+
+import _credentials
+import grpc
+
+helloworld_pb2, helloworld_pb2_grpc = grpc.protos_and_services(
+    "helloworld.proto"
+)
+
+_LOGGER = logging.getLogger(__name__)
+_LOGGER.setLevel(logging.INFO)
+
+_SERVER_ADDR_TEMPLATE = "localhost:%d"
+
+
+@contextlib.contextmanager
+def create_client_channel(addr):
+    # Call credential object will be invoked for every single RPC
+    call_credentials = grpc.access_token_call_credentials(
+        "example_oauth2_token"
+    )
+    # Channel credential will be valid for the entire channel
+    channel_credential = grpc.ssl_channel_credentials(
+        _credentials.ROOT_CERTIFICATE
+    )
+    # Combining channel credentials and call credentials together
+    composite_credentials = grpc.composite_channel_credentials(
+        channel_credential,
+        call_credentials,
+    )
+    channel = grpc.secure_channel(addr, composite_credentials)
+    yield channel
+
+
+def send_rpc(channel):
+    stub = helloworld_pb2_grpc.GreeterStub(channel)
+    request = helloworld_pb2.HelloRequest(name="you")
+    try:
+        response = stub.SayHello(request)
+    except grpc.RpcError as rpc_error:
+        _LOGGER.error("Received error: %s", rpc_error)
+        return rpc_error
+    else:
+        _LOGGER.info("Received message: %s", response)
+        return response
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--port",
+        nargs="?",
+        type=int,
+        default=50051,
+        help="the address of server",
+    )
+    args = parser.parse_args()
+
+    with create_client_channel(_SERVER_ADDR_TEMPLATE % args.port) as channel:
+        send_rpc(channel)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    main()
diff --git a/examples/python/auth/token_based_auth_server.py b/examples/python/auth/token_based_auth_server.py
new file mode 100644
index 0000000000..e6b9c00ef2
--- /dev/null
+++ b/examples/python/auth/token_based_auth_server.py
@@ -0,0 +1,105 @@
+# Copyright 2024 The gRPC Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Server of the Python example of token based authentication mechanism."""
+
+import argparse
+from concurrent import futures
+import contextlib
+import logging
+
+import _credentials
+import grpc
+
+helloworld_pb2, helloworld_pb2_grpc = grpc.protos_and_services(
+    "helloworld.proto"
+)
+
+_LOGGER = logging.getLogger(__name__)
+_LOGGER.setLevel(logging.INFO)
+
+_LISTEN_ADDRESS_TEMPLATE = "localhost:%d"
+_AUTH_HEADER_KEY = "authorization"
+_AUTH_HEADER_VALUE = "Bearer example_oauth2_token"
+
+
+class SignatureValidationInterceptor(grpc.ServerInterceptor):
+    def __init__(self):
+        def abort(ignored_request, context):
+            context.abort(grpc.StatusCode.UNAUTHENTICATED, "Invalid signature")
+
+        self._abort_handler = grpc.unary_unary_rpc_method_handler(abort)
+
+    def intercept_service(self, continuation, handler_call_details):
+        # Example HandlerCallDetails object:
+        #     _HandlerCallDetails(
+        #       method=u'/helloworld.Greeter/SayHello',
+        #       invocation_metadata=...)
+        expected_metadata = (_AUTH_HEADER_KEY, _AUTH_HEADER_VALUE)
+        if expected_metadata in handler_call_details.invocation_metadata:
+            return continuation(handler_call_details)
+        else:
+            return self._abort_handler
+
+
+class SimpleGreeter(helloworld_pb2_grpc.GreeterServicer):
+    def SayHello(self, request, unused_context):
+        return helloworld_pb2.HelloReply(message="Hello, %s!" % request.name)
+
+
+@contextlib.contextmanager
+def run_server(port):
+    # Bind interceptor to server
+    server = grpc.server(
+        futures.ThreadPoolExecutor(),
+        interceptors=(SignatureValidationInterceptor(),),
+    )
+    helloworld_pb2_grpc.add_GreeterServicer_to_server(SimpleGreeter(), server)
+
+    # Loading credentials
+    server_credentials = grpc.ssl_server_credentials(
+        (
+            (
+                _credentials.SERVER_CERTIFICATE_KEY,
+                _credentials.SERVER_CERTIFICATE,
+            ),
+        )
+    )
+
+    # Pass down credentials
+    port = server.add_secure_port(
+        _LISTEN_ADDRESS_TEMPLATE % port, server_credentials
+    )
+
+    server.start()
+    try:
+        yield server, port
+    finally:
+        server.stop(0)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--port", nargs="?", type=int, default=50051, help="the listening port"
+    )
+    args = parser.parse_args()
+
+    with run_server(args.port) as (server, port):
+        logging.info("Server is listening at port :%d", port)
+        server.wait_for_termination()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    main()