[VOL-4291] Rw-core updates for gRPC migration
Change-Id: I8d5a554409115b29318089671ca4e1ab3fa98810
diff --git a/vendor/go.etcd.io/etcd/embed/serve.go b/vendor/go.etcd.io/etcd/embed/serve.go
index a3b20c4..62b8b57 100644
--- a/vendor/go.etcd.io/etcd/embed/serve.go
+++ b/vendor/go.etcd.io/etcd/embed/serve.go
@@ -16,41 +16,36 @@
import (
"context"
- "fmt"
"io/ioutil"
defaultLog "log"
"net"
"net/http"
"strings"
- "go.etcd.io/etcd/clientv3/credentials"
- "go.etcd.io/etcd/etcdserver"
- "go.etcd.io/etcd/etcdserver/api/v3client"
- "go.etcd.io/etcd/etcdserver/api/v3election"
- "go.etcd.io/etcd/etcdserver/api/v3election/v3electionpb"
- v3electiongw "go.etcd.io/etcd/etcdserver/api/v3election/v3electionpb/gw"
- "go.etcd.io/etcd/etcdserver/api/v3lock"
- "go.etcd.io/etcd/etcdserver/api/v3lock/v3lockpb"
- v3lockgw "go.etcd.io/etcd/etcdserver/api/v3lock/v3lockpb/gw"
- "go.etcd.io/etcd/etcdserver/api/v3rpc"
- etcdservergw "go.etcd.io/etcd/etcdserver/etcdserverpb/gw"
- "go.etcd.io/etcd/pkg/debugutil"
- "go.etcd.io/etcd/pkg/httputil"
- "go.etcd.io/etcd/pkg/transport"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v3client"
+ "github.com/coreos/etcd/etcdserver/api/v3election"
+ "github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb"
+ v3electiongw "github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb/gw"
+ "github.com/coreos/etcd/etcdserver/api/v3lock"
+ "github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb"
+ v3lockgw "github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb/gw"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc"
+ etcdservergw "github.com/coreos/etcd/etcdserver/etcdserverpb/gw"
+ "github.com/coreos/etcd/pkg/debugutil"
+ "github.com/coreos/etcd/pkg/transport"
gw "github.com/grpc-ecosystem/grpc-gateway/runtime"
"github.com/soheilhy/cmux"
"github.com/tmc/grpc-websocket-proxy/wsproxy"
- "go.uber.org/zap"
"golang.org/x/net/trace"
"google.golang.org/grpc"
+ "google.golang.org/grpc/credentials"
)
type serveCtx struct {
- lg *zap.Logger
l net.Listener
addr string
- network string
secure bool
insecure bool
@@ -68,14 +63,10 @@
http *http.Server
}
-func newServeCtx(lg *zap.Logger) *serveCtx {
+func newServeCtx() *serveCtx {
ctx, cancel := context.WithCancel(context.Background())
- return &serveCtx{
- lg: lg,
- ctx: ctx,
- cancel: cancel,
- userHandlers: make(map[string]http.Handler),
- serversC: make(chan *servers, 2), // in case sctx.insecure,sctx.secure true
+ return &serveCtx{ctx: ctx, cancel: cancel, userHandlers: make(map[string]http.Handler),
+ serversC: make(chan *servers, 2), // in case sctx.insecure,sctx.secure true
}
}
@@ -90,10 +81,7 @@
gopts ...grpc.ServerOption) (err error) {
logger := defaultLog.New(ioutil.Discard, "etcdhttp", 0)
<-s.ReadyNotify()
-
- if sctx.lg == nil {
- plog.Info("ready to serve client requests")
- }
+ plog.Info("ready to serve client requests")
m := cmux.New(sctx.l)
v3c := v3client.New(s)
@@ -118,31 +106,22 @@
go func() { errHandler(gs.Serve(grpcl)) }()
var gwmux *gw.ServeMux
- if s.Cfg.EnableGRPCGateway {
- gwmux, err = sctx.registerGateway([]grpc.DialOption{grpc.WithInsecure()})
- if err != nil {
- return err
- }
+ gwmux, err = sctx.registerGateway([]grpc.DialOption{grpc.WithInsecure()})
+ if err != nil {
+ return err
}
httpmux := sctx.createMux(gwmux, handler)
srvhttp := &http.Server{
- Handler: createAccessController(sctx.lg, s, httpmux),
+ Handler: wrapMux(httpmux),
ErrorLog: logger, // do not log user error
}
httpl := m.Match(cmux.HTTP1())
go func() { errHandler(srvhttp.Serve(httpl)) }()
sctx.serversC <- &servers{grpc: gs, http: srvhttp}
- if sctx.lg != nil {
- sctx.lg.Info(
- "serving client traffic insecurely; this is strongly discouraged!",
- zap.String("address", sctx.l.Addr().String()),
- )
- } else {
- plog.Noticef("serving insecure client requests on %s, this is strongly discouraged!", sctx.l.Addr().String())
- }
+ plog.Noticef("serving insecure client requests on %s, this is strongly discouraged!", sctx.l.Addr().String())
}
if sctx.secure {
@@ -158,17 +137,15 @@
}
handler = grpcHandlerFunc(gs, handler)
+ dtls := tlscfg.Clone()
+ // trust local server
+ dtls.InsecureSkipVerify = true
+ creds := credentials.NewTLS(dtls)
+ opts := []grpc.DialOption{grpc.WithTransportCredentials(creds)}
var gwmux *gw.ServeMux
- if s.Cfg.EnableGRPCGateway {
- dtls := tlscfg.Clone()
- // trust local server
- dtls.InsecureSkipVerify = true
- bundle := credentials.NewBundle(credentials.Config{TLSConfig: dtls})
- opts := []grpc.DialOption{grpc.WithTransportCredentials(bundle.TransportCredentials())}
- gwmux, err = sctx.registerGateway(opts)
- if err != nil {
- return err
- }
+ gwmux, err = sctx.registerGateway(opts)
+ if err != nil {
+ return err
}
var tlsl net.Listener
@@ -180,21 +157,14 @@
httpmux := sctx.createMux(gwmux, handler)
srv := &http.Server{
- Handler: createAccessController(sctx.lg, s, httpmux),
+ Handler: wrapMux(httpmux),
TLSConfig: tlscfg,
ErrorLog: logger, // do not log user error
}
go func() { errHandler(srv.Serve(tlsl)) }()
sctx.serversC <- &servers{secure: true, grpc: gs, http: srv}
- if sctx.lg != nil {
- sctx.lg.Info(
- "serving client traffic securely",
- zap.String("address", sctx.l.Addr().String()),
- )
- } else {
- plog.Infof("serving client requests on %s", sctx.l.Addr().String())
- }
+ plog.Infof("serving client requests on %s", sctx.l.Addr().String())
}
close(sctx.serversC)
@@ -222,14 +192,7 @@
func (sctx *serveCtx) registerGateway(opts []grpc.DialOption) (*gw.ServeMux, error) {
ctx := sctx.ctx
-
- addr := sctx.addr
- if network := sctx.network; network == "unix" {
- // explicitly define unix network for gRPC socket support
- addr = fmt.Sprintf("%s://%s", network, addr)
- }
-
- conn, err := grpc.DialContext(ctx, addr, opts...)
+ conn, err := grpc.DialContext(ctx, sctx.addr, opts...)
if err != nil {
return nil, err
}
@@ -253,15 +216,7 @@
go func() {
<-ctx.Done()
if cerr := conn.Close(); cerr != nil {
- if sctx.lg != nil {
- sctx.lg.Warn(
- "failed to close connection",
- zap.String("address", sctx.l.Addr().String()),
- zap.Error(cerr),
- )
- } else {
- plog.Warningf("failed to close conn to %s: %v", sctx.l.Addr().String(), cerr)
- }
+ plog.Warningf("failed to close conn to %s: %v", sctx.l.Addr().String(), cerr)
}
}()
@@ -274,148 +229,43 @@
httpmux.Handle(path, h)
}
- if gwmux != nil {
- httpmux.Handle(
- "/v3/",
- wsproxy.WebsocketProxy(
- gwmux,
- wsproxy.WithRequestMutator(
- // Default to the POST method for streams
- func(_ *http.Request, outgoing *http.Request) *http.Request {
- outgoing.Method = "POST"
- return outgoing
- },
- ),
+ httpmux.Handle(
+ "/v3beta/",
+ wsproxy.WebsocketProxy(
+ gwmux,
+ wsproxy.WithRequestMutator(
+ // Default to the POST method for streams
+ func(incoming *http.Request, outgoing *http.Request) *http.Request {
+ outgoing.Method = "POST"
+ return outgoing
+ },
),
- )
- }
+ ),
+ )
if handler != nil {
httpmux.Handle("/", handler)
}
return httpmux
}
-// createAccessController wraps HTTP multiplexer:
-// - mutate gRPC gateway request paths
-// - check hostname whitelist
-// client HTTP requests goes here first
-func createAccessController(lg *zap.Logger, s *etcdserver.EtcdServer, mux *http.ServeMux) http.Handler {
- return &accessController{lg: lg, s: s, mux: mux}
-}
+// wraps HTTP multiplexer to mute requests to /v3alpha
+// TODO: deprecate this in 3.4 release
+func wrapMux(mux *http.ServeMux) http.Handler { return &v3alphaMutator{mux: mux} }
-type accessController struct {
- lg *zap.Logger
- s *etcdserver.EtcdServer
+type v3alphaMutator struct {
mux *http.ServeMux
}
-func (ac *accessController) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
- // redirect for backward compatibilities
- if req != nil && req.URL != nil && strings.HasPrefix(req.URL.Path, "/v3beta/") {
- req.URL.Path = strings.Replace(req.URL.Path, "/v3beta/", "/v3/", 1)
+func (m *v3alphaMutator) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
+ if req != nil && req.URL != nil && strings.HasPrefix(req.URL.Path, "/v3alpha/") {
+ req.URL.Path = strings.Replace(req.URL.Path, "/v3alpha/", "/v3beta/", 1)
}
-
- if req.TLS == nil { // check origin if client connection is not secure
- host := httputil.GetHostname(req)
- if !ac.s.AccessController.IsHostWhitelisted(host) {
- if ac.lg != nil {
- ac.lg.Warn(
- "rejecting HTTP request to prevent DNS rebinding attacks",
- zap.String("host", host),
- )
- } else {
- plog.Warningf("rejecting HTTP request from %q to prevent DNS rebinding attacks", host)
- }
- // TODO: use Go's "http.StatusMisdirectedRequest" (421)
- // https://github.com/golang/go/commit/4b8a7eafef039af1834ef9bfa879257c4a72b7b5
- http.Error(rw, errCVE20185702(host), 421)
- return
- }
- } else if ac.s.Cfg.ClientCertAuthEnabled && ac.s.Cfg.EnableGRPCGateway &&
- ac.s.AuthStore().IsAuthEnabled() && strings.HasPrefix(req.URL.Path, "/v3/") {
- for _, chains := range req.TLS.VerifiedChains {
- if len(chains) < 1 {
- continue
- }
- if len(chains[0].Subject.CommonName) != 0 {
- http.Error(rw, "CommonName of client sending a request against gateway will be ignored and not used as expected", 400)
- return
- }
- }
- }
-
- // Write CORS header.
- if ac.s.AccessController.OriginAllowed("*") {
- addCORSHeader(rw, "*")
- } else if origin := req.Header.Get("Origin"); ac.s.OriginAllowed(origin) {
- addCORSHeader(rw, origin)
- }
-
- if req.Method == "OPTIONS" {
- rw.WriteHeader(http.StatusOK)
- return
- }
-
- ac.mux.ServeHTTP(rw, req)
-}
-
-// addCORSHeader adds the correct cors headers given an origin
-func addCORSHeader(w http.ResponseWriter, origin string) {
- w.Header().Add("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE")
- w.Header().Add("Access-Control-Allow-Origin", origin)
- w.Header().Add("Access-Control-Allow-Headers", "accept, content-type, authorization")
-}
-
-// https://github.com/transmission/transmission/pull/468
-func errCVE20185702(host string) string {
- return fmt.Sprintf(`
-etcd received your request, but the Host header was unrecognized.
-
-To fix this, choose one of the following options:
-- Enable TLS, then any HTTPS request will be allowed.
-- Add the hostname you want to use to the whitelist in settings.
- - e.g. etcd --host-whitelist %q
-
-This requirement has been added to help prevent "DNS Rebinding" attacks (CVE-2018-5702).
-`, host)
-}
-
-// WrapCORS wraps existing handler with CORS.
-// TODO: deprecate this after v2 proxy deprecate
-func WrapCORS(cors map[string]struct{}, h http.Handler) http.Handler {
- return &corsHandler{
- ac: &etcdserver.AccessController{CORS: cors},
- h: h,
- }
-}
-
-type corsHandler struct {
- ac *etcdserver.AccessController
- h http.Handler
-}
-
-func (ch *corsHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
- if ch.ac.OriginAllowed("*") {
- addCORSHeader(rw, "*")
- } else if origin := req.Header.Get("Origin"); ch.ac.OriginAllowed(origin) {
- addCORSHeader(rw, origin)
- }
-
- if req.Method == "OPTIONS" {
- rw.WriteHeader(http.StatusOK)
- return
- }
-
- ch.h.ServeHTTP(rw, req)
+ m.mux.ServeHTTP(rw, req)
}
func (sctx *serveCtx) registerUserHandler(s string, h http.Handler) {
if sctx.userHandlers[s] != nil {
- if sctx.lg != nil {
- sctx.lg.Warn("path is already registered by user handler", zap.String("path", s))
- } else {
- plog.Warningf("path %s already registered by user handler", s)
- }
+ plog.Warningf("path %s already registered by user handler", s)
return
}
sctx.userHandlers[s] = h