Add support for remote_cluster, implements gravitational#1526

This commit adds remote cluster resource that specifies
connection and trust of the remote trusted cluster to the local
cluster. Deleting remote cluster resource deletes trust
established between clusters on the local cluster side
and terminates all reverse tunnel connections.

Migrations make sure that remote cluster resources exist
after upgrade of the auth server.

constants.go

@@ -266,3 +266,12 @@ const DefaultImplicitRole = "default-implicit-role"
 
 // APIDomain is a default domain name for Auth server API
 const APIDomain = "teleport.cluster.local"
+
+const (
+	// RemoteClusterStatusOffline indicates that cluster is considered as
+	// offline, since it has missed a series of heartbeats
+	RemoteClusterStatusOffline = "offline"
+	// RemoteClusterStatusOnline indicates that cluster is sending heartbeats
+	// at expected interval
+	RemoteClusterStatusOnline = "online"
+)

integration/integration_test.go

@@ -1075,6 +1075,180 @@ func (s *IntSuite) TestMapRoles(c *check.C) {
 	c.Assert(aux.Stop(true), check.IsNil)
 }
 
+// TestRemoteClusters tests disconnecting remote clusters
+// using remote cluster feature
+func (s *IntSuite) TestRemoteClusters(c *check.C) {
+	username := s.me.Username
+
+	clusterMain := "cluster-main"
+	clusterAux := "cluster-aux"
+	main := NewInstance(clusterMain, HostID, Host, s.getPorts(5), s.priv, s.pub)
+	aux := NewInstance(clusterAux, HostID, Host, s.getPorts(5), s.priv, s.pub)
+
+	// main cluster has a local user and belongs to role "main-devs"
+	mainDevs := "main-devs"
+	role, err := services.NewRole(mainDevs, services.RoleSpecV3{
+		Allow: services.RoleConditions{
+			Logins: []string{username},
+		},
+	})
+	c.Assert(err, check.IsNil)
+	main.AddUserWithRole(username, role)
+
+	// for role mapping test we turn on Web API on the main cluster
+	// as it's used
+	makeConfig := func(enableSSH bool) ([]*InstanceSecrets, *service.Config) {
+		tconf := service.MakeDefaultConfig()
+		tconf.SSH.Enabled = enableSSH
+		tconf.Console = nil
+		tconf.Proxy.DisableWebService = false
+		tconf.Proxy.DisableWebInterface = true
+		return nil, tconf
+	}
+	lib.SetInsecureDevMode(true)
+	defer lib.SetInsecureDevMode(false)
+
+	c.Assert(main.CreateEx(makeConfig(false)), check.IsNil)
+	c.Assert(aux.CreateEx(makeConfig(true)), check.IsNil)
+
+	// auxiliary cluster has a role aux-devs
+	// connect aux cluster to main cluster
+	// using trusted clusters, so remote user will be allowed to assume
+	// role specified by mapping remote role "devs" to local role "local-devs"
+	auxDevs := "aux-devs"
+	role, err = services.NewRole(auxDevs, services.RoleSpecV3{
+		Allow: services.RoleConditions{
+			Logins: []string{username},
+		},
+	})
+	c.Assert(err, check.IsNil)
+	err = aux.Process.GetAuthServer().UpsertRole(role, backend.Forever)
+	c.Assert(err, check.IsNil)
+	trustedClusterToken := "trusted-clsuter-token"
+	err = main.Process.GetAuthServer().UpsertToken(trustedClusterToken, []teleport.Role{teleport.RoleTrustedCluster}, backend.Forever)
+	c.Assert(err, check.IsNil)
+	trustedCluster := main.Secrets.AsTrustedCluster(trustedClusterToken, services.RoleMap{
+		{Remote: mainDevs, Local: []string{auxDevs}},
+	})
+
+	// modify trusted cluster resource name so it would not
+	// match the cluster name to check that it does not matter
+	trustedCluster.SetName(main.Secrets.SiteName + "-cluster")
+
+	c.Assert(main.Start(), check.IsNil)
+	c.Assert(aux.Start(), check.IsNil)
+
+	err = trustedCluster.CheckAndSetDefaults()
+	c.Assert(err, check.IsNil)
+
+	// try and upsert a trusted cluster
+	var upsertSuccess bool
+	for i := 0; i < 10; i++ {
+		log.Debugf("Will create trusted cluster %v, attempt %v", trustedCluster, i)
+		err = aux.Process.GetAuthServer().UpsertTrustedCluster(trustedCluster)
+		if err != nil {
+			if trace.IsConnectionProblem(err) {
+				log.Debugf("retrying on connection problem: %v", err)
+				continue
+			}
+			c.Fatalf("got non connection problem %v", err)
+		}
+		upsertSuccess = true
+		break
+	}
+	// make sure we upsert a trusted cluster
+	c.Assert(upsertSuccess, check.Equals, true)
+
+	nodePorts := s.getPorts(3)
+	sshPort, proxyWebPort, proxySSHPort := nodePorts[0], nodePorts[1], nodePorts[2]
+	c.Assert(aux.StartNodeAndProxy("aux-node", sshPort, proxyWebPort, proxySSHPort), check.IsNil)
+
+	// wait for both sites to see each other via their reverse tunnels (for up to 10 seconds)
+	abortTime := time.Now().Add(time.Second * 10)
+	for len(main.Tunnel.GetSites()) < 2 && len(main.Tunnel.GetSites()) < 2 {
+		time.Sleep(time.Millisecond * 2000)
+		if time.Now().After(abortTime) {
+			c.Fatalf("two clusters do not see each other: tunnels are not working")
+		}
+	}
+
+	cmd := []string{"echo", "hello world"}
+	tc, err := main.NewClient(ClientConfig{Login: username, Cluster: clusterAux, Host: "127.0.0.1", Port: sshPort})
+	c.Assert(err, check.IsNil)
+	output := &bytes.Buffer{}
+	tc.Stdout = output
+	c.Assert(err, check.IsNil)
+	// try to execute an SSH command using the same old client  to Site-B
+	// "site-A" and "site-B" reverse tunnels are supposed to reconnect,
+	// and 'tc' (client) is also supposed to reconnect
+	for i := 0; i < 10; i++ {
+		time.Sleep(time.Millisecond * 50)
+		err = tc.SSH(context.TODO(), cmd, false)
+		if err == nil {
+			break
+		}
+	}
+	c.Assert(err, check.IsNil)
+	c.Assert(output.String(), check.Equals, "hello world\n")
+
+	// check that remote cluster has been provisioned
+	remoteClusters, err := main.Process.GetAuthServer().GetRemoteClusters()
+	c.Assert(err, check.IsNil)
+	c.Assert(remoteClusters, check.HasLen, 1)
+	c.Assert(remoteClusters[0].GetName(), check.Equals, clusterAux)
+
+	// after removing the remote cluster, the connection will start failing
+	err = main.Process.GetAuthServer().DeleteRemoteCluster(clusterAux)
+	c.Assert(err, check.IsNil)
+	for i := 0; i < 10; i++ {
+		time.Sleep(time.Millisecond * 50)
+		err = tc.SSH(context.TODO(), cmd, false)
+		if err != nil {
+			break
+		}
+	}
+	c.Assert(err, check.NotNil, check.Commentf("expected tunnel to close and SSH client to start failing"))
+
+	// remove trusted cluster from aux cluster side, and recrete right after
+	// this should re-establish connection
+	err = aux.Process.GetAuthServer().DeleteTrustedCluster(trustedCluster.GetName())
+	c.Assert(err, check.IsNil)
+	err = aux.Process.GetAuthServer().UpsertTrustedCluster(trustedCluster)
+	c.Assert(err, check.IsNil)
+
+	// check that remote cluster has been re-provisioned
+	remoteClusters, err = main.Process.GetAuthServer().GetRemoteClusters()
+	c.Assert(err, check.IsNil)
+	c.Assert(remoteClusters, check.HasLen, 1)
+	c.Assert(remoteClusters[0].GetName(), check.Equals, clusterAux)
+
+	// wait for both sites to see each other via their reverse tunnels (for up to 10 seconds)
+	abortTime = time.Now().Add(time.Second * 10)
+	for len(main.Tunnel.GetSites()) < 2 {
+		time.Sleep(time.Millisecond * 2000)
+		if time.Now().After(abortTime) {
+			c.Fatalf("two clusters do not see each other: tunnels are not working")
+		}
+	}
+
+	// connection and client should recover and work again
+	output = &bytes.Buffer{}
+	tc.Stdout = output
+	for i := 0; i < 10; i++ {
+		time.Sleep(time.Millisecond * 50)
+		err = tc.SSH(context.TODO(), cmd, false)
+		if err == nil {
+			break
+		}
+	}
+	c.Assert(err, check.IsNil)
+	c.Assert(output.String(), check.Equals, "hello world\n")
+
+	// stop clusters and remaining nodes
+	c.Assert(main.Stop(true), check.IsNil)
+	c.Assert(aux.Stop(true), check.IsNil)
+}
+
 // TestDiscovery tests case for multiple proxies and a reverse tunnel
 // agent that eventually connnects to the the right proxy
 func (s *IntSuite) TestDiscovery(c *check.C) {

lib/auth/apiserver.go

@@ -119,6 +119,12 @@ func NewAPIServer(config *APIConfig) http.Handler {
 	// Server Credentials
 	srv.POST("/:version/server/credentials", srv.withAuth(srv.generateServerKeys))
 
+	srv.POST("/:version/remoteclusters", srv.withAuth(srv.createRemoteCluster))
+	srv.GET("/:version/remoteclusters/:cluster", srv.withAuth(srv.getRemoteCluster))
+	srv.GET("/:version/remoteclusters", srv.withAuth(srv.getRemoteClusters))
+	srv.DELETE("/:version/remoteclusters/:cluster", srv.withAuth(srv.deleteRemoteCluster))
+	srv.DELETE("/:version/remoteclusters", srv.withAuth(srv.deleteAllRemoteClusters))
+
 	// Reverse tunnels
 	srv.POST("/:version/reversetunnels", srv.withAuth(srv.upsertReverseTunnel))
 	srv.GET("/:version/reversetunnels", srv.withAuth(srv.getReverseTunnels))
@@ -2199,6 +2205,71 @@ func (s *APIServer) deleteAllTunnelConnections(auth ClientI, w http.ResponseWrit
 	return message("ok"), nil
 }
 
+type createRemoteClusterRawReq struct {
+	// RemoteCluster is marshalled remote cluster resource
+	RemoteCluster json.RawMessage `json:"remote_cluster"`
+}
+
+// createRemoteCluster creates remote cluster
+func (s *APIServer) createRemoteCluster(auth ClientI, w http.ResponseWriter, r *http.Request, p httprouter.Params, version string) (interface{}, error) {
+	var req createRemoteClusterRawReq
+	if err := httplib.ReadJSON(r, &req); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	conn, err := services.UnmarshalRemoteCluster(req.RemoteCluster)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := auth.CreateRemoteCluster(conn); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return message("ok"), nil
+}
+
+// getRemoteClusters returns a list of remote clusters
+func (s *APIServer) getRemoteClusters(auth ClientI, w http.ResponseWriter, r *http.Request, p httprouter.Params, version string) (interface{}, error) {
+	clusters, err := auth.GetRemoteClusters()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	items := make([]json.RawMessage, len(clusters))
+	for i, cluster := range clusters {
+		data, err := services.MarshalRemoteCluster(cluster, services.WithVersion(version))
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		items[i] = data
+	}
+	return items, nil
+}
+
+// getRemoteCluster returns a remote cluster by name
+func (s *APIServer) getRemoteCluster(auth ClientI, w http.ResponseWriter, r *http.Request, p httprouter.Params, version string) (interface{}, error) {
+	cluster, err := auth.GetRemoteCluster(p.ByName("cluster"))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return rawMessage(services.MarshalRemoteCluster(cluster, services.WithVersion(version)))
+}
+
+// deleteRemoteCluster deletes remote cluster by name
+func (s *APIServer) deleteRemoteCluster(auth ClientI, w http.ResponseWriter, r *http.Request, p httprouter.Params, version string) (interface{}, error) {
+	err := auth.DeleteRemoteCluster(p.ByName("cluster"))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return message("ok"), nil
+}
+
+// deleteAllRemoteClusters deletes all remote clusters
+func (s *APIServer) deleteAllRemoteClusters(auth ClientI, w http.ResponseWriter, r *http.Request, p httprouter.Params, version string) (interface{}, error) {
+	err := auth.DeleteAllRemoteClusters()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return message("ok"), nil
+}
+
 func message(msg string) map[string]interface{} {
 	return map[string]interface{}{"message": msg}
 }

lib/auth/auth_with_roles.go

@@ -1108,6 +1108,44 @@ func (a *AuthWithRoles) DeleteAllTunnelConnections() error {
 	return a.authServer.DeleteAllTunnelConnections()
 }
 
+func (a *AuthWithRoles) CreateRemoteCluster(conn services.RemoteCluster) error {
+	if err := a.action(defaults.Namespace, services.KindRemoteCluster, services.VerbCreate); err != nil {
+		return trace.Wrap(err)
+	}
+	return a.authServer.CreateRemoteCluster(conn)
+}
+
+func (a *AuthWithRoles) GetRemoteCluster(clusterName string) (services.RemoteCluster, error) {
+	if err := a.action(defaults.Namespace, services.KindRemoteCluster, services.VerbRead); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return a.authServer.GetRemoteCluster(clusterName)
+}
+
+func (a *AuthWithRoles) GetRemoteClusters() ([]services.RemoteCluster, error) {
+	if err := a.action(defaults.Namespace, services.KindRemoteCluster, services.VerbList); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return a.authServer.GetRemoteClusters()
+}
+
+func (a *AuthWithRoles) DeleteRemoteCluster(clusterName string) error {
+	if err := a.action(defaults.Namespace, services.KindRemoteCluster, services.VerbDelete); err != nil {
+		return trace.Wrap(err)
+	}
+	return a.authServer.DeleteRemoteCluster(clusterName)
+}
+
+func (a *AuthWithRoles) DeleteAllRemoteClusters() error {
+	if err := a.action(defaults.Namespace, services.KindRemoteCluster, services.VerbList); err != nil {
+		return trace.Wrap(err)
+	}
+	if err := a.action(defaults.Namespace, services.KindRemoteCluster, services.VerbDelete); err != nil {
+		return trace.Wrap(err)
+	}
+	return a.authServer.DeleteAllRemoteClusters()
+}
+
 func (a *AuthWithRoles) Close() error {
 	return a.authServer.Close()
 }

lib/auth/clt.go

@@ -686,6 +686,67 @@ func (c *Client) GetUserLoginAttempts(user string) ([]services.LoginAttempt, err
 	panic("not implemented")
 }
 
+// GetRemoteClusters returns a list of remote clusters
+func (c *Client) GetRemoteClusters() ([]services.RemoteCluster, error) {
+	out, err := c.Get(c.Endpoint("remoteclusters"), url.Values{})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	var items []json.RawMessage
+	if err := json.Unmarshal(out.Bytes(), &items); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	conns := make([]services.RemoteCluster, len(items))
+	for i, raw := range items {
+		conn, err := services.UnmarshalRemoteCluster(raw)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		conns[i] = conn
+	}
+	return conns, nil
+}
+
+// GetRemoteCluster returns a remote cluster by name
+func (c *Client) GetRemoteCluster(clusterName string) (services.RemoteCluster, error) {
+	if clusterName == "" {
+		return nil, trace.BadParameter("missing cluster name")
+	}
+	out, err := c.Get(c.Endpoint("remoteclusters", clusterName), url.Values{})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return services.UnmarshalRemoteCluster(out.Bytes())
+}
+
+// DeleteRemoteCluster deletes remote cluster by name
+func (c *Client) DeleteRemoteCluster(clusterName string) error {
+	if clusterName == "" {
+		return trace.BadParameter("missing parameter cluster name")
+	}
+	_, err := c.Delete(c.Endpoint("remoteclusters", clusterName))
+	return trace.Wrap(err)
+}
+
+// DeleteAllRemoteClusters deletes all remote clusters
+func (c *Client) DeleteAllRemoteClusters() error {
+	_, err := c.Delete(c.Endpoint("remoteclusters"))
+	return trace.Wrap(err)
+}
+
+// CreateRemoteCluster creates remote cluster resource
+func (c *Client) CreateRemoteCluster(rc services.RemoteCluster) error {
+	data, err := services.MarshalRemoteCluster(rc)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	args := &createRemoteClusterRawReq{
+		RemoteCluster: data,
+	}
+	_, err = c.PostJSON(c.Endpoint("remoteclusters"), args)
+	return trace.Wrap(err)
+}
+
 // UpsertAuthServer is used by auth servers to report their presence
 // to other auth servers in form of hearbeat expiring after ttl period.
 func (c *Client) UpsertAuthServer(s services.Server) error {