Merge pull request #12374 from markdroth/plugin_credentials_api_fix

Change plugin credentials API to support both sync and async modes

doc/environment_variables.md

@@ -58,6 +58,7 @@ some configuration as environment variables that can be set.
     completion queue
   - round_robin - traces the round_robin load balancing policy
   - pick_first - traces the pick first load balancing policy
+  - plugin_credentials - traces plugin credentials
   - resource_quota - trace resource quota objects internals
   - glb - traces the grpclb load balancer
   - queue_pluck

include/grpc/grpc_security.h

@@ -249,19 +249,40 @@ typedef struct {
   void *reserved;
 } grpc_auth_metadata_context;
 
+/** Maximum number of metadata entries returnable by a credentials plugin via
+    a synchronous return. */
+#define GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX 4
+
 /** grpc_metadata_credentials plugin is an API user provided structure used to
    create grpc_credentials objects that can be set on a channel (composed) or
    a call. See grpc_credentials_metadata_create_from_plugin below.
    The grpc client stack will call the get_metadata method of the plugin for
    every call in scope for the credentials created from it. */
 typedef struct {
-  /** The implementation of this method has to be non-blocking.
-     - context is the information that can be used by the plugin to create auth
-       metadata.
-     - cb is the callback that needs to be called when the metadata is ready.
-     - user_data needs to be passed as the first parameter of the callback. */
-  void (*get_metadata)(void *state, grpc_auth_metadata_context context,
-                       grpc_credentials_plugin_metadata_cb cb, void *user_data);
+  /** The implementation of this method has to be non-blocking, but can
+     be performed synchronously or asynchronously.
+
+     If processing occurs synchronously, returns non-zero and populates
+     creds_md, num_creds_md, status, and error_details.  In this case,
+     the caller takes ownership of the entries in creds_md and of
+     error_details.  Note that if the plugin needs to return more than
+     GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX entries in creds_md, it must
+     return asynchronously.
+
+     If processing occurs asynchronously, returns zero and invokes \a cb
+     when processing is completed.  \a user_data will be passed as the
+     first parameter of the callback.  NOTE: \a cb MUST be invoked in a
+     different thread, not from the thread in which \a get_metadata() is
+     invoked.
+
+     \a context is the information that can be used by the plugin to create
+     auth metadata. */
+  int (*get_metadata)(
+      void *state, grpc_auth_metadata_context context,
+      grpc_credentials_plugin_metadata_cb cb, void *user_data,
+      grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+      size_t *num_creds_md, grpc_status_code *status,
+      const char **error_details);
 
   /** Destroys the plugin state. */
   void (*destroy)(void *state);

src/core/lib/security/credentials/plugin/plugin_credentials.c

@@ -31,6 +31,9 @@
 #include "src/core/lib/surface/api_trace.h"
 #include "src/core/lib/surface/validate_metadata.h"
 
+grpc_tracer_flag grpc_plugin_credentials_trace =
+    GRPC_TRACER_INITIALIZER(false, "plugin_credentials");
+
 static void plugin_destruct(grpc_exec_ctx *exec_ctx,
                             grpc_call_credentials *creds) {
   grpc_plugin_credentials *c = (grpc_plugin_credentials *)creds;
@@ -53,6 +56,62 @@ static void pending_request_remove_locked(
   }
 }
 
+// Checks if the request has been cancelled.
+// If not, removes it from the pending list, so that it cannot be
+// cancelled out from under us.
+// When this returns, r->cancelled indicates whether the request was
+// cancelled before completion.
+static void pending_request_complete(
+    grpc_exec_ctx *exec_ctx, grpc_plugin_credentials_pending_request *r) {
+  gpr_mu_lock(&r->creds->mu);
+  if (!r->cancelled) pending_request_remove_locked(r->creds, r);
+  gpr_mu_unlock(&r->creds->mu);
+  // Ref to credentials not needed anymore.
+  grpc_call_credentials_unref(exec_ctx, &r->creds->base);
+}
+
+static grpc_error *process_plugin_result(
+    grpc_exec_ctx *exec_ctx, grpc_plugin_credentials_pending_request *r,
+    const grpc_metadata *md, size_t num_md, grpc_status_code status,
+    const char *error_details) {
+  grpc_error *error = GRPC_ERROR_NONE;
+  if (status != GRPC_STATUS_OK) {
+    char *msg;
+    gpr_asprintf(&msg, "Getting metadata from plugin failed with error: %s",
+                 error_details);
+    error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg);
+    gpr_free(msg);
+  } else {
+    bool seen_illegal_header = false;
+    for (size_t i = 0; i < num_md; ++i) {
+      if (!GRPC_LOG_IF_ERROR("validate_metadata_from_plugin",
+                             grpc_validate_header_key_is_legal(md[i].key))) {
+        seen_illegal_header = true;
+        break;
+      } else if (!grpc_is_binary_header(md[i].key) &&
+                 !GRPC_LOG_IF_ERROR(
+                     "validate_metadata_from_plugin",
+                     grpc_validate_header_nonbin_value_is_legal(md[i].value))) {
+        gpr_log(GPR_ERROR, "Plugin added invalid metadata value.");
+        seen_illegal_header = true;
+        break;
+      }
+    }
+    if (seen_illegal_header) {
+      error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Illegal metadata");
+    } else {
+      for (size_t i = 0; i < num_md; ++i) {
+        grpc_mdelem mdelem = grpc_mdelem_from_slices(
+            exec_ctx, grpc_slice_ref_internal(md[i].key),
+            grpc_slice_ref_internal(md[i].value));
+        grpc_credentials_mdelem_array_add(r->md_array, mdelem);
+        GRPC_MDELEM_UNREF(exec_ctx, mdelem);
+      }
+    }
+  }
+  return error;
+}
+
 static void plugin_md_request_metadata_ready(void *request,
                                              const grpc_metadata *md,
                                              size_t num_md,
@@ -64,54 +123,24 @@ static void plugin_md_request_metadata_ready(void *request,
       NULL, NULL);
   grpc_plugin_credentials_pending_request *r =
       (grpc_plugin_credentials_pending_request *)request;
-  // Check if the request has been cancelled.
-  // If not, remove it from the pending list, so that it cannot be
-  // cancelled out from under us.
-  gpr_mu_lock(&r->creds->mu);
-  if (!r->cancelled) pending_request_remove_locked(r->creds, r);
-  gpr_mu_unlock(&r->creds->mu);
-  grpc_call_credentials_unref(&exec_ctx, &r->creds->base);
+  if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) {
+    gpr_log(GPR_INFO,
+            "plugin_credentials[%p]: request %p: plugin returned "
+            "asynchronously",
+            r->creds, r);
+  }
+  // Remove request from pending list if not previously cancelled.
+  pending_request_complete(&exec_ctx, r);
   // If it has not been cancelled, process it.
   if (!r->cancelled) {
-    if (status != GRPC_STATUS_OK) {
-      char *msg;
-      gpr_asprintf(&msg, "Getting metadata from plugin failed with error: %s",
-                   error_details);
-      GRPC_CLOSURE_SCHED(&exec_ctx, r->on_request_metadata,
-                         GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg));
-      gpr_free(msg);
-    } else {
-      bool seen_illegal_header = false;
-      for (size_t i = 0; i < num_md; ++i) {
-        if (!GRPC_LOG_IF_ERROR("validate_metadata_from_plugin",
-                               grpc_validate_header_key_is_legal(md[i].key))) {
-          seen_illegal_header = true;
-          break;
-        } else if (!grpc_is_binary_header(md[i].key) &&
-                   !GRPC_LOG_IF_ERROR(
-                       "validate_metadata_from_plugin",
-                       grpc_validate_header_nonbin_value_is_legal(
-                           md[i].value))) {
-          gpr_log(GPR_ERROR, "Plugin added invalid metadata value.");
-          seen_illegal_header = true;
-          break;
-        }
-      }
-      if (seen_illegal_header) {
-        GRPC_CLOSURE_SCHED(
-            &exec_ctx, r->on_request_metadata,
-            GRPC_ERROR_CREATE_FROM_STATIC_STRING("Illegal metadata"));
-      } else {
-        for (size_t i = 0; i < num_md; ++i) {
-          grpc_mdelem mdelem = grpc_mdelem_from_slices(
-              &exec_ctx, grpc_slice_ref_internal(md[i].key),
-              grpc_slice_ref_internal(md[i].value));
-          grpc_credentials_mdelem_array_add(r->md_array, mdelem);
-          GRPC_MDELEM_UNREF(&exec_ctx, mdelem);
-        }
-        GRPC_CLOSURE_SCHED(&exec_ctx, r->on_request_metadata, GRPC_ERROR_NONE);
-      }
-    }
+    grpc_error *error =
+        process_plugin_result(&exec_ctx, r, md, num_md, status, error_details);
+    GRPC_CLOSURE_SCHED(&exec_ctx, r->on_request_metadata, error);
+  } else if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) {
+    gpr_log(GPR_INFO,
+            "plugin_credentials[%p]: request %p: plugin was previously "
+            "cancelled",
+            r->creds, r);
   }
   gpr_free(r);
   grpc_exec_ctx_finish(&exec_ctx);
