package agent import ( "crypto/tls" "encoding/json" "fmt" "io" "log" "net" "net/http" "net/http/pprof" "net/url" "os" "strconv" "strings" "time" "github.com/armon/go-metrics" "github.com/hashicorp/consul/consul/structs" "github.com/hashicorp/consul/tlsutil" "github.com/mitchellh/mapstructure" ) var ( // scadaHTTPAddr is the address associated with the // HTTPServer. When populating an ACL token for a request, // this is checked to switch between the ACLToken and // AtlasACLToken scadaHTTPAddr = "SCADA" ) // HTTPServer is used to wrap an Agent and expose various API's // in a RESTful manner type HTTPServer struct { agent *Agent mux *http.ServeMux listener net.Listener logger *log.Logger uiDir string addr string } // NewHTTPServers starts new HTTP servers to provide an interface to // the agent. func NewHTTPServers(agent *Agent, config *Config, logOutput io.Writer) ([]*HTTPServer, error) { if logOutput == nil { return nil, fmt.Errorf("Please provide a valid logOutput(io.Writer)") } var servers []*HTTPServer if config.Ports.HTTPS > 0 { httpAddr, err := config.ClientListener(config.Addresses.HTTPS, config.Ports.HTTPS) if err != nil { return nil, err } tlsConf := &tlsutil.Config{ VerifyIncoming: config.VerifyIncoming, VerifyOutgoing: config.VerifyOutgoing, CAFile: config.CAFile, CertFile: config.CertFile, KeyFile: config.KeyFile, NodeName: config.NodeName, ServerName: config.ServerName} tlsConfig, err := tlsConf.IncomingTLSConfig() if err != nil { return nil, err } ln, err := net.Listen(httpAddr.Network(), httpAddr.String()) if err != nil { return nil, fmt.Errorf("Failed to get Listen on %s: %v", httpAddr.String(), err) } list := tls.NewListener(tcpKeepAliveListener{ln.(*net.TCPListener)}, tlsConfig) // Create the mux mux := http.NewServeMux() // Create the server srv := &HTTPServer{ agent: agent, mux: mux, listener: list, logger: log.New(logOutput, "", log.LstdFlags), uiDir: config.UiDir, addr: httpAddr.String(), } srv.registerHandlers(config.EnableDebug) // Start the server go http.Serve(list, mux) servers = append(servers, srv) } if config.Ports.HTTP > 0 { httpAddr, err := config.ClientListener(config.Addresses.HTTP, config.Ports.HTTP) if err != nil { return nil, fmt.Errorf("Failed to get ClientListener address:port: %v", err) } // Error if we are trying to bind a domain socket to an existing path socketPath, isSocket := unixSocketAddr(config.Addresses.HTTP) if isSocket { if _, err := os.Stat(socketPath); !os.IsNotExist(err) { agent.logger.Printf("[WARN] agent: Replacing socket %q", socketPath) } if err := os.Remove(socketPath); err != nil && !os.IsNotExist(err) { return nil, fmt.Errorf("error removing socket file: %s", err) } } ln, err := net.Listen(httpAddr.Network(), httpAddr.String()) if err != nil { return nil, fmt.Errorf("Failed to get Listen on %s: %v", httpAddr.String(), err) } var list net.Listener if isSocket { // Set up ownership/permission bits on the socket file if err := setFilePermissions(socketPath, config.UnixSockets); err != nil { return nil, fmt.Errorf("Failed setting up HTTP socket: %s", err) } list = ln } else { list = tcpKeepAliveListener{ln.(*net.TCPListener)} } // Create the mux mux := http.NewServeMux() // Create the server srv := &HTTPServer{ agent: agent, mux: mux, listener: list, logger: log.New(logOutput, "", log.LstdFlags), uiDir: config.UiDir, addr: httpAddr.String(), } srv.registerHandlers(config.EnableDebug) // Start the server go http.Serve(list, mux) servers = append(servers, srv) } return servers, nil } // newScadaHttp creates a new HTTP server wrapping the SCADA // listener such that HTTP calls can be sent from the brokers. func newScadaHttp(agent *Agent, list net.Listener) *HTTPServer { // Create the mux mux := http.NewServeMux() // Create the server srv := &HTTPServer{ agent: agent, mux: mux, listener: list, logger: agent.logger, addr: scadaHTTPAddr, } srv.registerHandlers(false) // Never allow debug for SCADA // Start the server go http.Serve(list, mux) return srv } // tcpKeepAliveListener sets TCP keep-alive timeouts on accepted // connections. It's used by NewHttpServer so // dead TCP connections eventually go away. type tcpKeepAliveListener struct { *net.TCPListener } func (ln tcpKeepAliveListener) Accept() (c net.Conn, err error) { tc, err := ln.AcceptTCP() if err != nil { return } tc.SetKeepAlive(true) tc.SetKeepAlivePeriod(30 * time.Second) return tc, nil } // Shutdown is used to shutdown the HTTP server func (s *HTTPServer) Shutdown() { if s != nil { s.logger.Printf("[DEBUG] http: Shutting down http server (%v)", s.addr) s.listener.Close() } } // handleFuncMetrics takes the given pattern and handler and wraps to produce // metrics based on the pattern and request. func (s *HTTPServer) handleFuncMetrics(pattern string, handler func(http.ResponseWriter, *http.Request)) { // Get the parts of the pattern. We omit any initial empty for the // leading slash, and put an underscore as a "thing" placeholder if we // see a trailing slash, which means the part after is parsed. This lets // us distinguish from things like /v1/query and /v1/query/. var parts []string for i, part := range strings.Split(pattern, "/") { if part == "" { if i == 0 { continue } else { part = "_" } } parts = append(parts, part) } // Register the wrapper, which will close over the expensive-to-compute // parts from above. wrapper := func(resp http.ResponseWriter, req *http.Request) { start := time.Now() handler(resp, req) key := append([]string{"consul", "http", req.Method}, parts...) metrics.MeasureSince(key, start) } s.mux.HandleFunc(pattern, wrapper) } // registerHandlers is used to attach our handlers to the mux func (s *HTTPServer) registerHandlers(enableDebug bool) { s.mux.HandleFunc("/", s.Index) // API V1. if s.agent.config.ACLDatacenter != "" { s.handleFuncMetrics("/v1/acl/create", s.wrap(s.ACLCreate)) s.handleFuncMetrics("/v1/acl/update", s.wrap(s.ACLUpdate)) s.handleFuncMetrics("/v1/acl/destroy/", s.wrap(s.ACLDestroy)) s.handleFuncMetrics("/v1/acl/info/", s.wrap(s.ACLGet)) s.handleFuncMetrics("/v1/acl/clone/", s.wrap(s.ACLClone)) s.handleFuncMetrics("/v1/acl/list", s.wrap(s.ACLList)) s.handleFuncMetrics("/v1/acl/replication", s.wrap(s.ACLReplicationStatus)) } else { s.handleFuncMetrics("/v1/acl/create", s.wrap(aclDisabled)) s.handleFuncMetrics("/v1/acl/update", s.wrap(aclDisabled)) s.handleFuncMetrics("/v1/acl/destroy/", s.wrap(aclDisabled)) s.handleFuncMetrics("/v1/acl/info/", s.wrap(aclDisabled)) s.handleFuncMetrics("/v1/acl/clone/", s.wrap(aclDisabled)) s.handleFuncMetrics("/v1/acl/list", s.wrap(aclDisabled)) s.handleFuncMetrics("/v1/acl/replication", s.wrap(aclDisabled)) } s.handleFuncMetrics("/v1/agent/self", s.wrap(s.AgentSelf)) s.handleFuncMetrics("/v1/agent/maintenance", s.wrap(s.AgentNodeMaintenance)) s.handleFuncMetrics("/v1/agent/services", s.wrap(s.AgentServices)) s.handleFuncMetrics("/v1/agent/checks", s.wrap(s.AgentChecks)) s.handleFuncMetrics("/v1/agent/members", s.wrap(s.AgentMembers)) s.handleFuncMetrics("/v1/agent/join/", s.wrap(s.AgentJoin)) s.handleFuncMetrics("/v1/agent/force-leave/", s.wrap(s.AgentForceLeave)) s.handleFuncMetrics("/v1/agent/check/register", s.wrap(s.AgentRegisterCheck)) s.handleFuncMetrics("/v1/agent/check/deregister/", s.wrap(s.AgentDeregisterCheck)) s.handleFuncMetrics("/v1/agent/check/pass/", s.wrap(s.AgentCheckPass)) s.handleFuncMetrics("/v1/agent/check/warn/", s.wrap(s.AgentCheckWarn)) s.handleFuncMetrics("/v1/agent/check/fail/", s.wrap(s.AgentCheckFail)) s.handleFuncMetrics("/v1/agent/check/update/", s.wrap(s.AgentCheckUpdate)) s.handleFuncMetrics("/v1/agent/service/register", s.wrap(s.AgentRegisterService)) s.handleFuncMetrics("/v1/agent/service/deregister/", s.wrap(s.AgentDeregisterService)) s.handleFuncMetrics("/v1/agent/service/maintenance/", s.wrap(s.AgentServiceMaintenance)) s.handleFuncMetrics("/v1/catalog/register", s.wrap(s.CatalogRegister)) s.handleFuncMetrics("/v1/catalog/deregister", s.wrap(s.CatalogDeregister)) s.handleFuncMetrics("/v1/catalog/datacenters", s.wrap(s.CatalogDatacenters)) s.handleFuncMetrics("/v1/catalog/nodes", s.wrap(s.CatalogNodes)) s.handleFuncMetrics("/v1/catalog/services", s.wrap(s.CatalogServices)) s.handleFuncMetrics("/v1/catalog/service/", s.wrap(s.CatalogServiceNodes)) s.handleFuncMetrics("/v1/catalog/node/", s.wrap(s.CatalogNodeServices)) if !s.agent.config.DisableCoordinates { s.handleFuncMetrics("/v1/coordinate/datacenters", s.wrap(s.CoordinateDatacenters)) s.handleFuncMetrics("/v1/coordinate/nodes", s.wrap(s.CoordinateNodes)) } else { s.handleFuncMetrics("/v1/coordinate/datacenters", s.wrap(coordinateDisabled)) s.handleFuncMetrics("/v1/coordinate/nodes", s.wrap(coordinateDisabled)) } s.handleFuncMetrics("/v1/event/fire/", s.wrap(s.EventFire)) s.handleFuncMetrics("/v1/event/list", s.wrap(s.EventList)) s.handleFuncMetrics("/v1/health/node/", s.wrap(s.HealthNodeChecks)) s.handleFuncMetrics("/v1/health/checks/", s.wrap(s.HealthServiceChecks)) s.handleFuncMetrics("/v1/health/state/", s.wrap(s.HealthChecksInState)) s.handleFuncMetrics("/v1/health/service/", s.wrap(s.HealthServiceNodes)) s.handleFuncMetrics("/v1/internal/ui/nodes", s.wrap(s.UINodes)) s.handleFuncMetrics("/v1/internal/ui/node/", s.wrap(s.UINodeInfo)) s.handleFuncMetrics("/v1/internal/ui/services", s.wrap(s.UIServices)) s.handleFuncMetrics("/v1/kv/", s.wrap(s.KVSEndpoint)) s.handleFuncMetrics("/v1/operator/raft/configuration", s.wrap(s.OperatorRaftConfiguration)) s.handleFuncMetrics("/v1/operator/raft/peer", s.wrap(s.OperatorRaftPeer)) s.handleFuncMetrics("/v1/query", s.wrap(s.PreparedQueryGeneral)) s.handleFuncMetrics("/v1/query/", s.wrap(s.PreparedQuerySpecific)) s.handleFuncMetrics("/v1/session/create", s.wrap(s.SessionCreate)) s.handleFuncMetrics("/v1/session/destroy/", s.wrap(s.SessionDestroy)) s.handleFuncMetrics("/v1/session/renew/", s.wrap(s.SessionRenew)) s.handleFuncMetrics("/v1/session/info/", s.wrap(s.SessionGet)) s.handleFuncMetrics("/v1/session/node/", s.wrap(s.SessionsForNode)) s.handleFuncMetrics("/v1/session/list", s.wrap(s.SessionList)) s.handleFuncMetrics("/v1/status/leader", s.wrap(s.StatusLeader)) s.handleFuncMetrics("/v1/status/peers", s.wrap(s.StatusPeers)) s.handleFuncMetrics("/v1/snapshot", s.wrap(s.Snapshot)) s.handleFuncMetrics("/v1/txn", s.wrap(s.Txn)) // Debug endpoints. if enableDebug { s.handleFuncMetrics("/debug/pprof/", pprof.Index) s.handleFuncMetrics("/debug/pprof/cmdline", pprof.Cmdline) s.handleFuncMetrics("/debug/pprof/profile", pprof.Profile) s.handleFuncMetrics("/debug/pprof/symbol", pprof.Symbol) } // Use the custom UI dir if provided. if s.uiDir != "" { s.mux.Handle("/ui/", http.StripPrefix("/ui/", http.FileServer(http.Dir(s.uiDir)))) } else if s.agent.config.EnableUi { s.mux.Handle("/ui/", http.StripPrefix("/ui/", http.FileServer(assetFS()))) } } // wrap is used to wrap functions to make them more convenient func (s *HTTPServer) wrap(handler func(resp http.ResponseWriter, req *http.Request) (interface{}, error)) func(resp http.ResponseWriter, req *http.Request) { f := func(resp http.ResponseWriter, req *http.Request) { setHeaders(resp, s.agent.config.HTTPAPIResponseHeaders) setTranslateAddr(resp, s.agent.config.TranslateWanAddrs) // Obfuscate any tokens from appearing in the logs formVals, err := url.ParseQuery(req.URL.RawQuery) if err != nil { s.logger.Printf("[ERR] http: Failed to decode query: %s from=%s", err, req.RemoteAddr) resp.WriteHeader(http.StatusInternalServerError) // 500 return } logURL := req.URL.String() if tokens, ok := formVals["token"]; ok { for _, token := range tokens { if token == "" { logURL += "" continue } logURL = strings.Replace(logURL, token, "", -1) } } // TODO (slackpad) We may want to consider redacting prepared // query names/IDs here since they are proxies for tokens. But, // knowing one only gives you read access to service listings // which is pretty trivial, so it's probably not worth the code // complexity and overhead of filtering them out. You can't // recover the token it's a proxy for with just the query info; // you'd need the actual token (or a management token) to read // that back. // Invoke the handler start := time.Now() defer func() { s.logger.Printf("[DEBUG] http: Request %s %v (%v) from=%s", req.Method, logURL, time.Now().Sub(start), req.RemoteAddr) }() obj, err := handler(resp, req) // Check for an error HAS_ERR: if err != nil { s.logger.Printf("[ERR] http: Request %s %v, error: %v from=%s", req.Method, logURL, err, req.RemoteAddr) code := http.StatusInternalServerError // 500 errMsg := err.Error() if strings.Contains(errMsg, "Permission denied") || strings.Contains(errMsg, "ACL not found") { code = http.StatusForbidden // 403 } resp.WriteHeader(code) resp.Write([]byte(err.Error())) return } if obj != nil { var buf []byte buf, err = s.marshalJSON(req, obj) if err != nil { goto HAS_ERR } resp.Header().Set("Content-Type", "application/json") resp.Write(buf) } } return f } // marshalJSON marshals the object into JSON, respecting the user's pretty-ness // configuration. func (s *HTTPServer) marshalJSON(req *http.Request, obj interface{}) ([]byte, error) { if _, ok := req.URL.Query()["pretty"]; ok { buf, err := json.MarshalIndent(obj, "", " ") if err != nil { return nil, err } buf = append(buf, "\n"...) return buf, nil } buf, err := json.Marshal(obj) if err != nil { return nil, err } return buf, err } // Returns true if the UI is enabled. func (s *HTTPServer) IsUIEnabled() bool { return s.uiDir != "" || s.agent.config.EnableUi } // Renders a simple index page func (s *HTTPServer) Index(resp http.ResponseWriter, req *http.Request) { // Check if this is a non-index path if req.URL.Path != "/" { resp.WriteHeader(http.StatusNotFound) // 404 return } // Give them something helpful if there's no UI so they at least know // what this server is. if !s.IsUIEnabled() { resp.Write([]byte("Consul Agent")) return } // Redirect to the UI endpoint http.Redirect(resp, req, "/ui/", http.StatusMovedPermanently) // 301 } // decodeBody is used to decode a JSON request body func decodeBody(req *http.Request, out interface{}, cb func(interface{}) error) error { var raw interface{} dec := json.NewDecoder(req.Body) if err := dec.Decode(&raw); err != nil { return err } // Invoke the callback prior to decode if cb != nil { if err := cb(raw); err != nil { return err } } return mapstructure.Decode(raw, out) } // setTranslateAddr is used to set the address translation header. This is only // present if the feature is active. func setTranslateAddr(resp http.ResponseWriter, active bool) { if active { resp.Header().Set("X-Consul-Translate-Addresses", "true") } } // setIndex is used to set the index response header func setIndex(resp http.ResponseWriter, index uint64) { resp.Header().Set("X-Consul-Index", strconv.FormatUint(index, 10)) } // setKnownLeader is used to set the known leader header func setKnownLeader(resp http.ResponseWriter, known bool) { s := "true" if !known { s = "false" } resp.Header().Set("X-Consul-KnownLeader", s) } // setLastContact is used to set the last contact header func setLastContact(resp http.ResponseWriter, last time.Duration) { lastMsec := uint64(last / time.Millisecond) resp.Header().Set("X-Consul-LastContact", strconv.FormatUint(lastMsec, 10)) } // setMeta is used to set the query response meta data func setMeta(resp http.ResponseWriter, m *structs.QueryMeta) { setIndex(resp, m.Index) setLastContact(resp, m.LastContact) setKnownLeader(resp, m.KnownLeader) } // setHeaders is used to set canonical response header fields func setHeaders(resp http.ResponseWriter, headers map[string]string) { for field, value := range headers { resp.Header().Set(http.CanonicalHeaderKey(field), value) } } // parseWait is used to parse the ?wait and ?index query params // Returns true on error func parseWait(resp http.ResponseWriter, req *http.Request, b *structs.QueryOptions) bool { query := req.URL.Query() if wait := query.Get("wait"); wait != "" { dur, err := time.ParseDuration(wait) if err != nil { resp.WriteHeader(http.StatusBadRequest) // 400 resp.Write([]byte("Invalid wait time")) return true } b.MaxQueryTime = dur } if idx := query.Get("index"); idx != "" { index, err := strconv.ParseUint(idx, 10, 64) if err != nil { resp.WriteHeader(http.StatusBadRequest) // 400 resp.Write([]byte("Invalid index")) return true } b.MinQueryIndex = index } return false } // parseConsistency is used to parse the ?stale and ?consistent query params. // Returns true on error func parseConsistency(resp http.ResponseWriter, req *http.Request, b *structs.QueryOptions) bool { query := req.URL.Query() if _, ok := query["stale"]; ok { b.AllowStale = true } if _, ok := query["consistent"]; ok { b.RequireConsistent = true } if b.AllowStale && b.RequireConsistent { resp.WriteHeader(http.StatusBadRequest) // 400 resp.Write([]byte("Cannot specify ?stale with ?consistent, conflicting semantics.")) return true } return false } // parseDC is used to parse the ?dc query param func (s *HTTPServer) parseDC(req *http.Request, dc *string) { if other := req.URL.Query().Get("dc"); other != "" { *dc = other } else if *dc == "" { *dc = s.agent.config.Datacenter } } // parseToken is used to parse the ?token query param or the X-Consul-Token header func (s *HTTPServer) parseToken(req *http.Request, token *string) { if other := req.URL.Query().Get("token"); other != "" { *token = other return } if other := req.Header.Get("X-Consul-Token"); other != "" { *token = other return } // Set the AtlasACLToken if SCADA if s.addr == scadaHTTPAddr && s.agent.config.AtlasACLToken != "" { *token = s.agent.config.AtlasACLToken return } // Set the default ACLToken *token = s.agent.config.ACLToken } // parseSource is used to parse the ?near= query parameter, used for // sorting by RTT based on a source node. We set the source's DC to the target // DC in the request, if given, or else the agent's DC. func (s *HTTPServer) parseSource(req *http.Request, source *structs.QuerySource) { s.parseDC(req, &source.Datacenter) if node := req.URL.Query().Get("near"); node != "" { if node == "_agent" { source.Node = s.agent.config.NodeName } else { source.Node = node } } } // parse is a convenience method for endpoints that need // to use both parseWait and parseDC. func (s *HTTPServer) parse(resp http.ResponseWriter, req *http.Request, dc *string, b *structs.QueryOptions) bool { s.parseDC(req, dc) s.parseToken(req, &b.Token) if parseConsistency(resp, req, b) { return true } return parseWait(resp, req, b) }