package api import ( "bytes" "context" "crypto/tls" "encoding/json" "fmt" "io" "io/ioutil" "log" "net" "net/http" "net/url" "os" "strconv" "strings" "time" "github.com/hashicorp/go-cleanhttp" "github.com/hashicorp/go-rootcerts" ) const ( // HTTPAddrEnvName defines an environment variable name which sets // the HTTP address if there is no -http-addr specified. HTTPAddrEnvName = "CONSUL_HTTP_ADDR" // HTTPTokenEnvName defines an environment variable name which sets // the HTTP token. HTTPTokenEnvName = "CONSUL_HTTP_TOKEN" // HTTPAuthEnvName defines an environment variable name which sets // the HTTP authentication header. HTTPAuthEnvName = "CONSUL_HTTP_AUTH" // HTTPSSLEnvName defines an environment variable name which sets // whether or not to use HTTPS. HTTPSSLEnvName = "CONSUL_HTTP_SSL" // HTTPCAFile defines an environment variable name which sets the // CA file to use for talking to Consul over TLS. HTTPCAFile = "CONSUL_CACERT" // HTTPCAPath defines an environment variable name which sets the // path to a directory of CA certs to use for talking to Consul over TLS. HTTPCAPath = "CONSUL_CAPATH" // HTTPClientCert defines an environment variable name which sets the // client cert file to use for talking to Consul over TLS. HTTPClientCert = "CONSUL_CLIENT_CERT" // HTTPClientKey defines an environment variable name which sets the // client key file to use for talking to Consul over TLS. HTTPClientKey = "CONSUL_CLIENT_KEY" // HTTPTLSServerName defines an environment variable name which sets the // server name to use as the SNI host when connecting via TLS HTTPTLSServerName = "CONSUL_TLS_SERVER_NAME" // HTTPSSLVerifyEnvName defines an environment variable name which sets // whether or not to disable certificate checking. HTTPSSLVerifyEnvName = "CONSUL_HTTP_SSL_VERIFY" ) // QueryOptions are used to parameterize a query type QueryOptions struct { // Providing a datacenter overwrites the DC provided // by the Config Datacenter string // AllowStale allows any Consul server (non-leader) to service // a read. This allows for lower latency and higher throughput AllowStale bool // RequireConsistent forces the read to be fully consistent. // This is more expensive but prevents ever performing a stale // read. RequireConsistent bool // WaitIndex is used to enable a blocking query. Waits // until the timeout or the next index is reached WaitIndex uint64 // WaitHash is used by some endpoints instead of WaitIndex to perform blocking // on state based on a hash of the response rather than a monotonic index. // This is required when the state being blocked on is not stored in Raft, for // example agent-local proxy configuration. WaitHash string // WaitTime is used to bound the duration of a wait. // Defaults to that of the Config, but can be overridden. WaitTime time.Duration // Token is used to provide a per-request ACL token // which overrides the agent's default token. Token string // Near is used to provide a node name that will sort the results // in ascending order based on the estimated round trip time from // that node. Setting this to "_agent" will use the agent's node // for the sort. Near string // NodeMeta is used to filter results by nodes with the given // metadata key/value pairs. Currently, only one key/value pair can // be provided for filtering. NodeMeta map[string]string // RelayFactor is used in keyring operations to cause responses to be // relayed back to the sender through N other random nodes. Must be // a value from 0 to 5 (inclusive). RelayFactor uint8 // Connect filters prepared query execution to only include Connect-capable // services. This currently affects prepared query execution. Connect bool // ctx is an optional context pass through to the underlying HTTP // request layer. Use Context() and WithContext() to manage this. ctx context.Context } func (o *QueryOptions) Context() context.Context { if o != nil && o.ctx != nil { return o.ctx } return context.Background() } func (o *QueryOptions) WithContext(ctx context.Context) *QueryOptions { o2 := new(QueryOptions) if o != nil { *o2 = *o } o2.ctx = ctx return o2 } // WriteOptions are used to parameterize a write type WriteOptions struct { // Providing a datacenter overwrites the DC provided // by the Config Datacenter string // Token is used to provide a per-request ACL token // which overrides the agent's default token. Token string // RelayFactor is used in keyring operations to cause responses to be // relayed back to the sender through N other random nodes. Must be // a value from 0 to 5 (inclusive). RelayFactor uint8 // ctx is an optional context pass through to the underlying HTTP // request layer. Use Context() and WithContext() to manage this. ctx context.Context } func (o *WriteOptions) Context() context.Context { if o != nil && o.ctx != nil { return o.ctx } return context.Background() } func (o *WriteOptions) WithContext(ctx context.Context) *WriteOptions { o2 := new(WriteOptions) if o != nil { *o2 = *o } o2.ctx = ctx return o2 } // QueryMeta is used to return meta data about a query type QueryMeta struct { // LastIndex. This can be used as a WaitIndex to perform // a blocking query LastIndex uint64 // LastContentHash. This can be used as a WaitHash to perform a blocking query // for endpoints that support hash-based blocking. Endpoints that do not // support it will return an empty hash. LastContentHash string // Time of last contact from the leader for the // server servicing the request LastContact time.Duration // Is there a known leader KnownLeader bool // How long did the request take RequestTime time.Duration // Is address translation enabled for HTTP responses on this agent AddressTranslationEnabled bool } // WriteMeta is used to return meta data about a write type WriteMeta struct { // How long did the request take RequestTime time.Duration } // HttpBasicAuth is used to authenticate http client with HTTP Basic Authentication type HttpBasicAuth struct { // Username to use for HTTP Basic Authentication Username string // Password to use for HTTP Basic Authentication Password string } // Config is used to configure the creation of a client type Config struct { // Address is the address of the Consul server Address string // Scheme is the URI scheme for the Consul server Scheme string // Datacenter to use. If not provided, the default agent datacenter is used. Datacenter string // Transport is the Transport to use for the http client. Transport *http.Transport // HttpClient is the client to use. Default will be // used if not provided. HttpClient *http.Client // HttpAuth is the auth info to use for http access. HttpAuth *HttpBasicAuth // WaitTime limits how long a Watch will block. If not provided, // the agent default values will be used. WaitTime time.Duration // Token is used to provide a per-request ACL token // which overrides the agent's default token. Token string TLSConfig TLSConfig } // TLSConfig is used to generate a TLSClientConfig that's useful for talking to // Consul using TLS. type TLSConfig struct { // Address is the optional address of the Consul server. The port, if any // will be removed from here and this will be set to the ServerName of the // resulting config. Address string // CAFile is the optional path to the CA certificate used for Consul // communication, defaults to the system bundle if not specified. CAFile string // CAPath is the optional path to a directory of CA certificates to use for // Consul communication, defaults to the system bundle if not specified. CAPath string // CertFile is the optional path to the certificate for Consul // communication. If this is set then you need to also set KeyFile. CertFile string // KeyFile is the optional path to the private key for Consul communication. // If this is set then you need to also set CertFile. KeyFile string // InsecureSkipVerify if set to true will disable TLS host verification. InsecureSkipVerify bool } // DefaultConfig returns a default configuration for the client. By default this // will pool and reuse idle connections to Consul. If you have a long-lived // client object, this is the desired behavior and should make the most efficient // use of the connections to Consul. If you don't reuse a client object , which // is not recommended, then you may notice idle connections building up over // time. To avoid this, use the DefaultNonPooledConfig() instead. func DefaultConfig() *Config { return defaultConfig(cleanhttp.DefaultPooledTransport) } // DefaultNonPooledConfig returns a default configuration for the client which // does not pool connections. This isn't a recommended configuration because it // will reconnect to Consul on every request, but this is useful to avoid the // accumulation of idle connections if you make many client objects during the // lifetime of your application. func DefaultNonPooledConfig() *Config { return defaultConfig(cleanhttp.DefaultTransport) } // defaultConfig returns the default configuration for the client, using the // given function to make the transport. func defaultConfig(transportFn func() *http.Transport) *Config { config := &Config{ Address: "127.0.0.1:8500", Scheme: "http", Transport: transportFn(), } if addr := os.Getenv(HTTPAddrEnvName); addr != "" { config.Address = addr } if token := os.Getenv(HTTPTokenEnvName); token != "" { config.Token = token } if auth := os.Getenv(HTTPAuthEnvName); auth != "" { var username, password string if strings.Contains(auth, ":") { split := strings.SplitN(auth, ":", 2) username = split[0] password = split[1] } else { username = auth } config.HttpAuth = &HttpBasicAuth{ Username: username, Password: password, } } if ssl := os.Getenv(HTTPSSLEnvName); ssl != "" { enabled, err := strconv.ParseBool(ssl) if err != nil { log.Printf("[WARN] client: could not parse %s: %s", HTTPSSLEnvName, err) } if enabled { config.Scheme = "https" } } if v := os.Getenv(HTTPTLSServerName); v != "" { config.TLSConfig.Address = v } if v := os.Getenv(HTTPCAFile); v != "" { config.TLSConfig.CAFile = v } if v := os.Getenv(HTTPCAPath); v != "" { config.TLSConfig.CAPath = v } if v := os.Getenv(HTTPClientCert); v != "" { config.TLSConfig.CertFile = v } if v := os.Getenv(HTTPClientKey); v != "" { config.TLSConfig.KeyFile = v } if v := os.Getenv(HTTPSSLVerifyEnvName); v != "" { doVerify, err := strconv.ParseBool(v) if err != nil { log.Printf("[WARN] client: could not parse %s: %s", HTTPSSLVerifyEnvName, err) } if !doVerify { config.TLSConfig.InsecureSkipVerify = true } } return config } // TLSConfig is used to generate a TLSClientConfig that's useful for talking to // Consul using TLS. func SetupTLSConfig(tlsConfig *TLSConfig) (*tls.Config, error) { tlsClientConfig := &tls.Config{ InsecureSkipVerify: tlsConfig.InsecureSkipVerify, } if tlsConfig.Address != "" { server := tlsConfig.Address hasPort := strings.LastIndex(server, ":") > strings.LastIndex(server, "]") if hasPort { var err error server, _, err = net.SplitHostPort(server) if err != nil { return nil, err } } tlsClientConfig.ServerName = server } if tlsConfig.CertFile != "" && tlsConfig.KeyFile != "" { tlsCert, err := tls.LoadX509KeyPair(tlsConfig.CertFile, tlsConfig.KeyFile) if err != nil { return nil, err } tlsClientConfig.Certificates = []tls.Certificate{tlsCert} } if tlsConfig.CAFile != "" || tlsConfig.CAPath != "" { rootConfig := &rootcerts.Config{ CAFile: tlsConfig.CAFile, CAPath: tlsConfig.CAPath, } if err := rootcerts.ConfigureTLS(tlsClientConfig, rootConfig); err != nil { return nil, err } } return tlsClientConfig, nil } // Client provides a client to the Consul API type Client struct { config Config } // NewClient returns a new client func NewClient(config *Config) (*Client, error) { // bootstrap the config defConfig := DefaultConfig() if len(config.Address) == 0 { config.Address = defConfig.Address } if len(config.Scheme) == 0 { config.Scheme = defConfig.Scheme } if config.Transport == nil { config.Transport = defConfig.Transport } if config.TLSConfig.Address == "" { config.TLSConfig.Address = defConfig.TLSConfig.Address } if config.TLSConfig.CAFile == "" { config.TLSConfig.CAFile = defConfig.TLSConfig.CAFile } if config.TLSConfig.CAPath == "" { config.TLSConfig.CAPath = defConfig.TLSConfig.CAPath } if config.TLSConfig.CertFile == "" { config.TLSConfig.CertFile = defConfig.TLSConfig.CertFile } if config.TLSConfig.KeyFile == "" { config.TLSConfig.KeyFile = defConfig.TLSConfig.KeyFile } if !config.TLSConfig.InsecureSkipVerify { config.TLSConfig.InsecureSkipVerify = defConfig.TLSConfig.InsecureSkipVerify } if config.HttpClient == nil { var err error config.HttpClient, err = NewHttpClient(config.Transport, config.TLSConfig) if err != nil { return nil, err } } parts := strings.SplitN(config.Address, "://", 2) if len(parts) == 2 { switch parts[0] { case "http": config.Scheme = "http" case "https": config.Scheme = "https" case "unix": trans := cleanhttp.DefaultTransport() trans.DialContext = func(_ context.Context, _, _ string) (net.Conn, error) { return net.Dial("unix", parts[1]) } config.HttpClient = &http.Client{ Transport: trans, } default: return nil, fmt.Errorf("Unknown protocol scheme: %s", parts[0]) } config.Address = parts[1] } if config.Token == "" { config.Token = defConfig.Token } return &Client{config: *config}, nil } // NewHttpClient returns an http client configured with the given Transport and TLS // config. func NewHttpClient(transport *http.Transport, tlsConf TLSConfig) (*http.Client, error) { client := &http.Client{ Transport: transport, } // TODO (slackpad) - Once we get some run time on the HTTP/2 support we // should turn it on by default if TLS is enabled. We would basically // just need to call http2.ConfigureTransport(transport) here. We also // don't want to introduce another external dependency on // golang.org/x/net/http2 at this time. For a complete recipe for how // to enable HTTP/2 support on a transport suitable for the API client // library see agent/http_test.go:TestHTTPServer_H2. if transport.TLSClientConfig == nil { tlsClientConfig, err := SetupTLSConfig(&tlsConf) if err != nil { return nil, err } transport.TLSClientConfig = tlsClientConfig } return client, nil } // request is used to help build up a request type request struct { config *Config method string url *url.URL params url.Values body io.Reader header http.Header obj interface{} ctx context.Context } // setQueryOptions is used to annotate the request with // additional query options func (r *request) setQueryOptions(q *QueryOptions) { if q == nil { return } if q.Datacenter != "" { r.params.Set("dc", q.Datacenter) } if q.AllowStale { r.params.Set("stale", "") } if q.RequireConsistent { r.params.Set("consistent", "") } if q.WaitIndex != 0 { r.params.Set("index", strconv.FormatUint(q.WaitIndex, 10)) } if q.WaitTime != 0 { r.params.Set("wait", durToMsec(q.WaitTime)) } if q.WaitHash != "" { r.params.Set("hash", q.WaitHash) } if q.Token != "" { r.header.Set("X-Consul-Token", q.Token) } if q.Near != "" { r.params.Set("near", q.Near) } if len(q.NodeMeta) > 0 { for key, value := range q.NodeMeta { r.params.Add("node-meta", key+":"+value) } } if q.RelayFactor != 0 { r.params.Set("relay-factor", strconv.Itoa(int(q.RelayFactor))) } if q.Connect { r.params.Set("connect", "true") } r.ctx = q.ctx } // durToMsec converts a duration to a millisecond specified string. If the // user selected a positive value that rounds to 0 ms, then we will use 1 ms // so they get a short delay, otherwise Consul will translate the 0 ms into // a huge default delay. func durToMsec(dur time.Duration) string { ms := dur / time.Millisecond if dur > 0 && ms == 0 { ms = 1 } return fmt.Sprintf("%dms", ms) } // serverError is a string we look for to detect 500 errors. const serverError = "Unexpected response code: 500" // IsRetryableError returns true for 500 errors from the Consul servers, and // network connection errors. These are usually retryable at a later time. // This applies to reads but NOT to writes. This may return true for errors // on writes that may have still gone through, so do not use this to retry // any write operations. func IsRetryableError(err error) bool { if err == nil { return false } if _, ok := err.(net.Error); ok { return true } // TODO (slackpad) - Make a real error type here instead of using // a string check. return strings.Contains(err.Error(), serverError) } // setWriteOptions is used to annotate the request with // additional write options func (r *request) setWriteOptions(q *WriteOptions) { if q == nil { return } if q.Datacenter != "" { r.params.Set("dc", q.Datacenter) } if q.Token != "" { r.header.Set("X-Consul-Token", q.Token) } if q.RelayFactor != 0 { r.params.Set("relay-factor", strconv.Itoa(int(q.RelayFactor))) } r.ctx = q.ctx } // toHTTP converts the request to an HTTP request func (r *request) toHTTP() (*http.Request, error) { // Encode the query parameters r.url.RawQuery = r.params.Encode() // Check if we should encode the body if r.body == nil && r.obj != nil { b, err := encodeBody(r.obj) if err != nil { return nil, err } r.body = b } // Create the HTTP request req, err := http.NewRequest(r.method, r.url.RequestURI(), r.body) if err != nil { return nil, err } req.URL.Host = r.url.Host req.URL.Scheme = r.url.Scheme req.Host = r.url.Host req.Header = r.header // Setup auth if r.config.HttpAuth != nil { req.SetBasicAuth(r.config.HttpAuth.Username, r.config.HttpAuth.Password) } if r.ctx != nil { return req.WithContext(r.ctx), nil } return req, nil } // newRequest is used to create a new request func (c *Client) newRequest(method, path string) *request { r := &request{ config: &c.config, method: method, url: &url.URL{ Scheme: c.config.Scheme, Host: c.config.Address, Path: path, }, params: make(map[string][]string), header: make(http.Header), } if c.config.Datacenter != "" { r.params.Set("dc", c.config.Datacenter) } if c.config.WaitTime != 0 { r.params.Set("wait", durToMsec(r.config.WaitTime)) } if c.config.Token != "" { r.header.Set("X-Consul-Token", r.config.Token) } return r } // doRequest runs a request with our client func (c *Client) doRequest(r *request) (time.Duration, *http.Response, error) { req, err := r.toHTTP() if err != nil { return 0, nil, err } start := time.Now() resp, err := c.config.HttpClient.Do(req) diff := time.Since(start) return diff, resp, err } // Query is used to do a GET request against an endpoint // and deserialize the response into an interface using // standard Consul conventions. func (c *Client) query(endpoint string, out interface{}, q *QueryOptions) (*QueryMeta, error) { r := c.newRequest("GET", endpoint) r.setQueryOptions(q) rtt, resp, err := requireOK(c.doRequest(r)) if err != nil { return nil, err } defer resp.Body.Close() qm := &QueryMeta{} parseQueryMeta(resp, qm) qm.RequestTime = rtt if err := decodeBody(resp, out); err != nil { return nil, err } return qm, nil } // write is used to do a PUT request against an endpoint // and serialize/deserialized using the standard Consul conventions. func (c *Client) write(endpoint string, in, out interface{}, q *WriteOptions) (*WriteMeta, error) { r := c.newRequest("PUT", endpoint) r.setWriteOptions(q) r.obj = in rtt, resp, err := requireOK(c.doRequest(r)) if err != nil { return nil, err } defer resp.Body.Close() wm := &WriteMeta{RequestTime: rtt} if out != nil { if err := decodeBody(resp, &out); err != nil { return nil, err } } else if _, err := ioutil.ReadAll(resp.Body); err != nil { return nil, err } return wm, nil } // parseQueryMeta is used to help parse query meta-data func parseQueryMeta(resp *http.Response, q *QueryMeta) error { header := resp.Header // Parse the X-Consul-Index (if it's set - hash based blocking queries don't // set this) if indexStr := header.Get("X-Consul-Index"); indexStr != "" { index, err := strconv.ParseUint(indexStr, 10, 64) if err != nil { return fmt.Errorf("Failed to parse X-Consul-Index: %v", err) } q.LastIndex = index } q.LastContentHash = header.Get("X-Consul-ContentHash") // Parse the X-Consul-LastContact last, err := strconv.ParseUint(header.Get("X-Consul-LastContact"), 10, 64) if err != nil { return fmt.Errorf("Failed to parse X-Consul-LastContact: %v", err) } q.LastContact = time.Duration(last) * time.Millisecond // Parse the X-Consul-KnownLeader switch header.Get("X-Consul-KnownLeader") { case "true": q.KnownLeader = true default: q.KnownLeader = false } // Parse X-Consul-Translate-Addresses switch header.Get("X-Consul-Translate-Addresses") { case "true": q.AddressTranslationEnabled = true default: q.AddressTranslationEnabled = false } return nil } // decodeBody is used to JSON decode a body func decodeBody(resp *http.Response, out interface{}) error { dec := json.NewDecoder(resp.Body) return dec.Decode(out) } // encodeBody is used to encode a request body func encodeBody(obj interface{}) (io.Reader, error) { buf := bytes.NewBuffer(nil) enc := json.NewEncoder(buf) if err := enc.Encode(obj); err != nil { return nil, err } return buf, nil } // requireOK is used to wrap doRequest and check for a 200 func requireOK(d time.Duration, resp *http.Response, e error) (time.Duration, *http.Response, error) { if e != nil { if resp != nil { resp.Body.Close() } return d, nil, e } if resp.StatusCode != 200 { var buf bytes.Buffer io.Copy(&buf, resp.Body) resp.Body.Close() return d, nil, fmt.Errorf("Unexpected response code: %d (%s)", resp.StatusCode, buf.Bytes()) } return d, resp, nil }