@@ -125,6 +154,7 @@ static bool plugin_get_request_metadata(grpc_exec_ctx *exec_ctx,
                                         grpc_closure *on_request_metadata,
                                         grpc_error **error) {
   grpc_plugin_credentials *c = (grpc_plugin_credentials *)creds;
+  bool retval = true;  // Synchronous return.
   if (c->plugin.get_metadata != NULL) {
     // Create pending_request object.
     grpc_plugin_credentials_pending_request *pending_request =
@@ -142,12 +172,60 @@ static bool plugin_get_request_metadata(grpc_exec_ctx *exec_ctx,
     c->pending_requests = pending_request;
     gpr_mu_unlock(&c->mu);
     // Invoke the plugin.  The callback holds a ref to us.
+    if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) {
+      gpr_log(GPR_INFO, "plugin_credentials[%p]: request %p: invoking plugin",
+              c, pending_request);
+    }
     grpc_call_credentials_ref(creds);
-    c->plugin.get_metadata(c->plugin.state, context,
-                           plugin_md_request_metadata_ready, pending_request);
-    return false;
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX];
+    size_t num_creds_md = 0;
+    grpc_status_code status = GRPC_STATUS_OK;
+    const char *error_details = NULL;
+    if (!c->plugin.get_metadata(c->plugin.state, context,
+                                plugin_md_request_metadata_ready,
+                                pending_request, creds_md, &num_creds_md,
+                                &status, &error_details)) {
+      if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) {
+        gpr_log(GPR_INFO,
+                "plugin_credentials[%p]: request %p: plugin will return "
+                "asynchronously",
+                c, pending_request);
+      }
+      return false;  // Asynchronous return.
+    }
+    // Returned synchronously.
+    // Remove request from pending list if not previously cancelled.
+    pending_request_complete(exec_ctx, pending_request);
+    // If the request was cancelled, the error will have been returned
+    // asynchronously by plugin_cancel_get_request_metadata(), so return
+    // false.  Otherwise, process the result.
+    if (pending_request->cancelled) {
+      if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) {
+        gpr_log(GPR_INFO,
+                "plugin_credentials[%p]: request %p was cancelled, error "
+                "will be returned asynchronously",
+                c, pending_request);
+      }
+      retval = false;
+    } else {
+      if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) {
+        gpr_log(GPR_INFO,
+                "plugin_credentials[%p]: request %p: plugin returned "
+                "synchronously",
+                c, pending_request);
+      }
+      *error = process_plugin_result(exec_ctx, pending_request, creds_md,
+                                     num_creds_md, status, error_details);
+    }
+    // Clean up.
+    for (size_t i = 0; i < num_creds_md; ++i) {
+      grpc_slice_unref_internal(exec_ctx, creds_md[i].key);
+      grpc_slice_unref_internal(exec_ctx, creds_md[i].value);
+    }
+    gpr_free((void *)error_details);
+    gpr_free(pending_request);
   }
