stream_op cleanup: security changes

src/core/security/client_auth_filter.c

@@ -59,8 +59,6 @@ typedef struct {
      progress */
   grpc_pollset *pollset;
   grpc_transport_stream_op op;
-  size_t op_md_idx;
-  int sent_initial_metadata;
   gpr_uint8 security_context_set;
   grpc_linked_mdelem md_links[MAX_CREDENTIALS_METADATA_COUNT];
   char *service_url;
@@ -108,9 +106,8 @@ static void on_credentials_metadata(grpc_exec_ctx *exec_ctx, void *user_data,
     return;
   }
   GPR_ASSERT(num_md <= MAX_CREDENTIALS_METADATA_COUNT);
-  GPR_ASSERT(op->send_ops && op->send_ops->nops > calld->op_md_idx &&
-             op->send_ops->ops[calld->op_md_idx].type == GRPC_OP_METADATA);
-  mdb = &op->send_ops->ops[calld->op_md_idx].data.metadata;
+  GPR_ASSERT(op->send_initial_metadata != NULL);
+  mdb = op->send_initial_metadata;
   for (i = 0; i < num_md; i++) {
     grpc_metadata_batch_add_tail(
         mdb, &calld->md_links[i],
@@ -209,7 +206,6 @@ static void auth_start_transport_op(grpc_exec_ctx *exec_ctx,
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
   grpc_linked_mdelem *l;
-  size_t i;
   grpc_client_security_context *sec_ctx = NULL;
 
   if (calld->security_context_set == 0 &&
@@ -228,19 +224,8 @@ static void auth_start_transport_op(grpc_exec_ctx *exec_ctx,
         chand->security_connector->base.auth_context, "client_auth_filter");
   }
 
-  if (op->bind_pollset != NULL) {
-    calld->pollset = op->bind_pollset;
-  }
-
-  if (op->send_ops != NULL && !calld->sent_initial_metadata) {
-    size_t nops = op->send_ops->nops;
-    grpc_stream_op *ops = op->send_ops->ops;
-    for (i = 0; i < nops; i++) {
-      grpc_stream_op *sop = &ops[i];
-      if (sop->type != GRPC_OP_METADATA) continue;
-      calld->op_md_idx = i;
-      calld->sent_initial_metadata = 1;
-      for (l = sop->data.metadata.list.head; l != NULL; l = l->next) {
+  if (op->send_initial_metadata != NULL) {
+    for (l = op->send_initial_metadata->list.head; l != NULL; l = l->next) {
       grpc_mdelem *md = l->md;
       /* Pointer comparison is OK for md_elems created from the same context.
        */
@@ -275,7 +260,6 @@ static void auth_start_transport_op(grpc_exec_ctx *exec_ctx,
     send_security_metadata(exec_ctx, elem, op);
     return; /* early exit */
   }
-  }
 
   /* pass control down the stack */
   grpc_call_next_op(exec_ctx, elem, op);
@@ -283,11 +267,15 @@ static void auth_start_transport_op(grpc_exec_ctx *exec_ctx,
 
 /* Constructor for call_data */
 static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
-                           const void *server_transport_data,
-                           grpc_transport_stream_op *initial_op) {
+                           grpc_call_element_args *args) {
   call_data *calld = elem->call_data;
   memset(calld, 0, sizeof(*calld));
-  GPR_ASSERT(!initial_op || !initial_op->send_ops);
+}
+
+static void set_pollset(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+                        grpc_pollset *pollset) {
+  call_data *calld = elem->call_data;
+  calld->pollset = pollset;
 }
 
 /* Destructor for call_data */
@@ -306,18 +294,17 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
 
 /* Constructor for channel_data */
 static void init_channel_elem(grpc_exec_ctx *exec_ctx,
-                              grpc_channel_element *elem, grpc_channel *master,
-                              const grpc_channel_args *args,
-                              grpc_mdctx *metadata_context, int is_first,
-                              int is_last) {
-  grpc_security_connector *sc = grpc_find_security_connector_in_args(args);
+                              grpc_channel_element *elem,
+                              grpc_channel_element_args *args) {
+  grpc_security_connector *sc =
+      grpc_find_security_connector_in_args(args->channel_args);
   /* grab pointers to our data from the channel element */
   channel_data *chand = elem->channel_data;
 
   /* The first and the last filters tend to be implemented differently to
      handle the case that there's no 'next' filter to call on the up or down
      path */
-  GPR_ASSERT(!is_last);
+  GPR_ASSERT(!args->is_last);
   GPR_ASSERT(sc != NULL);
 
   /* initialize members */
@@ -325,7 +312,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
   chand->security_connector =
       (grpc_channel_security_connector *)GRPC_SECURITY_CONNECTOR_REF(
           sc, "client_auth_filter");
-  chand->md_ctx = metadata_context;
+  chand->md_ctx = args->metadata_context;
   chand->authority_string = grpc_mdstr_from_string(chand->md_ctx, ":authority");
   chand->path_string = grpc_mdstr_from_string(chand->md_ctx, ":path");
   chand->error_msg_key = grpc_mdstr_from_string(chand->md_ctx, "grpc-message");
@@ -356,6 +343,6 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
 
 const grpc_channel_filter grpc_client_auth_filter = {
     auth_start_transport_op, grpc_channel_next_op, sizeof(call_data),
-    init_call_elem,          destroy_call_elem,    sizeof(channel_data),
+    init_call_elem, set_pollset, destroy_call_elem, sizeof(channel_data),
     init_channel_elem, destroy_channel_elem, grpc_call_next_get_peer,
     "client-auth"};

src/core/security/credentials.c

@@ -200,8 +200,7 @@ grpc_arg grpc_server_credentials_to_arg(grpc_server_credentials *p) {
   return arg;
 }
 
-grpc_server_credentials *grpc_server_credentials_from_arg(
-    const grpc_arg *arg) {
+grpc_server_credentials *grpc_server_credentials_from_arg(const grpc_arg *arg) {
   if (strcmp(arg->key, GRPC_SERVER_CREDENTIALS_ARG) != 0) return NULL;
   if (arg->type != GRPC_ARG_POINTER) {
     gpr_log(GPR_ERROR, "Invalid type %d for arg %s", arg->type,

src/core/security/credentials.h

@@ -34,7 +34,7 @@
 #ifndef GRPC_INTERNAL_CORE_SECURITY_CREDENTIALS_H
 #define GRPC_INTERNAL_CORE_SECURITY_CREDENTIALS_H
 
-#include "src/core/transport/stream_op.h"
+#include "src/core/transport/metadata_batch.h"
 #include <grpc/grpc.h>
 #include <grpc/grpc_security.h>
 #include <grpc/support/sync.h>

src/core/security/security_context.c

@@ -324,8 +324,7 @@ grpc_arg grpc_auth_context_to_arg(grpc_auth_context *p) {
   return arg;
 }
 
-grpc_auth_context *grpc_auth_context_from_arg(
-    const grpc_arg *arg) {
+grpc_auth_context *grpc_auth_context_from_arg(const grpc_arg *arg) {
   if (strcmp(arg->key, GRPC_AUTH_CONTEXT_ARG) != 0) return NULL;
   if (arg->type != GRPC_ARG_POINTER) {
     gpr_log(GPR_ERROR, "Invalid type %d for arg %s", arg->type,
@@ -340,8 +339,7 @@ grpc_auth_context *grpc_find_auth_context_in_args(
   size_t i;
   if (args == NULL) return NULL;
   for (i = 0; i < args->num_args; i++) {
-    grpc_auth_context *p =
-        grpc_auth_context_from_arg(&args->args[i]);
+    grpc_auth_context *p = grpc_auth_context_from_arg(&args->args[i]);
     if (p != NULL) return p;
   }
   return NULL;

src/core/security/server_auth_filter.c

@@ -41,8 +41,7 @@
 #include <grpc/support/log.h>
 
 typedef struct call_data {
-  gpr_uint8 got_client_metadata;
-  grpc_stream_op_buffer *recv_ops;
+  grpc_metadata_batch *recv_initial_metadata;
   /* Closure to call when finished with the auth_on_recv hook. */
   grpc_closure *on_done_recv;
   /* Receive closures are chained: we inject this closure as the on_done_recv
@@ -53,7 +52,6 @@ typedef struct call_data {
   grpc_metadata_array md;
   const grpc_metadata *consumed_md;
   size_t num_consumed_md;
-  grpc_stream_op *md_op;
   grpc_auth_context *auth_context;
 } call_data;
 
@@ -128,20 +126,28 @@ static void on_md_processing_done(
   if (status == GRPC_STATUS_OK) {
     calld->consumed_md = consumed_md;
     calld->num_consumed_md = num_consumed_md;
-    grpc_metadata_batch_filter(&calld->md_op->data.metadata, remove_consumed_md,
+    grpc_metadata_batch_filter(calld->recv_initial_metadata, remove_consumed_md,
                                elem);
     grpc_metadata_array_destroy(&calld->md);
     calld->on_done_recv->cb(&exec_ctx, calld->on_done_recv->cb_arg, 1);
   } else {
     gpr_slice message;
+    grpc_transport_stream_op close_op;
+    memset(&close_op, 0, sizeof(close_op));
     grpc_metadata_array_destroy(&calld->md);
     error_details = error_details != NULL
                         ? error_details
                         : "Authentication metadata processing failed.";
     message = gpr_slice_from_copied_string(error_details);
-    grpc_sopb_reset(calld->recv_ops);
-    grpc_transport_stream_op_add_close(&calld->transport_op, status, &message);
-    grpc_call_next_op(&exec_ctx, elem, &calld->transport_op);
+    calld->transport_op.send_initial_metadata = NULL;
+    if (calld->transport_op.send_message != NULL) {
+      grpc_byte_stream_destroy(calld->transport_op.send_message);
+      calld->transport_op.send_message = NULL;
+    }
+    calld->transport_op.send_trailing_metadata = NULL;
+    grpc_transport_stream_op_add_close(&close_op, status, &message);
+    grpc_call_next_op(&exec_ctx, elem, &close_op);
+    calld->on_done_recv->cb(&exec_ctx, calld->on_done_recv->cb_arg, 0);
   }
 
   grpc_exec_ctx_finish(&exec_ctx);
@@ -153,16 +159,8 @@ static void auth_on_recv(grpc_exec_ctx *exec_ctx, void *user_data,
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
   if (success) {
-    size_t i;
-    size_t nops = calld->recv_ops->nops;
-    grpc_stream_op *ops = calld->recv_ops->ops;
-    for (i = 0; i < nops; i++) {
-      grpc_stream_op *op = &ops[i];
-      if (op->type != GRPC_OP_METADATA || calld->got_client_metadata) continue;
-      calld->got_client_metadata = 1;
-      if (chand->creds->processor.process == NULL) continue;
-      calld->md_op = op;
-      calld->md = metadata_batch_to_md_array(&op->data.metadata);
+    if (chand->creds->processor.process != NULL) {
+      calld->md = metadata_batch_to_md_array(calld->recv_initial_metadata);
       chand->creds->processor.process(
           chand->creds->processor.state, calld->auth_context,
           calld->md.metadata, calld->md.count, on_md_processing_done, elem);
@@ -176,11 +174,11 @@ static void set_recv_ops_md_callbacks(grpc_call_element *elem,
                                       grpc_transport_stream_op *op) {
   call_data *calld = elem->call_data;
 
-  if (op->recv_ops && !calld->got_client_metadata) {
+  if (op->recv_initial_metadata != NULL) {
     /* substitute our callback for the higher callback */
-    calld->recv_ops = op->recv_ops;
-    calld->on_done_recv = op->on_done_recv;
-    op->on_done_recv = &calld->auth_on_recv;
+    calld->recv_initial_metadata = op->recv_initial_metadata;
+    calld->on_done_recv = op->on_complete;
+    op->on_complete = &calld->auth_on_recv;
     calld->transport_op = *op;
   }
 }
@@ -199,8 +197,7 @@ static void auth_start_transport_op(grpc_exec_ctx *exec_ctx,
 
 /* Constructor for call_data */
 static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
-                           const void *server_transport_data,
-                           grpc_transport_stream_op *initial_op) {
+                           grpc_call_element_args *args) {
   /* grab pointers to our data from the call element */
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
@@ -210,47 +207,39 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
   memset(calld, 0, sizeof(*calld));
   grpc_closure_init(&calld->auth_on_recv, auth_on_recv, elem);
 
-  GPR_ASSERT(initial_op && initial_op->context != NULL &&
-             initial_op->context[GRPC_CONTEXT_SECURITY].value == NULL);
-
-  /* Create a security context for the call and reference the auth context from
-     the channel. */
-  if (initial_op->context[GRPC_CONTEXT_SECURITY].value != NULL) {
-    initial_op->context[GRPC_CONTEXT_SECURITY].destroy(
-        initial_op->context[GRPC_CONTEXT_SECURITY].value);
+  if (args->context[GRPC_CONTEXT_SECURITY].value != NULL) {
+    args->context[GRPC_CONTEXT_SECURITY].destroy(
+        args->context[GRPC_CONTEXT_SECURITY].value);
   }
+
   server_ctx = grpc_server_security_context_create();
-  server_ctx->auth_context =
-      grpc_auth_context_create(chand->auth_context);
-  server_ctx->auth_context->pollset = initial_op->bind_pollset;
-  initial_op->context[GRPC_CONTEXT_SECURITY].value = server_ctx;
-  initial_op->context[GRPC_CONTEXT_SECURITY].destroy =
-      grpc_server_security_context_destroy;
+  server_ctx->auth_context = grpc_auth_context_create(chand->auth_context);
   calld->auth_context = server_ctx->auth_context;
 
-  /* Set the metadata callbacks. */
-  set_recv_ops_md_callbacks(elem, initial_op);
+  args->context[GRPC_CONTEXT_SECURITY].value = server_ctx;
+  args->context[GRPC_CONTEXT_SECURITY].destroy =
+      grpc_server_security_context_destroy;
 }
 
+static void set_pollset(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+                        grpc_pollset *pollset) {}
+
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
                               grpc_call_element *elem) {}
 
 /* Constructor for channel_data */
 static void init_channel_elem(grpc_exec_ctx *exec_ctx,
-                              grpc_channel_element *elem, grpc_channel *master,
-                              const grpc_channel_args *args, grpc_mdctx *mdctx,
-                              int is_first, int is_last) {
-  grpc_auth_context *auth_context = grpc_find_auth_context_in_args(args);
-  grpc_server_credentials *creds = grpc_find_server_credentials_in_args(args);
+                              grpc_channel_element *elem,
+                              grpc_channel_element_args *args) {
+  grpc_auth_context *auth_context =
+      grpc_find_auth_context_in_args(args->channel_args);
+  grpc_server_credentials *creds =
+      grpc_find_server_credentials_in_args(args->channel_args);
   /* grab pointers to our data from the channel element */
   channel_data *chand = elem->channel_data;
 
-  /* The first and the last filters tend to be implemented differently to
-     handle the case that there's no 'next' filter to call on the up or down
-     path */
-  GPR_ASSERT(!is_first);
-  GPR_ASSERT(!is_last);
+  GPR_ASSERT(!args->is_last);
   GPR_ASSERT(auth_context != NULL);
   GPR_ASSERT(creds != NULL);
 
@@ -258,7 +247,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
   chand->auth_context =
       GRPC_AUTH_CONTEXT_REF(auth_context, "server_auth_filter");
   chand->creds = grpc_server_credentials_ref(creds);
-  chand->mdctx = mdctx;
+  chand->mdctx = args->metadata_context;
 }
 
 /* Destructor for channel data */
@@ -272,6 +261,6 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
 
 const grpc_channel_filter grpc_server_auth_filter = {
     auth_start_transport_op, grpc_channel_next_op, sizeof(call_data),
-    init_call_elem,          destroy_call_elem,    sizeof(channel_data),
+    init_call_elem, set_pollset, destroy_call_elem, sizeof(channel_data),
     init_channel_elem, destroy_channel_elem, grpc_call_next_get_peer,
     "server-auth"};

src/core/security/server_secure_chttp2.c

@@ -94,8 +94,7 @@ static void setup_transport(grpc_exec_ctx *exec_ctx, void *statep,
   grpc_channel_args *args_copy;
   grpc_arg args_to_add[2];
   args_to_add[0] = grpc_server_credentials_to_arg(state->creds);
-  args_to_add[1] =
-      grpc_auth_context_to_arg(state->sc->auth_context);
+  args_to_add[1] = grpc_auth_context_to_arg(state->sc->auth_context);
   args_copy = grpc_channel_args_copy_and_add(
       grpc_server_get_channel_args(state->server), args_to_add,
       GPR_ARRAY_SIZE(args_to_add));