44e6b8122d
The error handling of the ACL code relies on the presence of certain magic error messages. Since the error values are sent via RPC between older and newer consul agents we cannot just replace the magic values with typed errors and switch to type checks since this would break compatibility with older clients. Therefore, this patch moves all magic ACL error messages into the acl package and provides default error values and helper functions which determine the type of error.
310 lines
8.7 KiB
Go
310 lines
8.7 KiB
Go
package consul
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/armon/go-metrics"
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/agent/consul/state"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/ipaddr"
|
|
"github.com/hashicorp/consul/types"
|
|
"github.com/hashicorp/go-memdb"
|
|
"github.com/hashicorp/go-uuid"
|
|
)
|
|
|
|
// Catalog endpoint is used to manipulate the service catalog
|
|
type Catalog struct {
|
|
srv *Server
|
|
}
|
|
|
|
// Register is used register that a node is providing a given service.
|
|
func (c *Catalog) Register(args *structs.RegisterRequest, reply *struct{}) error {
|
|
if done, err := c.srv.forward("Catalog.Register", args, args, reply); done {
|
|
return err
|
|
}
|
|
defer metrics.MeasureSince([]string{"consul", "catalog", "register"}, time.Now())
|
|
|
|
// Verify the args.
|
|
if args.Node == "" || args.Address == "" {
|
|
return fmt.Errorf("Must provide node and address")
|
|
}
|
|
if args.ID != "" {
|
|
if _, err := uuid.ParseUUID(string(args.ID)); err != nil {
|
|
return fmt.Errorf("Bad node ID: %v", err)
|
|
}
|
|
}
|
|
|
|
// Fetch the ACL token, if any.
|
|
rule, err := c.srv.resolveToken(args.Token)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Handle a service registration.
|
|
if args.Service != nil {
|
|
// If no service id, but service name, use default
|
|
if args.Service.ID == "" && args.Service.Service != "" {
|
|
args.Service.ID = args.Service.Service
|
|
}
|
|
|
|
// Verify ServiceName provided if ID.
|
|
if args.Service.ID != "" && args.Service.Service == "" {
|
|
return fmt.Errorf("Must provide service name with ID")
|
|
}
|
|
|
|
// Check the service address here and in the agent endpoint
|
|
// since service registration isn't sychronous.
|
|
if ipaddr.IsAny(args.Service.Address) {
|
|
return fmt.Errorf("Invalid service address")
|
|
}
|
|
|
|
// Apply the ACL policy if any. The 'consul' service is excluded
|
|
// since it is managed automatically internally (that behavior
|
|
// is going away after version 0.8). We check this same policy
|
|
// later if version 0.8 is enabled, so we can eventually just
|
|
// delete this and do all the ACL checks down there.
|
|
if args.Service.Service != structs.ConsulServiceName {
|
|
if rule != nil && !rule.ServiceWrite(args.Service.Service) {
|
|
return acl.ErrPermissionDenied
|
|
}
|
|
}
|
|
}
|
|
|
|
// Move the old format single check into the slice, and fixup IDs.
|
|
if args.Check != nil {
|
|
args.Checks = append(args.Checks, args.Check)
|
|
args.Check = nil
|
|
}
|
|
for _, check := range args.Checks {
|
|
if check.CheckID == "" && check.Name != "" {
|
|
check.CheckID = types.CheckID(check.Name)
|
|
}
|
|
if check.Node == "" {
|
|
check.Node = args.Node
|
|
}
|
|
}
|
|
|
|
// Check the complete register request against the given ACL policy.
|
|
if rule != nil && c.srv.config.ACLEnforceVersion8 {
|
|
state := c.srv.fsm.State()
|
|
_, ns, err := state.NodeServices(nil, args.Node)
|
|
if err != nil {
|
|
return fmt.Errorf("Node lookup failed: %v", err)
|
|
}
|
|
if err := vetRegisterWithACL(rule, args, ns); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
resp, err := c.srv.raftApply(structs.RegisterRequestType, args)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if respErr, ok := resp.(error); ok {
|
|
return respErr
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Deregister is used to remove a service registration for a given node.
|
|
func (c *Catalog) Deregister(args *structs.DeregisterRequest, reply *struct{}) error {
|
|
if done, err := c.srv.forward("Catalog.Deregister", args, args, reply); done {
|
|
return err
|
|
}
|
|
defer metrics.MeasureSince([]string{"consul", "catalog", "deregister"}, time.Now())
|
|
|
|
// Verify the args
|
|
if args.Node == "" {
|
|
return fmt.Errorf("Must provide node")
|
|
}
|
|
|
|
// Fetch the ACL token, if any.
|
|
rule, err := c.srv.resolveToken(args.Token)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Check the complete deregister request against the given ACL policy.
|
|
if rule != nil && c.srv.config.ACLEnforceVersion8 {
|
|
state := c.srv.fsm.State()
|
|
|
|
var ns *structs.NodeService
|
|
if args.ServiceID != "" {
|
|
_, ns, err = state.NodeService(args.Node, args.ServiceID)
|
|
if err != nil {
|
|
return fmt.Errorf("Service lookup failed: %v", err)
|
|
}
|
|
}
|
|
|
|
var nc *structs.HealthCheck
|
|
if args.CheckID != "" {
|
|
_, nc, err = state.NodeCheck(args.Node, args.CheckID)
|
|
if err != nil {
|
|
return fmt.Errorf("Check lookup failed: %v", err)
|
|
}
|
|
}
|
|
|
|
if err := vetDeregisterWithACL(rule, args, ns, nc); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if _, err := c.srv.raftApply(structs.DeregisterRequestType, args); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ListDatacenters is used to query for the list of known datacenters
|
|
func (c *Catalog) ListDatacenters(args *struct{}, reply *[]string) error {
|
|
dcs, err := c.srv.router.GetDatacentersByDistance()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
*reply = dcs
|
|
return nil
|
|
}
|
|
|
|
// ListNodes is used to query the nodes in a DC
|
|
func (c *Catalog) ListNodes(args *structs.DCSpecificRequest, reply *structs.IndexedNodes) error {
|
|
if done, err := c.srv.forward("Catalog.ListNodes", args, args, reply); done {
|
|
return err
|
|
}
|
|
|
|
return c.srv.blockingQuery(
|
|
&args.QueryOptions,
|
|
&reply.QueryMeta,
|
|
func(ws memdb.WatchSet, state *state.Store) error {
|
|
var index uint64
|
|
var nodes structs.Nodes
|
|
var err error
|
|
if len(args.NodeMetaFilters) > 0 {
|
|
index, nodes, err = state.NodesByMeta(ws, args.NodeMetaFilters)
|
|
} else {
|
|
index, nodes, err = state.Nodes(ws)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
reply.Index, reply.Nodes = index, nodes
|
|
if err := c.srv.filterACL(args.Token, reply); err != nil {
|
|
return err
|
|
}
|
|
return c.srv.sortNodesByDistanceFrom(args.Source, reply.Nodes)
|
|
})
|
|
}
|
|
|
|
// ListServices is used to query the services in a DC
|
|
func (c *Catalog) ListServices(args *structs.DCSpecificRequest, reply *structs.IndexedServices) error {
|
|
if done, err := c.srv.forward("Catalog.ListServices", args, args, reply); done {
|
|
return err
|
|
}
|
|
|
|
return c.srv.blockingQuery(
|
|
&args.QueryOptions,
|
|
&reply.QueryMeta,
|
|
func(ws memdb.WatchSet, state *state.Store) error {
|
|
var index uint64
|
|
var services structs.Services
|
|
var err error
|
|
if len(args.NodeMetaFilters) > 0 {
|
|
index, services, err = state.ServicesByNodeMeta(ws, args.NodeMetaFilters)
|
|
} else {
|
|
index, services, err = state.Services(ws)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
reply.Index, reply.Services = index, services
|
|
return c.srv.filterACL(args.Token, reply)
|
|
})
|
|
}
|
|
|
|
// ServiceNodes returns all the nodes registered as part of a service
|
|
func (c *Catalog) ServiceNodes(args *structs.ServiceSpecificRequest, reply *structs.IndexedServiceNodes) error {
|
|
if done, err := c.srv.forward("Catalog.ServiceNodes", args, args, reply); done {
|
|
return err
|
|
}
|
|
|
|
// Verify the arguments
|
|
if args.ServiceName == "" {
|
|
return fmt.Errorf("Must provide service name")
|
|
}
|
|
|
|
err := c.srv.blockingQuery(
|
|
&args.QueryOptions,
|
|
&reply.QueryMeta,
|
|
func(ws memdb.WatchSet, state *state.Store) error {
|
|
var index uint64
|
|
var services structs.ServiceNodes
|
|
var err error
|
|
if args.TagFilter {
|
|
index, services, err = state.ServiceTagNodes(ws, args.ServiceName, args.ServiceTag)
|
|
} else {
|
|
index, services, err = state.ServiceNodes(ws, args.ServiceName)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
reply.Index, reply.ServiceNodes = index, services
|
|
if len(args.NodeMetaFilters) > 0 {
|
|
var filtered structs.ServiceNodes
|
|
for _, service := range services {
|
|
if structs.SatisfiesMetaFilters(service.NodeMeta, args.NodeMetaFilters) {
|
|
filtered = append(filtered, service)
|
|
}
|
|
}
|
|
reply.ServiceNodes = filtered
|
|
}
|
|
if err := c.srv.filterACL(args.Token, reply); err != nil {
|
|
return err
|
|
}
|
|
return c.srv.sortNodesByDistanceFrom(args.Source, reply.ServiceNodes)
|
|
})
|
|
|
|
// Provide some metrics
|
|
if err == nil {
|
|
metrics.IncrCounterWithLabels([]string{"consul", "catalog", "service", "query"}, 1,
|
|
[]metrics.Label{{Name: "service", Value: args.ServiceName}})
|
|
if args.ServiceTag != "" {
|
|
metrics.IncrCounterWithLabels([]string{"consul", "catalog", "service", "query-tag"}, 1,
|
|
[]metrics.Label{{Name: "service", Value: args.ServiceName}, {Name: "tag", Value: args.ServiceTag}})
|
|
}
|
|
if len(reply.ServiceNodes) == 0 {
|
|
metrics.IncrCounterWithLabels([]string{"consul", "catalog", "service", "not-found"}, 1,
|
|
[]metrics.Label{{Name: "service", Value: args.ServiceName}})
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
// NodeServices returns all the services registered as part of a node
|
|
func (c *Catalog) NodeServices(args *structs.NodeSpecificRequest, reply *structs.IndexedNodeServices) error {
|
|
if done, err := c.srv.forward("Catalog.NodeServices", args, args, reply); done {
|
|
return err
|
|
}
|
|
|
|
// Verify the arguments
|
|
if args.Node == "" {
|
|
return fmt.Errorf("Must provide node")
|
|
}
|
|
|
|
return c.srv.blockingQuery(
|
|
&args.QueryOptions,
|
|
&reply.QueryMeta,
|
|
func(ws memdb.WatchSet, state *state.Store) error {
|
|
index, services, err := state.NodeServices(ws, args.Node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
reply.Index, reply.NodeServices = index, services
|
|
return c.srv.filterACL(args.Token, reply)
|
|
})
|
|
}
|