-  return true;
+  return retval;
 }
 
 static void plugin_cancel_get_request_metadata(
@@ -159,6 +237,10 @@ static void plugin_cancel_get_request_metadata(
            c->pending_requests;
        pending_request != NULL; pending_request = pending_request->next) {
     if (pending_request->md_array == md_array) {
+      if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) {
+        gpr_log(GPR_INFO, "plugin_credentials[%p]: cancelling request %p", c,
+                pending_request);
+      }
       pending_request->cancelled = true;
       GRPC_CLOSURE_SCHED(exec_ctx, pending_request->on_request_metadata,
                          GRPC_ERROR_REF(error));

src/core/lib/security/credentials/plugin/plugin_credentials.h

@@ -21,6 +21,8 @@
 
 #include "src/core/lib/security/credentials/credentials.h"
 
+extern grpc_tracer_flag grpc_plugin_credentials_trace;
+
 struct grpc_plugin_credentials;
 
 typedef struct grpc_plugin_credentials_pending_request {

src/core/lib/surface/init_secure.c

@@ -25,6 +25,7 @@
 
 #include "src/core/lib/debug/trace.h"
 #include "src/core/lib/security/credentials/credentials.h"
+#include "src/core/lib/security/credentials/plugin/plugin_credentials.h"
 #include "src/core/lib/security/transport/auth_filters.h"
 #include "src/core/lib/security/transport/secure_endpoint.h"
 #include "src/core/lib/security/transport/security_connector.h"
@@ -84,4 +85,7 @@ void grpc_register_security_filters(void) {
                                    maybe_prepend_server_auth_filter, NULL);
 }
 
-void grpc_security_init() { grpc_security_register_handshaker_factories(); }
+void grpc_security_init() {
+  grpc_security_register_handshaker_factories();
+  grpc_register_tracer(&grpc_plugin_credentials_trace);
+}

src/cpp/client/secure_credentials.cc

@@ -21,6 +21,7 @@
 #include <grpc++/impl/grpc_library.h>
 #include <grpc++/support/channel_arguments.h>
 #include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
 #include "src/cpp/client/create_channel_internal.h"
 #include "src/cpp/common/secure_auth_context.h"
 
@@ -169,28 +170,50 @@ void MetadataCredentialsPluginWrapper::Destroy(void* wrapper) {
   delete w;
 }
 
-void MetadataCredentialsPluginWrapper::GetMetadata(
+int MetadataCredentialsPluginWrapper::GetMetadata(
     void* wrapper, grpc_auth_metadata_context context,
-    grpc_credentials_plugin_metadata_cb cb, void* user_data) {
+    grpc_credentials_plugin_metadata_cb cb, void* user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t* num_creds_md, grpc_status_code* status,
+    const char** error_details) {
   GPR_ASSERT(wrapper);
   MetadataCredentialsPluginWrapper* w =
       reinterpret_cast<MetadataCredentialsPluginWrapper*>(wrapper);
   if (!w->plugin_) {
-    cb(user_data, NULL, 0, GRPC_STATUS_OK, NULL);
-    return;
+    *num_creds_md = 0;
+    *status = GRPC_STATUS_OK;
+    *error_details = nullptr;
+    return true;
   }
   if (w->plugin_->IsBlocking()) {
+    // Asynchronous return.
     w->thread_pool_->Add(
         std::bind(&MetadataCredentialsPluginWrapper::InvokePlugin, w, context,
-                  cb, user_data));
+                  cb, user_data, nullptr, nullptr, nullptr, nullptr));
+    return 0;
   } else {
-    w->InvokePlugin(context, cb, user_data);
+    // Synchronous return.
+    w->InvokePlugin(context, cb, user_data, creds_md, num_creds_md, status,
+                    error_details);
+    return 1;
   }
 }
 
+namespace {
+
+void UnrefMetadata(const std::vector<grpc_metadata>& md) {
+  for (auto it = md.begin(); it != md.end(); ++it) {
+    grpc_slice_unref(it->key);
+    grpc_slice_unref(it->value);
+  }
+}
+
+}  // namespace
+
 void MetadataCredentialsPluginWrapper::InvokePlugin(
     grpc_auth_metadata_context context, grpc_credentials_plugin_metadata_cb cb,
-    void* user_data) {
+    void* user_data, grpc_metadata creds_md[4], size_t* num_creds_md,
+    grpc_status_code* status_code, const char** error_details) {
   std::multimap<grpc::string, grpc::string> metadata;
 
   // const_cast is safe since the SecureAuthContext does not take owndership and
@@ -208,12 +231,31 @@ void MetadataCredentialsPluginWrapper::InvokePlugin(
     md_entry.flags = 0;
     md.push_back(md_entry);
   }
-  cb(user_data, md.empty() ? nullptr : &md[0], md.size(),
-     static_cast<grpc_status_code>(status.error_code()),
-     status.error_message().c_str());
-  for (auto it = md.begin(); it != md.end(); ++it) {
-    grpc_slice_unref(it->key);
-    grpc_slice_unref(it->value);
+  if (creds_md != nullptr) {
+    // Synchronous return.
+    if (md.size() > GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX) {
+      *num_creds_md = 0;
+      *status_code = GRPC_STATUS_INTERNAL;
+      *error_details = gpr_strdup(
+          "blocking plugin credentials returned too many metadata keys");
+      UnrefMetadata(md);
+    } else {
+      for (const auto& elem : md) {
+        creds_md[*num_creds_md].key = elem.key;
+        creds_md[*num_creds_md].value = elem.value;
+        creds_md[*num_creds_md].flags = elem.flags;
+        ++(*num_creds_md);
+      }
+      *status_code = static_cast<grpc_status_code>(status.error_code());
+      *error_details =
+          status.ok() ? nullptr : gpr_strdup(status.error_message().c_str());
+    }
+  } else {
+    // Asynchronous return.
+    cb(user_data, md.empty() ? nullptr : &md[0], md.size(),
+       static_cast<grpc_status_code>(status.error_code()),
+       status.error_message().c_str());
+    UnrefMetadata(md);
   }
 }
 

src/cpp/client/secure_credentials.h

@@ -58,16 +58,23 @@ class SecureCallCredentials final : public CallCredentials {
 class MetadataCredentialsPluginWrapper final : private GrpcLibraryCodegen {
  public:
   static void Destroy(void* wrapper);
-  static void GetMetadata(void* wrapper, grpc_auth_metadata_context context,
-                          grpc_credentials_plugin_metadata_cb cb,
-                          void* user_data);
+  static int GetMetadata(
+      void* wrapper, grpc_auth_metadata_context context,
+      grpc_credentials_plugin_metadata_cb cb, void* user_data,
+      grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+      size_t* num_creds_md, grpc_status_code* status,
+      const char** error_details);
 
   explicit MetadataCredentialsPluginWrapper(
       std::unique_ptr<MetadataCredentialsPlugin> plugin);
 
  private:
-  void InvokePlugin(grpc_auth_metadata_context context,
-                    grpc_credentials_plugin_metadata_cb cb, void* user_data);
+  void InvokePlugin(
+      grpc_auth_metadata_context context,
+      grpc_credentials_plugin_metadata_cb cb, void* user_data,
+      grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+      size_t* num_creds_md, grpc_status_code* status,
+      const char** error_details);
   std::unique_ptr<ThreadPoolInterface> thread_pool_;
   std::unique_ptr<MetadataCredentialsPlugin> plugin_;
 };

src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs

@@ -61,12 +61,9 @@ namespace Grpc.Core.Internal
 
             try
             {
-                var context = new AuthInterceptorContext(Marshal.PtrToStringAnsi(serviceUrlPtr),
-                                                         Marshal.PtrToStringAnsi(methodNamePtr));
-                // Don't await, we are in a native callback and need to return.
-                #pragma warning disable 4014
-                GetMetadataAsync(context, callbackPtr, userDataPtr);
-                #pragma warning restore 4014
+                var context = new AuthInterceptorContext(Marshal.PtrToStringAnsi(serviceUrlPtr), Marshal.PtrToStringAnsi(methodNamePtr));
+                // Make a guarantee that credentials_notify_from_plugin is invoked async to be compliant with c-core API.
+                ThreadPool.QueueUserWorkItem(async (stateInfo) => await GetMetadataAsync(context, callbackPtr, userDataPtr));
             }
             catch (Exception e)
             {

src/csharp/Grpc.IntegrationTesting/MetadataCredentialsTest.cs

@@ -89,6 +89,54 @@ namespace Grpc.IntegrationTesting
             client.UnaryCall(new SimpleRequest { }, new CallOptions(credentials: callCredentials));
         }
 
+        [Test]
+        public async Task MetadataCredentials_Composed()
+        {
+            var first = CallCredentials.FromInterceptor(new AsyncAuthInterceptor((context, metadata) => {
+                // Attempt to exercise the case where async callback is inlineable/synchronously-runnable.
+                metadata.Add("first_authorization", "FIRST_SECRET_TOKEN");
+                return TaskUtils.CompletedTask;
+            }));
+            var second = CallCredentials.FromInterceptor(new AsyncAuthInterceptor((context, metadata) => {
+                metadata.Add("second_authorization", "SECOND_SECRET_TOKEN");
+                return TaskUtils.CompletedTask;
+            }));
+            var third = CallCredentials.FromInterceptor(new AsyncAuthInterceptor((context, metadata) => {
+                metadata.Add("third_authorization", "THIRD_SECRET_TOKEN");
+                return TaskUtils.CompletedTask;
+            }));
+            var channelCredentials = ChannelCredentials.Create(TestCredentials.CreateSslCredentials(),
+                CallCredentials.Compose(first, second, third));
+            channel = new Channel(Host, server.Ports.Single().BoundPort, channelCredentials, options);
+            var client = new TestService.TestServiceClient(channel);
+            var call = client.StreamingOutputCall(new StreamingOutputCallRequest { });
+            Assert.IsTrue(await call.ResponseStream.MoveNext());
+            Assert.IsFalse(await call.ResponseStream.MoveNext());
+        }
+
+        [Test]
+        public async Task MetadataCredentials_ComposedPerCall()
+        {
+            channel = new Channel(Host, server.Ports.Single().BoundPort, TestCredentials.CreateSslCredentials(), options);
+            var client = new TestService.TestServiceClient(channel);
+            var first = CallCredentials.FromInterceptor(new AsyncAuthInterceptor((context, metadata) => {
+                metadata.Add("first_authorization", "FIRST_SECRET_TOKEN");
+                return TaskUtils.CompletedTask;
+            }));
+            var second = CallCredentials.FromInterceptor(new AsyncAuthInterceptor((context, metadata) => {
+                metadata.Add("second_authorization", "SECOND_SECRET_TOKEN");
+                return TaskUtils.CompletedTask;
+            }));
+            var third = CallCredentials.FromInterceptor(new AsyncAuthInterceptor((context, metadata) => {
+                metadata.Add("third_authorization", "THIRD_SECRET_TOKEN");
+                return TaskUtils.CompletedTask;
+            }));
+            var call = client.StreamingOutputCall(new StreamingOutputCallRequest{ },
+                new CallOptions(credentials: CallCredentials.Compose(first, second, third)));
+            Assert.IsTrue(await call.ResponseStream.MoveNext());
+            Assert.IsFalse(await call.ResponseStream.MoveNext());
+        }
+
         [Test]
         public void MetadataCredentials_InterceptorLeavesMetadataEmpty()
         {
@@ -125,6 +173,17 @@ namespace Grpc.IntegrationTesting
                 Assert.AreEqual("SECRET_TOKEN", authToken);
                 return Task.FromResult(new SimpleResponse());
             }
+
+            public override async Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
+            {
+                var first = context.RequestHeaders.First((entry) => entry.Key == "first_authorization").Value;
+                Assert.AreEqual("FIRST_SECRET_TOKEN", first);
+                var second = context.RequestHeaders.First((entry) => entry.Key == "second_authorization").Value;
+                Assert.AreEqual("SECOND_SECRET_TOKEN", second);
+                var third = context.RequestHeaders.First((entry) => entry.Key == "third_authorization").Value;
+                Assert.AreEqual("THIRD_SECRET_TOKEN", third);
+                await responseStream.WriteAsync(new StreamingOutputCallResponse());
+            }
         }
     }
 }

