diff --git a/fixtures/trusted_clusters/cluster-a b/fixtures/trusted_clusters/cluster-a
index ee18845d51056..67cff02cb36fa 120000
--- a/fixtures/trusted_clusters/cluster-a
+++ b/fixtures/trusted_clusters/cluster-a
@@ -1 +1 @@
-../../vagrant/libvirt/data/a-auth/cluster-a
\ No newline at end of file
+../../vagrant/libvirt/data/var/cluster-a
\ No newline at end of file
diff --git a/fixtures/trusted_clusters/cluster-b b/fixtures/trusted_clusters/cluster-b
deleted file mode 120000
index 1473e240c0b61..0000000000000
--- a/fixtures/trusted_clusters/cluster-b
+++ /dev/null
@@ -1 +0,0 @@
-../../vagrant/libvirt/data/b-auth/cluster-b
\ No newline at end of file
diff --git a/lib/config/config_test.go b/lib/config/config_test.go
index 008c65ae4d1d6..173aba1a47657 100644
--- a/lib/config/config_test.go
+++ b/lib/config/config_test.go
@@ -213,11 +213,19 @@ func (s *ConfigTestSuite) TestLocateWebAssets(c *check.C) {
 }
 
 func (s *ConfigTestSuite) TestTrustedClusters(c *check.C) {
-	files := []string{
-		"../../fixtures/trusted_clusters/cluster-a",
-	}
-	authorities, err := readTrustedClusters(files)
+	err := readTrustedClusters(nil, nil)
 	c.Assert(err, check.IsNil)
+
+	var conf service.Config
+	err = readTrustedClusters([]TrustedCluster{
+		{
+			AllowedLogins: "vagrant, root",
+			KeyFile:       "../../fixtures/trusted_clusters/cluster-a",
+			TunnelAddr:    "one,two",
+		},
+	}, &conf)
+	c.Assert(err, check.IsNil)
+	authorities := conf.Auth.Authorities
 	c.Assert(len(authorities), check.Equals, 2)
 	c.Assert(authorities[0].DomainName, check.Equals, "cluster-a")
 	c.Assert(authorities[0].Type, check.Equals, services.HostCA)
@@ -228,11 +236,40 @@ func (s *ConfigTestSuite) TestTrustedClusters(c *check.C) {
 	_, _, _, _, err = ssh.ParseAuthorizedKey(authorities[1].CheckingKeys[0])
 	c.Assert(err, check.IsNil)
 
-	// try to read the file of a wrong format:
-	authorities, err = readTrustedClusters([]string{"../../README.md"})
+	tunnels := conf.ReverseTunnels
+	c.Assert(len(tunnels), check.Equals, 1)
+	c.Assert(tunnels[0].DomainName, check.Equals, "cluster-a")
+	c.Assert(len(tunnels[0].DialAddrs), check.Equals, 2)
+	c.Assert(tunnels[0].DialAddrs[0], check.Equals, "tcp://one:3024")
+	c.Assert(tunnels[0].DialAddrs[1], check.Equals, "tcp://two:3024")
+
+	// invalid data:
+	err = readTrustedClusters([]TrustedCluster{
+		{
+			AllowedLogins: "vagrant, root",
+			KeyFile:       "non-existing",
+			TunnelAddr:    "one,two",
+		},
+	}, &conf)
 	c.Assert(err, check.NotNil)
-	c.Assert(authorities, check.IsNil)
-	c.Assert(err, check.ErrorMatches, "^.*invalid file format.*$")
+	c.Assert(err, check.ErrorMatches, "^.*reading trusted cluster keys.*$")
+	err = readTrustedClusters([]TrustedCluster{
+		{
+			KeyFile:    "../../fixtures/trusted_clusters/cluster-a",
+			TunnelAddr: "one,two",
+		},
+	}, &conf)
+	c.Assert(err, check.ErrorMatches, ".*needs allow_logins parameter")
+	conf.ReverseTunnels = nil
+	err = readTrustedClusters([]TrustedCluster{
+		{
+			KeyFile:       "../../fixtures/trusted_clusters/cluster-a",
+			AllowedLogins: "vagrant",
+			TunnelAddr:    "",
+		},
+	}, &conf)
+	c.Assert(err, check.IsNil)
+	c.Assert(len(conf.ReverseTunnels), check.Equals, 0)
 }
 
 func (s *ConfigTestSuite) TestApplyConfig(c *check.C) {
diff --git a/lib/config/configuration.go b/lib/config/configuration.go
index 48b381caf98ba..eaa6ecb4491eb 100644
--- a/lib/config/configuration.go
+++ b/lib/config/configuration.go
@@ -320,69 +320,107 @@ func ApplyFileConfig(fc *FileConfig, cfg *service.Config) error {
 	}
 	// read 'trusted_clusters' section:
 	if fc.Auth.Enabled() && len(fc.Auth.TrustedClusters) > 0 {
-		authorities, err := readTrustedClusters(fc.Auth.TrustedClusters)
-		if err != nil {
+		if err := readTrustedClusters(fc.Auth.TrustedClusters, cfg); err != nil {
 			return trace.Wrap(err)
 		}
-		cfg.Auth.Authorities = append(cfg.Auth.Authorities, authorities...)
 	}
 	return nil
 }
 
-// readTrustedClusters takes a list of files, parses them and returns a list of CertAuthority
-// objects. Each file is exected to have an output of "tctl auth export"
-func readTrustedClusters(certFiles []string) (cas []services.CertAuthority, err error) {
-	if len(certFiles) == 0 {
-		return nil, nil
+// readTrustedClusters parses the content of "trusted_clusters" YAML structure
+// and modifies Teleport 'conf' by adding "authorities" and "reverse tunnels"
+// to it
+func readTrustedClusters(clusters []TrustedCluster, conf *service.Config) error {
+	if len(clusters) == 0 {
+		return nil
 	}
-	processCert := func(bytes []byte) error {
-		// is this a "host CA" key (known host)
+	// parseCAKey() gets called for every line in a "CA key file" which is
+	// the same as 'known_hosts' format for openssh
+	parseCAKey := func(bytes []byte) (*services.CertAuthority, error) {
 		marker, options, pubKey, comment, _, err := ssh.ParseKnownHosts(bytes)
 		if marker != "cert-authority" {
-			return trace.Errorf("invalid file format. expected '@cert-authority` marker")
+			return nil, trace.Errorf("invalid file format. expected '@cert-authority` marker")
 		}
 		if err != nil {
-			return trace.Errorf("invalid public key")
+			return nil, trace.Errorf("invalid public key")
 		}
 		teleportOpts, err := url.ParseQuery(comment)
 		if err != nil {
-			return trace.Errorf("invalid key comment: '%s'", comment)
+			return nil, trace.Errorf("invalid key comment: '%s'", comment)
 		}
 		authType := services.CertAuthType(teleportOpts.Get("type"))
 		if authType != services.HostCA && authType != services.UserCA {
-			return trace.Errorf("unsupported CA type: '%s'", authType)
+			return nil, trace.Errorf("unsupported CA type: '%s'", authType)
 		}
 		if len(options) == 0 {
-			return trace.Errorf("key without cluster_name")
+			return nil, trace.Errorf("key without cluster_name")
 		}
 		const prefix = "*."
 		domainName := strings.TrimPrefix(options[0], prefix)
-		ca := services.CertAuthority{
+		return &services.CertAuthority{
 			CheckingKeys: [][]byte{ssh.MarshalAuthorizedKey(pubKey)},
 			Type:         authType,
 			DomainName:   domainName,
-		}
-		cas = append(cas, ca)
-		return nil
+		}, nil
 	}
-
-	// go over all certificate files, find and process all certs:
-	for _, fp := range certFiles {
-		log.Debugf("reading 'trusted cluster' file %s", fp)
-		f, err := os.Open(fp)
+	// go over all trusted clusters:
+	for i := range clusters {
+		tc := &clusters[i]
+		// parse "allow_logins"
+		var allowedLogins []string
+		for _, login := range strings.Split(tc.AllowedLogins, ",") {
+			login = strings.TrimSpace(login)
+			if login != "" {
+				allowedLogins = append(allowedLogins, login)
+			}
+		}
+		// open the key file for this cluster:
+		log.Debugf("reading trusted cluster key file %s", tc.KeyFile)
+		f, err := os.Open(tc.KeyFile)
 		if err != nil {
-			return nil, trace.Errorf("error reading trusted cluster keys: %v", err)
+			return trace.Errorf("reading trusted cluster keys: %v", err)
 		}
 		defer f.Close()
+		// read the keyfile for this cluster and get trusted CA keys:
+		var authorities []services.CertAuthority
 		scanner := bufio.NewScanner(f)
-		// every line in a file is a cert:
 		for line := 0; scanner.Scan(); {
-			if err = processCert(scanner.Bytes()); err != nil {
-				return nil, trace.Errorf("%s:L%d. %v", fp, line, err)
+			ca, err := parseCAKey(scanner.Bytes())
+			if err != nil {
+				return trace.Errorf("%s:L%d. %v", tc.KeyFile, line, err)
+			}
+			if ca.Type == services.UserCA && len(allowedLogins) == 0 && len(tc.TunnelAddr) > 0 {
+				return trace.Errorf("trusted cluster '%s' needs allow_logins parameter",
+					ca.DomainName)
+			}
+			ca.AllowedLogins = allowedLogins
+			authorities = append(authorities, *ca)
+		}
+		conf.Auth.Authorities = append(conf.Auth.Authorities, authorities...)
+		clusterName := authorities[0].DomainName
+		// parse "tunnel_addr"
+		var tunnelAddresses []string
+		for _, ta := range strings.Split(tc.TunnelAddr, ",") {
+			ta := strings.TrimSpace(ta)
+			if ta == "" {
+				continue
+			}
+			addr, err := utils.ParseHostPortAddr(ta, defaults.SSHProxyTunnelListenPort)
+			if err != nil {
+				return trace.Wrap(err,
+					"Invalid tunnel address '%s' for cluster '%s'. Expect host:port format",
+					ta, clusterName)
 			}
+			tunnelAddresses = append(tunnelAddresses, addr.FullAddress())
+		}
+		if len(tunnelAddresses) > 0 {
+			conf.ReverseTunnels = append(conf.ReverseTunnels, services.ReverseTunnel{
+				DomainName: clusterName,
+				DialAddrs:  tunnelAddresses,
+			})
 		}
 	}
-	return cas, nil
+	return nil
 }
 
 // applyString takes 'src' and overwrites target with it, unless 'src' is empty
diff --git a/lib/config/fileconf.go b/lib/config/fileconf.go
index fe70054824c41..ac959a58bf1d5 100644
--- a/lib/config/fileconf.go
+++ b/lib/config/fileconf.go
@@ -382,7 +382,7 @@ type Auth struct {
 
 	// TrustedClustersFile is a file path to a file containing public CA keys
 	// of clusters we trust. One key per line, those starting with '#' are comments
-	TrustedClusters []string `yaml:"trusted_clusters,omitempty"`
+	TrustedClusters []TrustedCluster `yaml:"trusted_clusters,omitempty"`
 
 	// FOR INTERNAL USE:
 	// Authorities : 3rd party certificate authorities (CAs) this auth service trusts.
@@ -404,6 +404,17 @@ type Auth struct {
 	StaticTokens []StaticToken `yaml:"tokens,omitempty"`
 }
 
+// TrustedCluster struct holds configuration values under "trusted_clusters" key
+type TrustedCluster struct {
+	// KeyFile is a path to a remote authority (AKA "trusted cluster") public keys
+	KeyFile string `yaml:"keyfile,omitempty"`
+	// AllowedLogins is a comma-separated list of user logins allowed from that cluster
+	AllowedLogins string `yaml:"allow_logins,omitempty"`
+	// TunnelAddr is a comma-separated list of reverse tunnel addressess to
+	// connect to
+	TunnelAddr string `yaml:"tunnel_addr,omitempty"`
+}
+
 type StaticToken string
 
 // SSH is 'ssh_service' section of the config file
diff --git a/lib/srv/sshserver.go b/lib/srv/sshserver.go
index 56f60f1e47746..859fddd5ea88b 100644
--- a/lib/srv/sshserver.go
+++ b/lib/srv/sshserver.go
@@ -403,7 +403,7 @@ func (s *Server) checkPermissionToLogin(cert ssh.PublicKey, teleportUser, osUser
 
 	// for other authorities, check for authoritiy permissions
 	for _, login := range ca.AllowedLogins {
-		if login == osUser {
+		if login == osUser || login == "*" {
 			return nil
 		}
 	}
diff --git a/vagrant/libvirt/data/opt/a-auth.yaml b/vagrant/libvirt/data/opt/a-auth.yaml
index 1dd2d7846cc6b..06f47484143f6 100644
--- a/vagrant/libvirt/data/opt/a-auth.yaml
+++ b/vagrant/libvirt/data/opt/a-auth.yaml
@@ -9,10 +9,9 @@ auth_service:
   enabled: yes
   cluster_name: cluster-a
   trusted_clusters:
-    - /opt/shared/cluster-b
-  reverse_tunnels:
-    - domain_name: cluster-b
-      addresses: [b-auth]
+      - keyfile: /opt/shared/cluster-b
+        allow_logins: root,vagrant
+        tunnel_addr: b-auth
 
 ssh_service:
   enabled: yes
diff --git a/vagrant/libvirt/data/opt/b-auth.yaml b/vagrant/libvirt/data/opt/b-auth.yaml
index d531b91704660..7a0e481876d0e 100644
--- a/vagrant/libvirt/data/opt/b-auth.yaml
+++ b/vagrant/libvirt/data/opt/b-auth.yaml
@@ -9,7 +9,7 @@ auth_service:
   enabled: yes
   cluster_name: cluster-b
   trusted_clusters:
-    - /opt/shared/cluster-a
+      - keyfile: /opt/shared/cluster-b
 
 ssh_service:
   enabled: yes