// handleAgentForwardNode will create a unix socket and serve the agent running
// on the client on it.
974802
func (s *Server) handleAgentForwardNode(req *ssh.Request, ctx *srv.ServerContext) error {
	// check if the user's RBAC role allows agent forwarding
	err := s.authHandlers.CheckAgentForward(ctx)
	if err != nil {
		return trace.Wrap(err)
	}

	// open a channel to the client where the client will serve an agent
	authChannel, _, err := ctx.Conn.OpenChannel(sshutils.AuthAgentRequest, nil)
	if err != nil {
		return trace.Wrap(err)
	}

	// save the agent in the context so it can be used later
	ctx.SetAgent(agent.NewClient(authChannel), authChannel)

	// serve an agent on a unix socket on this node
	err = s.serveAgent(ctx)
	if err != nil {
		return trace.Wrap(err)
	}

	return nil
}
1102013
func (idr *InitDryRunGetter) handleGetNode(action core.GetAction) (bool, runtime.Object, error) {
	if action.GetName() != idr.controlPlaneName || action.GetResource().Resource != "nodes" {
		// We can't handle this event
		return false, nil, nil
	}

	return true, &v1.Node{
		ObjectMeta: metav1.ObjectMeta{
			Name: idr.controlPlaneName,
			Labels: map[string]string{
				"kubernetes.io/hostname": idr.controlPlaneName,
			},
			Annotations: map[string]string{},
		},
	}, nil
}
1361060
func (s *Serf) handleNodeUpdate(n *memberlist.Node) {
	s.memberLock.Lock()
	defer s.memberLock.Unlock()

	member, ok := s.members[n.Name]
	if !ok {
		// We've never even heard of this node that is updating.
		// Just ignore it completely.
		return
	}

	// Update the member attributes
	member.Addr = net.IP(n.Addr)
	member.Port = n.Port
	member.Tags = s.decodeTags(n.Meta)

	// Snag the latest versions. NOTE - the current memberlist code will NOT
	// fire an update event if the metadata (for Serf, tags) stays the same
	// and only the protocol versions change. If we wake any Serf-level
	// protocol changes where we want to get this event under those
	// circumstances, we will need to update memberlist to do a check of
	// versions as well as the metadata.
	member.ProtocolMin = n.PMin
	member.ProtocolMax = n.PMax
	member.ProtocolCur = n.PCur
	member.DelegateMin = n.DMin
	member.DelegateMax = n.DMax
	member.DelegateCur = n.DCur

	// Update some metrics
	metrics.IncrCounter([]string{"serf", "member", "update"}, 1)

	// Send an event along
	s.logger.Printf("[INFO] serf: EventMemberUpdate: %s", member.Member.Name)
	if s.config.EventCh != nil {
		s.config.EventCh <- MemberEvent{
			Type:    EventMemberUpdate,
			Members: []Member{member.Member},
		}
	}
}
529631
func (f *fs) handleMknod(r *fuse.MknodRequest) {
	log.Println("Inside handleMknod")
	// NOTE: We probably will not need this since we implement Create
	r.RespondError(fuse.EIO)
}
1227033
func (c *NodeClient) listenAndForward(ctx context.Context, ln net.Listener, remoteAddr string) {
	defer ln.Close()
	defer c.Close()

	for {
		// Accept connections from the client.
		conn, err := ln.Accept()
		if err != nil {
			log.Errorf("Port forwarding failed: %v.", err)
			break
		}

		// Proxy the connection to the remote address.
		go func() {
			err := c.proxyConnection(ctx, conn, remoteAddr)
			if err != nil {
				log.Warnf("Failed to proxy connection: %v.", err)
			}
		}()
	}
}
1355634
func createAgent(me *user.User, privateKeyByte []byte, certificateBytes []byte) (*teleagent.AgentServer, string, string, error) {
	// create a path to the unix socket
	sockDir, err := ioutil.TempDir("", "int-test")
	if err != nil {
		return nil, "", "", trace.Wrap(err)
	}
	sockPath := filepath.Join(sockDir, "agent.sock")

	uid, err := strconv.Atoi(me.Uid)
	if err != nil {
		return nil, "", "", trace.Wrap(err)
	}
	gid, err := strconv.Atoi(me.Gid)
	if err != nil {
		return nil, "", "", trace.Wrap(err)
	}

	// transform the key and certificate bytes into something the agent can understand
	publicKey, _, _, _, err := ssh.ParseAuthorizedKey(certificateBytes)
	if err != nil {
		return nil, "", "", trace.Wrap(err)
	}
	privateKey, err := ssh.ParseRawPrivateKey(privateKeyByte)
	if err != nil {
		return nil, "", "", trace.Wrap(err)
	}
	agentKey := agent.AddedKey{
		PrivateKey:       privateKey,
		Certificate:      publicKey.(*ssh.Certificate),
		Comment:          "",
		LifetimeSecs:     0,
		ConfirmBeforeUse: false,
	}

	// create a (unstarted) agent and add the key to it
	teleAgent := teleagent.NewServer()
	teleAgent.Add(agentKey)

	// start the SSH agent
	err = teleAgent.ListenUnixSocket(sockPath, uid, gid, 0600)
	if err != nil {
		return nil, "", "", trace.Wrap(err)
	}
	go teleAgent.Serve()

	return teleAgent, sockDir, sockPath, nil
}
1141027
func (c *connections) ForAgent(agent string) Connections {
	var values []Value
	for _, value := range c.values {
		if value.Agent == agent {
			values = append(values, value)
		}
	}
	return &connections{model: c.model, values: values}
}
346023
func (c *Command) startAgent(config *Config, agent *Agent,
	logWriter *logWriter, logOutput io.Writer) *AgentIPC {
	// Add the script event handlers
	c.scriptHandler = &ScriptEventHandler{
		SelfFunc: func() serf.Member { return agent.Serf().LocalMember() },
		Scripts:  config.EventScripts(),
		Logger:   log.New(logOutput, "", log.LstdFlags),
	}
	agent.RegisterEventHandler(c.scriptHandler)

	// Start the agent after the handler is registered
	if err := agent.Start(); err != nil {
		c.Ui.Error(fmt.Sprintf("Failed to start the Serf agent: %v", err))
		return nil
	}

	// Parse the bind address information
	bindIP, bindPort, err := config.AddrParts(config.BindAddr)
	bindAddr := &net.TCPAddr{IP: net.ParseIP(bindIP), Port: bindPort}

	// Start the discovery layer
	if config.Discover != "" {
		// Use the advertise addr and port
		local := agent.Serf().Memberlist().LocalNode()

		// Get the bind interface if any
		iface, _ := config.NetworkInterface()

		_, err := NewAgentMDNS(agent, logOutput, config.ReplayOnJoin,
			config.NodeName, config.Discover, iface, local.Addr, int(local.Port))
		if err != nil {
			c.Ui.Error(fmt.Sprintf("Error starting mDNS listener: %s", err))
			return nil

		}
	}

	// Setup the RPC listener
	rpcListener, err := net.Listen("tcp", config.RPCAddr)
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Error starting RPC listener: %s", err))
		return nil
	}

	// Start the IPC layer
	c.Ui.Output("Starting Serf agent RPC...")
	ipc := NewAgentIPC(agent, config.RPCAuthKey, rpcListener, logOutput, logWriter)

	c.Ui.Output("Serf agent running!")
	c.Ui.Info(fmt.Sprintf("                  Node name: '%s'", config.NodeName))
	c.Ui.Info(fmt.Sprintf("                  Bind addr: '%s'", bindAddr.String()))

	if config.AdvertiseAddr != "" {
		advertiseIP, advertisePort, _ := config.AddrParts(config.AdvertiseAddr)
		advertiseAddr := (&net.TCPAddr{IP: net.ParseIP(advertiseIP), Port: advertisePort}).String()
		c.Ui.Info(fmt.Sprintf("Advertise addr: '%s'", advertiseAddr))
	}

	c.Ui.Info(fmt.Sprintf("                   RPC addr: '%s'", config.RPCAddr))
	c.Ui.Info(fmt.Sprintf("                  Encrypted: %#v", agent.serf.EncryptionEnabled()))
	c.Ui.Info(fmt.Sprintf("                   Snapshot: %v", config.SnapshotPath != ""))
	c.Ui.Info(fmt.Sprintf("                    Profile: %s", config.Profile))
	c.Ui.Info(fmt.Sprintf("Message Compression Enabled: %v", config.EnableCompression))

	if config.Discover != "" {
		c.Ui.Info(fmt.Sprintf("               mDNS cluster: %s", config.Discover))
	}
	return ipc
}
779436
func NewForward(ctx context.Context, next http.Handler, config config.ForwardAuth, name string) (http.Handler, error) {
	middlewares.GetLogger(ctx, name, forwardedTypeName).Debug("Creating middleware")

	fa := &forwardAuth{
		address:             config.Address,
		authResponseHeaders: config.AuthResponseHeaders,
		next:                next,
		name:                name,
		trustForwardHeader:  config.TrustForwardHeader,
	}

	if config.TLS != nil {
		tlsConfig, err := config.TLS.CreateTLSConfig()
		if err != nil {
			return nil, err
		}

		fa.tlsConfig = tlsConfig
	}

	return fa, nil
}
1186335
func (s *Switch) handlePacketForward(packet *htlcPacket) error {
	switch htlc := packet.htlc.(type) {

	// Channel link forwarded us a new htlc, therefore we initiate the
	// payment circuit within our internal state so we can properly forward
	// the ultimate settle message back latter.
	case *lnwire.UpdateAddHTLC:
		if packet.incomingChanID == sourceHop {
			// A blank incomingChanID indicates that this is
			// a pending user-initiated payment.
			return s.handleLocalDispatch(packet)
		}

		s.indexMtx.RLock()
		targetLink, err := s.getLinkByShortID(packet.outgoingChanID)
		if err != nil {
			s.indexMtx.RUnlock()

			// If packet was forwarded from another channel link
			// than we should notify this link that some error
			// occurred.
			failure := &lnwire.FailUnknownNextPeer{}
			addErr := fmt.Errorf("unable to find link with "+
				"destination %v", packet.outgoingChanID)

			return s.failAddPacket(packet, failure, addErr)
		}
		targetPeerKey := targetLink.Peer().PubKey()
		interfaceLinks, _ := s.getLinks(targetPeerKey)
		s.indexMtx.RUnlock()

		// We'll keep track of any HTLC failures during the link
		// selection process. This way we can return the error for
		// precise link that the sender selected, while optimistically
		// trying all links to utilize our available bandwidth.
		linkErrs := make(map[lnwire.ShortChannelID]lnwire.FailureMessage)

		// Try to find destination channel link with appropriate
		// bandwidth.
		var destination ChannelLink
		for _, link := range interfaceLinks {
			// We'll skip any links that aren't yet eligible for
			// forwarding.
			switch {
			case !link.EligibleToForward():
				continue

			// If the link doesn't yet have a source chan ID, then
			// we'll skip it as well.
			case link.ShortChanID() == sourceHop:
				continue
			}

			// Before we check the link's bandwidth, we'll ensure
			// that the HTLC satisfies the current forwarding
			// policy of this target link.
			currentHeight := atomic.LoadUint32(&s.bestHeight)
			err := link.HtlcSatifiesPolicy(
				htlc.PaymentHash, packet.incomingAmount,
				packet.amount, packet.incomingTimeout,
				packet.outgoingTimeout, currentHeight,
			)
			if err != nil {
				linkErrs[link.ShortChanID()] = err
				continue
			}

			if link.Bandwidth() >= htlc.Amount {
				destination = link

				break
			}
		}

		switch {
		// If the channel link we're attempting to forward the update
		// over has insufficient capacity, and didn't violate any
		// forwarding policies, then we'll cancel the htlc as the
		// payment cannot succeed.
		case destination == nil && len(linkErrs) == 0:
			// If packet was forwarded from another channel link
			// than we should notify this link that some error
			// occurred.
			var failure lnwire.FailureMessage
			update, err := s.cfg.FetchLastChannelUpdate(
				packet.outgoingChanID,
			)
			if err != nil {
				failure = &lnwire.FailTemporaryNodeFailure{}
			} else {
				failure = lnwire.NewTemporaryChannelFailure(update)
			}

			addErr := fmt.Errorf("unable to find appropriate "+
				"channel link insufficient capacity, need "+
				"%v towards node=%x", htlc.Amount, targetPeerKey)

			return s.failAddPacket(packet, failure, addErr)

		// If we had a forwarding failure due to the HTLC not
		// satisfying the current policy, then we'll send back an
		// error, but ensure we send back the error sourced at the
		// *target* link.
		case destination == nil && len(linkErrs) != 0:
			// At this point, some or all of the links rejected the
			// HTLC so we couldn't forward it. So we'll try to look
			// up the error that came from the source.
			linkErr, ok := linkErrs[packet.outgoingChanID]
			if !ok {
				// If we can't find the error of the source,
				// then we'll return an unknown next peer,
				// though this should never happen.
				linkErr = &lnwire.FailUnknownNextPeer{}
				log.Warnf("unable to find err source for "+
					"outgoing_link=%v, errors=%v",
					packet.outgoingChanID, newLogClosure(func() string {
						return spew.Sdump(linkErrs)
					}))
			}

			addErr := fmt.Errorf("incoming HTLC(%x) violated "+
				"target outgoing link (id=%v) policy: %v",
				htlc.PaymentHash[:], packet.outgoingChanID,
				linkErr)

			return s.failAddPacket(packet, linkErr, addErr)
		}

		// Send the packet to the destination channel link which
		// manages the channel.
		packet.outgoingChanID = destination.ShortChanID()
		return destination.HandleSwitchPacket(packet)

	case *lnwire.UpdateFailHTLC, *lnwire.UpdateFulfillHTLC:
		// If the source of this packet has not been set, use the
		// circuit map to lookup the origin.
		circuit, err := s.closeCircuit(packet)
		if err != nil {
			return err
		}

		fail, isFail := htlc.(*lnwire.UpdateFailHTLC)
		if isFail && !packet.hasSource {
			switch {
			case circuit.ErrorEncrypter == nil:
				// No message to encrypt, locally sourced
				// payment.

			case packet.isResolution:
				// If this is a resolution message, then we'll need to encrypt
				// it as it's actually internally sourced.
				var err error
				// TODO(roasbeef): don't need to pass actually?
				failure := &lnwire.FailPermanentChannelFailure{}
				fail.Reason, err = circuit.ErrorEncrypter.EncryptFirstHop(
					failure,
				)
				if err != nil {
					err = fmt.Errorf("unable to obfuscate "+
						"error: %v", err)
					log.Error(err)
				}

			default:
				// Otherwise, it's a forwarded error, so we'll perform a
				// wrapper encryption as normal.
				fail.Reason = circuit.ErrorEncrypter.IntermediateEncrypt(
					fail.Reason,
				)
			}
		} else if !isFail && circuit.Outgoing != nil {
			// If this is an HTLC settle, and it wasn't from a
			// locally initiated HTLC, then we'll log a forwarding
			// event so we can flush it to disk later.
			//
			// TODO(roasbeef): only do this once link actually
			// fully settles?
			localHTLC := packet.incomingChanID == sourceHop
			if !localHTLC {
				s.fwdEventMtx.Lock()
				s.pendingFwdingEvents = append(
					s.pendingFwdingEvents,
					channeldb.ForwardingEvent{
						Timestamp:      time.Now(),
						IncomingChanID: circuit.Incoming.ChanID,
						OutgoingChanID: circuit.Outgoing.ChanID,
						AmtIn:          circuit.IncomingAmount,
						AmtOut:         circuit.OutgoingAmount,
					},
				)
				s.fwdEventMtx.Unlock()
			}
		}

		// A blank IncomingChanID in a circuit indicates that it is a pending
		// user-initiated payment.
		if packet.incomingChanID == sourceHop {
			return s.handleLocalDispatch(packet)
		}

		// Check to see that the source link is online before removing
		// the circuit.
		return s.mailOrchestrator.Deliver(packet.incomingChanID, packet)

	default:
		return errors.New("wrong update type")
	}
}
1347559
func (d *domainClient) ResolveNode(ctx context.Context, args *ResolveNodeArgs) (reply *ResolveNodeReply, err error) {
	reply = new(ResolveNodeReply)
	if args != nil {
		err = rpcc.Invoke(ctx, "DOM.resolveNode", args, reply, d.conn)
	} else {
		err = rpcc.Invoke(ctx, "DOM.resolveNode", nil, reply, d.conn)
	}
	if err != nil {
		err = &internal.OpError{Domain: "DOM", Op: "ResolveNode", Err: err}
	}
	return
}
1011559
func (r *ChannelRouter) AddNode(node *channeldb.LightningNode) error {
	rMsg := &routingMsg{
		msg: node,
		err: make(chan error, 1),
	}

	select {
	case r.networkUpdates <- rMsg:
		select {
		case err := <-rMsg.err:
			return err
		case <-r.quit:
			return ErrRouterShuttingDown
		}
	case <-r.quit:
		return ErrRouterShuttingDown
	}
}
689980
func (c *testCluster) AddAgent() error {
	// first node
	if len(c.nodes) == 0 {
		return fmt.Errorf("there is no manager nodes")
	}
	joinAddr, err := c.RandomManager().node.RemoteAPIAddr()
	if err != nil {
		return err
	}
	clusterInfo, err := c.GetClusterInfo()
	if err != nil {
		return err
	}
	node, err := newTestNode(joinAddr, clusterInfo.RootCA.JoinTokens.Worker, false, c.fips)
	if err != nil {
		return err
	}
	return c.AddNode(node)
}
479028
func (p *Proxy) handleTunnel(session sockjs.Session, req *http.Request) {
	tokenString := req.URL.Query().Get("token")

	getPublicKey := func(token *jwt.Token) (interface{}, error) {
		if _, ok := token.Method.(*jwt.SigningMethodRSA); !ok {
			return nil, errors.New("invalid signing method")
		}

		return jwt.ParseRSAPublicKeyFromPEM([]byte(p.pubKey))
	}

	token, err := jwt.Parse(tokenString, getPublicKey)
	if err != nil {
		p.Kite.Log.Error("Invalid token: \"%s\"", tokenString)
		return
	}

	kiteID := token.Claims.(jwt.MapClaims)["sub"].(string)
	seq := uint64(token.Claims.(jwt.MapClaims)["seq"].(float64))

	client, ok := p.kites[kiteID]
	if !ok {
		p.Kite.Log.Error("Remote kite is not found: %s", kiteID)
		return
	}

	tunnel, ok := client.tunnels[seq]
	if !ok {
		p.Kite.Log.Error("Tunnel not found: %d", seq)
	}

	go tunnel.Run(session)

	<-tunnel.CloseNotify()

}
891257
func (n *Node) runAgent(ctx context.Context, db *bolt.DB, securityConfig *ca.SecurityConfig, ready chan<- struct{}) error {
	// First, get a channel for knowing when a remote peer has been selected.
	// The value returned from the remotesCh is ignored, we just need to know
	// when the peer is selected
	remotesCh := n.remotes.WaitSelect(ctx)
	// then, we set up a new context to pass specifically to
	// ListenControlSocket, and start that method to wait on a connection on
	// the cluster control API.
	waitCtx, waitCancel := context.WithCancel(ctx)
	controlCh := n.ListenControlSocket(waitCtx)

	// The goal here to wait either until we have a remote peer selected, or
	// connection to the control
	// socket. These are both ways to connect the
	// agent to a manager, and we need to wait until one or the other is
	// available to start the agent
waitPeer:
	for {
		select {
		case <-ctx.Done():
			break waitPeer
		case <-remotesCh:
			break waitPeer
		case conn := <-controlCh:
			// conn will probably be nil the first time we call this, probably,
			// but only a non-nil conn represent an actual connection.
			if conn != nil {
				break waitPeer
			}
		}
	}

	// We can stop listening for new control socket connections once we're
	// ready
	waitCancel()

	// NOTE(dperny): not sure why we need to recheck the context here. I guess
	// it avoids a race if the context was canceled at the same time that a
	// connection or peer was available. I think it's just an optimization.
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	// Now we can go ahead and configure, create, and start the agent.
	secChangesCh, secChangesCancel := securityConfig.Watch()
	defer secChangesCancel()

	rootCA := securityConfig.RootCA()
	issuer := securityConfig.IssuerInfo()

	agentConfig := &agent.Config{
		Hostname:         n.config.Hostname,
		ConnBroker:       n.connBroker,
		Executor:         n.config.Executor,
		DB:               db,
		NotifyNodeChange: n.notifyNodeChange,
		NotifyTLSChange:  secChangesCh,
		Credentials:      securityConfig.ClientTLSCreds,
		NodeTLSInfo: &api.NodeTLSInfo{
			TrustRoot:           rootCA.Certs,
			CertIssuerPublicKey: issuer.PublicKey,
			CertIssuerSubject:   issuer.Subject,
		},
		FIPS: n.config.FIPS,
	}
	// if a join address has been specified, then if the agent fails to connect
	// due to a TLS error, fail fast - don't keep re-trying to join
	if n.config.JoinAddr != "" {
		agentConfig.SessionTracker = &firstSessionErrorTracker{}
	}

	a, err := agent.New(agentConfig)
	if err != nil {
		return err
	}
	if err := a.Start(ctx); err != nil {
		return err
	}

	n.Lock()
	n.agent = a
	n.Unlock()

	defer func() {
		n.Lock()
		n.agent = nil
		n.Unlock()
	}()

	// when the agent indicates that it is ready, we close the ready channel.
	go func() {
		<-a.Ready()
		close(ready)
	}()

	// todo: manually call stop on context cancellation?

	return a.Err(context.Background())
}
639071
func handleNode(s *rpcServer, cmd interface{}, closeChan <-chan struct{}) (interface{}, error) {
	c := cmd.(*btcjson.NodeCmd)

	var addr string
	var nodeID uint64
	var errN, err error
	params := s.cfg.ChainParams
	switch c.SubCmd {
	case "disconnect":
		// If we have a valid uint disconnect by node id. Otherwise,
		// attempt to disconnect by address, returning an error if a
		// valid IP address is not supplied.
		if nodeID, errN = strconv.ParseUint(c.Target, 10, 32); errN == nil {
			err = s.cfg.ConnMgr.DisconnectByID(int32(nodeID))
		} else {
			if _, _, errP := net.SplitHostPort(c.Target); errP == nil || net.ParseIP(c.Target) != nil {
				addr = normalizeAddress(c.Target, params.DefaultPort)
				err = s.cfg.ConnMgr.DisconnectByAddr(addr)
			} else {
				return nil, &btcjson.RPCError{
					Code:    btcjson.ErrRPCInvalidParameter,
					Message: "invalid address or node ID",
				}
			}
		}
		if err != nil && peerExists(s.cfg.ConnMgr, addr, int32(nodeID)) {

			return nil, &btcjson.RPCError{
				Code:    btcjson.ErrRPCMisc,
				Message: "can't disconnect a permanent peer, use remove",
			}
		}

	case "remove":
		// If we have a valid uint disconnect by node id. Otherwise,
		// attempt to disconnect by address, returning an error if a
		// valid IP address is not supplied.
		if nodeID, errN = strconv.ParseUint(c.Target, 10, 32); errN == nil {
			err = s.cfg.ConnMgr.RemoveByID(int32(nodeID))
		} else {
			if _, _, errP := net.SplitHostPort(c.Target); errP == nil || net.ParseIP(c.Target) != nil {
				addr = normalizeAddress(c.Target, params.DefaultPort)
				err = s.cfg.ConnMgr.RemoveByAddr(addr)
			} else {
				return nil, &btcjson.RPCError{
					Code:    btcjson.ErrRPCInvalidParameter,
					Message: "invalid address or node ID",
				}
			}
		}
		if err != nil && peerExists(s.cfg.ConnMgr, addr, int32(nodeID)) {
			return nil, &btcjson.RPCError{
				Code:    btcjson.ErrRPCMisc,
				Message: "can't remove a temporary peer, use disconnect",
			}
		}

	case "connect":
		addr = normalizeAddress(c.Target, params.DefaultPort)

		// Default to temporary connections.
		subCmd := "temp"
		if c.ConnectSubCmd != nil {
			subCmd = *c.ConnectSubCmd
		}

		switch subCmd {
		case "perm", "temp":
			err = s.cfg.ConnMgr.Connect(addr, subCmd == "perm")
		default:
			return nil, &btcjson.RPCError{
				Code:    btcjson.ErrRPCInvalidParameter,
				Message: "invalid subcommand for node connect",
			}
		}
	default:
		return nil, &btcjson.RPCError{
			Code:    btcjson.ErrRPCInvalidParameter,
			Message: "invalid subcommand for node",
		}
	}

	if err != nil {
		return nil, &btcjson.RPCError{
			Code:    btcjson.ErrRPCInvalidParameter,
			Message: err.Error(),
		}
	}

	// no data returned unless an error.
	return nil, nil
}
1230196
func (n *Node) Agent() *agent.Agent {
	n.RLock()
	defer n.RUnlock()
	return n.agent
}