Merge github.com:grpc/grpc into churn-churn-churn-the-api-gently-down-the-stream

Conflicts:
	src/cpp/server/server.cc

9 files changed, 81 insertions, 16 deletions

diff --git a/src/core/iomgr/socket_utils_common_posix.c b/src/core/iomgr/socket_utils_common_posix.c
index 3c8cafa315..a9af594700 100644
--- a/src/core/iomgr/socket_utils_common_posix.c
+++ b/src/core/iomgr/socket_utils_common_posix.c
@@ -76,6 +76,19 @@ int grpc_set_socket_nonblocking(int fd, int non_blocking) {
   return 1;
 }
 
+int grpc_set_socket_no_sigpipe_if_possible(int fd) {
+#ifdef GPR_HAVE_SO_NOSIGPIPE
+  int val = 1;
+  int newval;
+  socklen_t intlen = sizeof(newval);
+  return 0 == setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &val, sizeof(val)) &&
+         0 == getsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &newval, &intlen) &&
+         (newval != 0) == val;
+#else
+  return 1;
+#endif
+}
+
 /* set a socket to close on exec */
 int grpc_set_socket_cloexec(int fd, int close_on_exec) {
   int oldflags = fcntl(fd, F_GETFD, 0);
diff --git a/src/core/iomgr/socket_utils_posix.h b/src/core/iomgr/socket_utils_posix.h
index c161082afc..d2a315b462 100644
--- a/src/core/iomgr/socket_utils_posix.h
+++ b/src/core/iomgr/socket_utils_posix.h
@@ -63,6 +63,11 @@ int grpc_set_socket_low_latency(int fd, int low_latency);
    state to library users, we turn off IPv6 sockets. */
 int grpc_ipv6_loopback_available(void);
 
+/* Tries to set SO_NOSIGPIPE if available on this platform.
+   Returns 1 on success, 0 on failure.
+   If SO_NO_SIGPIPE is not available, returns 1. */
+int grpc_set_socket_no_sigpipe_if_possible(int fd);
+
 /* An enum to keep track of IPv4/IPv6 socket modes.
 
    Currently, this information is only used when a socket is first created, but
diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c
index e20cc3d1b2..2401fe00e4 100644
--- a/src/core/iomgr/tcp_client_posix.c
+++ b/src/core/iomgr/tcp_client_posix.c
@@ -69,7 +69,8 @@ static int prepare_socket(const struct sockaddr *addr, int fd) {
   }
 
   if (!grpc_set_socket_nonblocking(fd, 1) || !grpc_set_socket_cloexec(fd, 1) ||
-      (addr->sa_family != AF_UNIX && !grpc_set_socket_low_latency(fd, 1))) {
+      (addr->sa_family != AF_UNIX && !grpc_set_socket_low_latency(fd, 1)) ||
+      !grpc_set_socket_no_sigpipe_if_possible(fd)) {
     gpr_log(GPR_ERROR, "Unable to configure socket %d: %s", fd,
             strerror(errno));
     goto error;
diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c
index 06725fbc89..f7dae5f86c 100644
--- a/src/core/iomgr/tcp_posix.c
+++ b/src/core/iomgr/tcp_posix.c
@@ -53,6 +53,12 @@
 #include <grpc/support/sync.h>
 #include <grpc/support/time.h>
 
+#ifdef GPR_HAVE_MSG_NOSIGNAL
+#define SENDMSG_FLAGS MSG_NOSIGNAL
+#else
+#define SENDMSG_FLAGS 0
+#endif
+
 /* Holds a slice array and associated state. */
 typedef struct grpc_tcp_slice_state {
   gpr_slice *slices;       /* Array of slices */
@@ -461,7 +467,7 @@ static grpc_endpoint_write_status grpc_tcp_flush(grpc_tcp *tcp) {
     GRPC_TIMER_BEGIN(GRPC_PTAG_SENDMSG, 0);
     do {
       /* TODO(klempner): Cork if this is a partial write */
-      sent_length = sendmsg(tcp->fd, &msg, 0);
+      sent_length = sendmsg(tcp->fd, &msg, SENDMSG_FLAGS);
     } while (sent_length < 0 && errno == EINTR);
     GRPC_TIMER_END(GRPC_PTAG_SENDMSG, 0);
 
diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c
index 7e31f2d7a5..d1cd8a769c 100644
--- a/src/core/iomgr/tcp_server_posix.c
+++ b/src/core/iomgr/tcp_server_posix.c
@@ -235,7 +235,8 @@ static int prepare_socket(int fd, const struct sockaddr *addr, int addr_len) {
 
   if (!grpc_set_socket_nonblocking(fd, 1) || !grpc_set_socket_cloexec(fd, 1) ||
       (addr->sa_family != AF_UNIX && (!grpc_set_socket_low_latency(fd, 1) ||
-                                      !grpc_set_socket_reuse_addr(fd, 1)))) {
+                                      !grpc_set_socket_reuse_addr(fd, 1))) ||
+      !grpc_set_socket_no_sigpipe_if_possible(fd)) {
     gpr_log(GPR_ERROR, "Unable to configure socket %d: %s", fd,
             strerror(errno));
     goto error;
@@ -296,6 +297,8 @@ static void on_read(void *arg, int success) {
       }
     }
 
+    grpc_set_socket_no_sigpipe_if_possible(fd);
+
     sp->server->cb(
         sp->server->cb_arg,
         grpc_tcp_create(grpc_fd_create(fd), GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 7ab9142947..9ee91785e8 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -739,14 +739,9 @@ static void call_on_done_recv(void *pc, int success) {
   GRPC_TIMER_BEGIN(GRPC_PTAG_CALL_ON_DONE_RECV, 0);
 }
 
-static grpc_mdelem_list chain_metadata_from_app(grpc_call *call, size_t count,
-                                                grpc_metadata *metadata) {
+static int prepare_application_metadata(grpc_call *call, size_t count,
+                                        grpc_metadata *metadata) {
   size_t i;
-  grpc_mdelem_list out;
-  if (count == 0) {
-    out.head = out.tail = NULL;
-    return out;
-  }
   for (i = 0; i < count; i++) {
     grpc_metadata *md = &metadata[i];
     grpc_metadata *next_md = (i == count - 1) ? NULL : &metadata[i + 1];
@@ -756,9 +751,27 @@ static grpc_mdelem_list chain_metadata_from_app(grpc_call *call, size_t count,
     l->md = grpc_mdelem_from_string_and_buffer(call->metadata_context, md->key,
                                                (const gpr_uint8 *)md->value,
                                                md->value_length);
+    if (!grpc_mdstr_is_legal_header(l->md->key)) {
+      gpr_log(GPR_ERROR, "attempt to send invalid metadata key");
+      return 0;
+    } else if (!grpc_mdstr_is_bin_suffixed(l->md->key) &&
+               !grpc_mdstr_is_legal_header(l->md->value)) {
+      gpr_log(GPR_ERROR, "attempt to send invalid metadata value");
+      return 0;
+    }
     l->next = next_md ? (grpc_linked_mdelem *)&next_md->internal_data : NULL;
     l->prev = prev_md ? (grpc_linked_mdelem *)&prev_md->internal_data : NULL;
   }
+  return 1;
+}
+
+static grpc_mdelem_list chain_metadata_from_app(grpc_call *call, size_t count,
+                                                grpc_metadata *metadata) {
+  grpc_mdelem_list out;
+  if (count == 0) {
+    out.head = out.tail = NULL;
+    return out;
+  }
   out.head = (grpc_linked_mdelem *)&(metadata[0].internal_data);
   out.tail = (grpc_linked_mdelem *)&(metadata[count - 1].internal_data);
   return out;
@@ -954,8 +967,16 @@ static grpc_call_error start_ioreq(grpc_call *call, const grpc_ioreq *reqs,
     } else if (call->request_set[op] == REQSET_DONE) {
       return start_ioreq_error(call, have_ops, GRPC_CALL_ERROR_ALREADY_INVOKED);
     }
-    have_ops |= 1u << op;
     data = reqs[i].data;
+    if (op == GRPC_IOREQ_SEND_INITIAL_METADATA ||
+        op == GRPC_IOREQ_SEND_TRAILING_METADATA) {
+      if (!prepare_application_metadata(call, data.send_metadata.count,
+                                        data.send_metadata.metadata)) {
+        return start_ioreq_error(call, have_ops,
+                                 GRPC_CALL_ERROR_INVALID_METADATA);
+      }
+    }
+    have_ops |= 1u << op;
 
     call->request_data[op] = data;
     call->request_set[op] = set;
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 885838ec5d..ea24796f35 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -824,12 +824,9 @@ static void unlock(transport *t) {
 
   /* gather any callbacks that need to be made */
   if (!t->calling_back) {
-    perform_callbacks = prepare_callbacks(t);
-    if (perform_callbacks) {
-      t->calling_back = 1;
-    }
+    t->calling_back = perform_callbacks = prepare_callbacks(t);
     if (cb) {
-      if (t->error_state == ERROR_STATE_SEEN && !t->writing && !t->calling_back) {
+      if (t->error_state == ERROR_STATE_SEEN && !t->writing) {
         call_closed = 1;
         t->calling_back = 1;
         t->cb = NULL; /* no more callbacks */
diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c
index 74e94b2c24..c80d67823f 100644
--- a/src/core/transport/metadata.c
+++ b/src/core/transport/metadata.c
@@ -569,3 +569,19 @@ void grpc_mdctx_locked_mdelem_unref(grpc_mdctx *ctx, grpc_mdelem *gmd) {
 }
 
 void grpc_mdctx_unlock(grpc_mdctx *ctx) { unlock(ctx); }
+
+int grpc_mdstr_is_legal_header(grpc_mdstr *s) {
+  /* TODO(ctiller): consider caching this, or computing it on construction */
+  const gpr_uint8 *p = GPR_SLICE_START_PTR(s->slice);
+  const gpr_uint8 *e = GPR_SLICE_END_PTR(s->slice);
+  for (; p != e; p++) {
+    if (*p < 32 || *p > 126) return 0;
+  }
+  return 1;
+}
+
+int grpc_mdstr_is_bin_suffixed(grpc_mdstr *s) {
+  /* TODO(ctiller): consider caching this */
+  return grpc_is_binary_header((const char *)GPR_SLICE_START_PTR(s->slice),
+                               GPR_SLICE_LENGTH(s->slice));
+}
diff --git a/src/core/transport/metadata.h b/src/core/transport/metadata.h
index 21b8ae2b78..e7508718f5 100644
--- a/src/core/transport/metadata.h
+++ b/src/core/transport/metadata.h
@@ -135,6 +135,9 @@ void grpc_mdelem_unref(grpc_mdelem *md);
    Does not promise that the returned string has no embedded nulls however. */
 const char *grpc_mdstr_as_c_string(grpc_mdstr *s);
 
+int grpc_mdstr_is_legal_header(grpc_mdstr *s);
+int grpc_mdstr_is_bin_suffixed(grpc_mdstr *s);
+
 /* Batch mode metadata functions.
    These API's have equivalents above, but allow taking the mdctx just once,
    performing a bunch of work, and then leaving the mdctx. */