src/csharp/ext/grpc_csharp_ext.c

@@ -1023,13 +1023,17 @@ typedef void(GPR_CALLTYPE *grpcsharp_metadata_interceptor_func)(
     grpc_credentials_plugin_metadata_cb cb, void *user_data,
     int32_t is_destroy);
 
-static void grpcsharp_get_metadata_handler(
+static int grpcsharp_get_metadata_handler(
     void *state, grpc_auth_metadata_context context,
-    grpc_credentials_plugin_metadata_cb cb, void *user_data) {
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status,
+    const char **error_details) {
   grpcsharp_metadata_interceptor_func interceptor =
       (grpcsharp_metadata_interceptor_func)(intptr_t)state;
   interceptor(state, context.service_url, context.method_name, cb, user_data,
               0);
+  return 0; /* Asynchronous return. */
 }
 
 static void grpcsharp_metadata_credentials_destroy_handler(void *state) {

src/node/ext/call_credentials.cc

@@ -238,9 +238,12 @@ NAUV_WORK_CB(SendPluginCallback) {
   }
 }
 
-void plugin_get_metadata(void *state, grpc_auth_metadata_context context,
-                         grpc_credentials_plugin_metadata_cb cb,
-                         void *user_data) {
+int plugin_get_metadata(
+    void *state, grpc_auth_metadata_context context,
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status,
+    const char **error_details) {
   plugin_state *p_state = reinterpret_cast<plugin_state *>(state);
   plugin_callback_data *data = new plugin_callback_data;
   data->service_url = context.service_url;
@@ -252,6 +255,7 @@ void plugin_get_metadata(void *state, grpc_auth_metadata_context context,
   uv_mutex_unlock(&p_state->plugin_mutex);
 
   uv_async_send(&p_state->plugin_async);
+  return 0;  // Async processing.
 }
 
 void plugin_uv_close_cb(uv_handle_t *handle) {

src/node/ext/call_credentials.h

@@ -75,9 +75,11 @@ typedef struct plugin_state {
   uv_async_t plugin_async;
 } plugin_state;
 
-void plugin_get_metadata(void *state, grpc_auth_metadata_context context,
-                         grpc_credentials_plugin_metadata_cb cb,
-                         void *user_data);
+int plugin_get_metadata(
+    void *state, grpc_auth_metadata_context context,
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status, const char **error_details);
 
 void plugin_destroy_state(void *state);
 

src/php/ext/grpc/call_credentials.c

@@ -35,6 +35,7 @@
 
 #include <grpc/grpc.h>
 #include <grpc/grpc_security.h>
+#include <grpc/support/string_util.h>
 
 zend_class_entry *grpc_ce_call_credentials;
 #if PHP_MAJOR_VERSION >= 7
@@ -143,9 +144,12 @@ PHP_METHOD(CallCredentials, createFromPlugin) {
 }
 
 /* Callback function for plugin creds API */
-void plugin_get_metadata(void *ptr, grpc_auth_metadata_context context,
-                         grpc_credentials_plugin_metadata_cb cb,
-                         void *user_data) {
+int plugin_get_metadata(
+    void *ptr, grpc_auth_metadata_context context,
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status,
+    const char **error_details) {
   TSRMLS_FETCH();
 
   plugin_state *state = (plugin_state *)ptr;
@@ -175,15 +179,19 @@ void plugin_get_metadata(void *ptr, grpc_auth_metadata_context context,
   /* call the user callback function */
   zend_call_function(state->fci, state->fci_cache TSRMLS_CC);
 
-  grpc_status_code code = GRPC_STATUS_OK;
+  *num_creds_md = 0;
+  *status = GRPC_STATUS_OK;
+  *error_details = NULL;
+
   grpc_metadata_array metadata;
-  bool cleanup = true;
 
   if (retval == NULL || Z_TYPE_P(retval) != IS_ARRAY) {
-    cleanup = false;
-    code = GRPC_STATUS_INVALID_ARGUMENT;
-  } else if (!create_metadata_array(retval, &metadata)) {
-    code = GRPC_STATUS_INVALID_ARGUMENT;
+    *status = GRPC_STATUS_INVALID_ARGUMENT;
+    return true;  // Synchronous return.
+  }
+  if (!create_metadata_array(retval, &metadata)) {
+    *status = GRPC_STATUS_INVALID_ARGUMENT;
+    return true;  // Synchronous return.
   }
 
   if (retval != NULL) {
@@ -197,14 +205,24 @@ void plugin_get_metadata(void *ptr, grpc_auth_metadata_context context,
 #endif
   }
 
-  /* Pass control back to core */
-  cb(user_data, metadata.metadata, metadata.count, code, NULL);
-  if (cleanup) {
-    for (int i = 0; i < metadata.count; i++) {
+  if (metadata.count > GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX) {
+    *status = GRPC_STATUS_INTERNAL;
+    *error_details = gpr_strdup(
+        "PHP plugin credentials returned too many metadata entries");
+    for (size_t i = 0; i < metadata.count; i++) {
+      // TODO(stanleycheung): Why don't we need to unref the key here?
       grpc_slice_unref(metadata.metadata[i].value);
     }
-    grpc_metadata_array_destroy(&metadata);
+  } else {
+    // Return data to core.
+    *num_creds_md = metadata.count;
+    for (size_t i = 0; i < metadata.count; ++i) {
+      creds_md[i] = metadata.metadata[i];
+    }
   }
+
+  grpc_metadata_array_destroy(&metadata);
+  return true;  // Synchronous return.
 }
 
 /* Cleanup function for plugin creds API */

src/php/ext/grpc/call_credentials.h

@@ -65,9 +65,12 @@ typedef struct plugin_state {
 } plugin_state;
 
 /* Callback function for plugin creds API */
-void plugin_get_metadata(void *state, grpc_auth_metadata_context context,
-                         grpc_credentials_plugin_metadata_cb cb,
-                         void *user_data);
+int plugin_get_metadata(
+  void *ptr, grpc_auth_metadata_context context,
+  grpc_credentials_plugin_metadata_cb cb, void *user_data,
+  grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+  size_t *num_creds_md, grpc_status_code *status,
+  const char **error_details);
 
 /* Cleanup function for plugin creds API */
 void plugin_destroy_state(void *ptr);

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

@@ -49,8 +49,11 @@ cdef class AuthMetadataContext:
   cdef grpc_auth_metadata_context context
 
 
-cdef void plugin_get_metadata(
+cdef int plugin_get_metadata(
     void *state, grpc_auth_metadata_context context,
-    grpc_credentials_plugin_metadata_cb cb, void *user_data) with gil
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status,
+    const char **error_details) with gil
 
 cdef void plugin_destroy_c_plugin_state(void *state) with gil

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

@@ -14,6 +14,7 @@
 
 cimport cpython
 
+import threading
 import traceback
 
 
@@ -122,9 +123,12 @@ cdef class AuthMetadataContext:
     grpc_shutdown()
 
 
-cdef void plugin_get_metadata(
+cdef int plugin_get_metadata(
     void *state, grpc_auth_metadata_context context,
-    grpc_credentials_plugin_metadata_cb cb, void *user_data) with gil:
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status,
+    const char **error_details) with gil:
   called_flag = [False]
   def python_callback(
       Metadata metadata, grpc_status_code status,
@@ -134,12 +138,15 @@ cdef void plugin_get_metadata(
   cdef CredentialsMetadataPlugin self = <CredentialsMetadataPlugin>state
   cdef AuthMetadataContext cy_context = AuthMetadataContext()
   cy_context.context = context
-  try:
-    self.plugin_callback(cy_context, python_callback)
-  except Exception as error:
-    if not called_flag[0]:
-      cb(user_data, NULL, 0, StatusCode.unknown,
-         traceback.format_exc().encode())
+  def async_callback():
+    try:
+      self.plugin_callback(cy_context, python_callback)
+    except Exception as error:
+      if not called_flag[0]:
+        cb(user_data, NULL, 0, StatusCode.unknown,
+           traceback.format_exc().encode())
+  threading.Thread(group=None, target=async_callback).start()
+  return 0  # Asynchronous return
 
 cdef void plugin_destroy_c_plugin_state(void *state) with gil:
   cpython.Py_DECREF(<CredentialsMetadataPlugin>state)

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

@@ -375,6 +375,10 @@ cdef extern from "grpc/grpc.h":
 
 cdef extern from "grpc/grpc_security.h":
 
+  # Declare this as an enum, this is the only way to make it a const in
+  # cython
+  enum: GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX
+
   ctypedef enum grpc_ssl_roots_override_result:
     GRPC_SSL_ROOTS_OVERRIDE_OK
     GRPC_SSL_ROOTS_OVERRIDE_FAILED_PERMANENTLY
@@ -462,9 +466,12 @@ cdef extern from "grpc/grpc_security.h":
       grpc_status_code status, const char *error_details)
 
   ctypedef struct grpc_metadata_credentials_plugin:
-    void (*get_metadata)(
+    int (*get_metadata)(
         void *state, grpc_auth_metadata_context context,
-        grpc_credentials_plugin_metadata_cb cb, void *user_data)
+        grpc_credentials_plugin_metadata_cb cb, void *user_data,
+        grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+        size_t *num_creds_md, grpc_status_code *status,
+        const char **error_details)
     void (*destroy)(void *state)
     void *state
     const char *type

src/ruby/ext/grpc/rb_call_credentials.c

@@ -112,9 +112,12 @@ static void grpc_rb_call_credentials_callback_with_gil(void *param) {
   gpr_free(params);
 }
 
-static void grpc_rb_call_credentials_plugin_get_metadata(
+static int grpc_rb_call_credentials_plugin_get_metadata(
     void *state, grpc_auth_metadata_context context,
-    grpc_credentials_plugin_metadata_cb cb, void *user_data) {
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status,
+    const char **error_details) {
   callback_params *params = gpr_malloc(sizeof(callback_params));
   params->get_metadata = (VALUE)state;
   params->context = context;
@@ -123,6 +126,7 @@ static void grpc_rb_call_credentials_plugin_get_metadata(
 
   grpc_rb_event_queue_enqueue(grpc_rb_call_credentials_callback_with_gil,
                               (void *)(params));
+  return 0;  // Async return.
 }
 
 static void grpc_rb_call_credentials_plugin_destroy(void *state) {

test/core/security/credentials_test.c

@@ -1036,39 +1036,46 @@ typedef enum {
 
 static const expected_md plugin_md[] = {{"foo", "bar"}, {"hi", "there"}};
 
-static void plugin_get_metadata_success(void *state,
-                                        grpc_auth_metadata_context context,
-                                        grpc_credentials_plugin_metadata_cb cb,
-                                        void *user_data) {
-  size_t i;
-  grpc_metadata md[GPR_ARRAY_SIZE(plugin_md)];
-  plugin_state *s = (plugin_state *)state;
+static int plugin_get_metadata_success(
+    void *state, grpc_auth_metadata_context context,
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status,
+    const char **error_details) {
   GPR_ASSERT(strcmp(context.service_url, test_service_url) == 0);
   GPR_ASSERT(strcmp(context.method_name, test_method) == 0);
   GPR_ASSERT(context.channel_auth_context == NULL);
   GPR_ASSERT(context.reserved == NULL);
+  GPR_ASSERT(GPR_ARRAY_SIZE(plugin_md) <
+             GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX);
+  plugin_state *s = (plugin_state *)state;
   *s = PLUGIN_GET_METADATA_CALLED_STATE;
-  for (i = 0; i < GPR_ARRAY_SIZE(plugin_md); i++) {
-    memset(&md[i], 0, sizeof(grpc_metadata));
-    md[i].key = grpc_slice_from_copied_string(plugin_md[i].key);
-    md[i].value = grpc_slice_from_copied_string(plugin_md[i].value);
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(plugin_md); ++i) {
+    memset(&creds_md[i], 0, sizeof(grpc_metadata));
+    creds_md[i].key = grpc_slice_from_copied_string(plugin_md[i].key);
+    creds_md[i].value = grpc_slice_from_copied_string(plugin_md[i].value);
   }
-  cb(user_data, md, GPR_ARRAY_SIZE(md), GRPC_STATUS_OK, NULL);
+  *num_creds_md = GPR_ARRAY_SIZE(plugin_md);
+  return true;  // Synchronous return.
 }
 
 static const char *plugin_error_details = "Could not get metadata for plugin.";
 
-static void plugin_get_metadata_failure(void *state,
-                                        grpc_auth_metadata_context context,
-                                        grpc_credentials_plugin_metadata_cb cb,
-                                        void *user_data) {
-  plugin_state *s = (plugin_state *)state;
+static int plugin_get_metadata_failure(
+    void *state, grpc_auth_metadata_context context,
+    grpc_credentials_plugin_metadata_cb cb, void *user_data,
+    grpc_metadata creds_md[GRPC_METADATA_CREDENTIALS_PLUGIN_SYNC_MAX],
+    size_t *num_creds_md, grpc_status_code *status,
+    const char **error_details) {
   GPR_ASSERT(strcmp(context.service_url, test_service_url) == 0);
   GPR_ASSERT(strcmp(context.method_name, test_method) == 0);
   GPR_ASSERT(context.channel_auth_context == NULL);
   GPR_ASSERT(context.reserved == NULL);
+  plugin_state *s = (plugin_state *)state;
   *s = PLUGIN_GET_METADATA_CALLED_STATE;
-  cb(user_data, NULL, 0, GRPC_STATUS_UNAUTHENTICATED, plugin_error_details);
+  *status = GRPC_STATUS_UNAUTHENTICATED;
+  *error_details = gpr_strdup(plugin_error_details);
+  return true;  // Synchronous return.
 }
 
 static void plugin_destroy(void *state) {