From 66d860bd7940a5f15fb3e9a7b37203b37b2e8db7 Mon Sep 17 00:00:00 2001 From: Joachim Bauch Date: Fri, 28 Aug 2020 17:03:32 +0200 Subject: [PATCH] Add support for fetching proxy tokens from etcd cluster. --- proxy.conf.in | 22 +++- src/proxy/proxy_server.go | 3 + src/proxy/proxy_tokens.go | 2 + src/proxy/proxy_tokens_etcd.go | 212 +++++++++++++++++++++++++++++++ src/proxy/proxy_tokens_static.go | 4 + 5 files changed, 242 insertions(+), 1 deletion(-) create mode 100644 src/proxy/proxy_tokens_etcd.go diff --git a/proxy.conf.in b/proxy.conf.in index 9aab79b..b2f9228 100644 --- a/proxy.conf.in +++ b/proxy.conf.in @@ -17,6 +17,7 @@ # # Possible values: # - static: A mapping of token id -> public key is configured below. +# - etcd: Token information are retrieved from an etcd cluster (see below). token_type = static [sessions] @@ -37,11 +38,30 @@ blockkey = -encryption-key- #url = nats://localhost:4222 [tokens] -# For token_type "static": Mapping of = of signaling +# For token type "static": Mapping of = of signaling # servers allowed to connect. #server1 = pubkey1.pem #server2 = pubkey2.pem +# For token type "etcd": Comma-separated list of static etcd endpoints to +# connect to. +#endpoints = 127.0.0.1:2379,127.0.0.1:22379,127.0.0.1:32379 + +# For token type "etcd": Options to perform endpoint discovery through DNS SRV. +# Only used if no endpoints are configured manually. +#discoverysrv = example.com +#discoveryservice = foo + +# For token type "etcd": Path to private key, client certificate and CA +# certificate if TLS authentication should be used. +#clientkey = /path/to/etcd-client.key +#clientcert = /path/to/etcd-client.crt +#cacert = /path/to/etcd-ca.crt + +# For token type "etcd": Format of key name to retrieve the public key from, +# "%s" will be replaced with the token id. +#keyformat = /signaling/proxy/tokens/%s/public-key + [mcu] # The type of the MCU to use. Currently only "janus" is supported. type = janus diff --git a/src/proxy/proxy_server.go b/src/proxy/proxy_server.go index b4e80af..8e16902 100644 --- a/src/proxy/proxy_server.go +++ b/src/proxy/proxy_server.go @@ -137,6 +137,8 @@ func NewProxyServer(r *mux.Router, version string, config *goconf.ConfigFile, na } switch tokenType { + case TokenTypeEtcd: + tokens, err = NewProxyTokensEtcd(config) case TokenTypeStatic: tokens, err = NewProxyTokensStatic(config) default: @@ -363,6 +365,7 @@ func (s *ProxyServer) Stop() { } s.mcu.Stop() + s.tokens.Close() } func (s *ProxyServer) ShutdownChannel() chan bool { diff --git a/src/proxy/proxy_tokens.go b/src/proxy/proxy_tokens.go index b0502d6..a5ed901 100644 --- a/src/proxy/proxy_tokens.go +++ b/src/proxy/proxy_tokens.go @@ -28,6 +28,7 @@ import ( ) const ( + TokenTypeEtcd = "etcd" TokenTypeStatic = "static" TokenTypeDefault = TokenTypeStatic @@ -42,4 +43,5 @@ type ProxyTokens interface { Get(id string) (*ProxyToken, error) Reload(config *goconf.ConfigFile) + Close() } diff --git a/src/proxy/proxy_tokens_etcd.go b/src/proxy/proxy_tokens_etcd.go new file mode 100644 index 0000000..cd1469b --- /dev/null +++ b/src/proxy/proxy_tokens_etcd.go @@ -0,0 +1,212 @@ +/** + * Standalone signaling server for the Nextcloud Spreed app. + * Copyright (C) 2020 struktur AG + * + * @author Joachim Bauch + * + * @license GNU AGPL version 3 or any later version + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package main + +import ( + "bytes" + "context" + "fmt" + "log" + "strings" + "sync/atomic" + "time" + + "github.com/dlintw/goconf" + + "go.etcd.io/etcd/clientv3" + "go.etcd.io/etcd/pkg/srv" + "go.etcd.io/etcd/pkg/transport" + + "gopkg.in/dgrijalva/jwt-go.v3" + + "signaling" +) + +const ( + tokenCacheSize = 4096 +) + +type tokenCacheEntry struct { + keyValue []byte + token *ProxyToken +} + +type tokensEtcd struct { + client atomic.Value + + tokenFormat atomic.Value + tokenCache *signaling.LruCache +} + +func NewProxyTokensEtcd(config *goconf.ConfigFile) (ProxyTokens, error) { + result := &tokensEtcd{ + tokenCache: signaling.NewLruCache(tokenCacheSize), + } + if err := result.load(config, false); err != nil { + return nil, err + } + + return result, nil +} + +func (t *tokensEtcd) getClient() *clientv3.Client { + c := t.client.Load() + if c == nil { + return nil + } + + return c.(*clientv3.Client) +} + +func (t *tokensEtcd) getKey(id string) string { + format := t.tokenFormat.Load().(string) + return fmt.Sprintf(format, id) +} + +func (t *tokensEtcd) Get(id string) (*ProxyToken, error) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + resp, err := t.getClient().Get(ctx, t.getKey(id)) + if err != nil { + return nil, err + } + + if len(resp.Kvs) == 0 { + return nil, nil + } else if len(resp.Kvs) > 1 { + log.Printf("Received multiple keys for %s, using last", id) + } + + keyValue := resp.Kvs[len(resp.Kvs)-1].Value + cached, _ := t.tokenCache.Get(id).(*tokenCacheEntry) + if cached == nil || !bytes.Equal(cached.keyValue, keyValue) { + // Parsed public keys are cached to avoid the parse overhead. + key, err := jwt.ParseRSAPublicKeyFromPEM(keyValue) + if err != nil { + return nil, fmt.Errorf("Could not parse public key for %s: %s", id, err) + } + + cached = &tokenCacheEntry{ + keyValue: keyValue, + token: &ProxyToken{ + id: id, + key: key, + }, + } + t.tokenCache.Set(id, cached) + } + + return cached.token, nil +} + +func (t *tokensEtcd) load(config *goconf.ConfigFile, ignoreErrors bool) error { + var endpoints []string + if endpointsString, _ := config.GetString("tokens", "endpoints"); endpointsString != "" { + for _, ep := range strings.Split(endpointsString, ",") { + ep := strings.TrimSpace(ep) + if ep != "" { + endpoints = append(endpoints, ep) + } + } + } else if discoverySrv, _ := config.GetString("tokens", "discoverysrv"); discoverySrv != "" { + discoveryService, _ := config.GetString("tokens", "discoveryservice") + clients, err := srv.GetClient("etcd-client", discoverySrv, discoveryService) + if err != nil { + if !ignoreErrors { + return fmt.Errorf("Could not discover endpoints for %s: %s", discoverySrv, err) + } + } else { + endpoints = clients.Endpoints + } + } + + if len(endpoints) == 0 { + if !ignoreErrors { + return fmt.Errorf("No token endpoints configured") + } + + log.Printf("No token endpoints configured, not changing client") + } else { + cfg := clientv3.Config{ + Endpoints: endpoints, + + // set timeout per request to fail fast when the target endpoint is unavailable + DialTimeout: time.Second, + } + + clientKey, _ := config.GetString("tokens", "clientkey") + clientCert, _ := config.GetString("tokens", "clientcert") + caCert, _ := config.GetString("tokens", "cacert") + if clientKey != "" && clientCert != "" && caCert != "" { + tlsInfo := transport.TLSInfo{ + CertFile: clientCert, + KeyFile: clientKey, + TrustedCAFile: caCert, + } + tlsConfig, err := tlsInfo.ClientConfig() + if err != nil { + if !ignoreErrors { + return fmt.Errorf("Could not setup TLS configuration: %s", err) + } + + log.Printf("Could not setup TLS configuration, will be disabled (%s)", err) + } else { + cfg.TLS = tlsConfig + } + } + + c, err := clientv3.New(cfg) + if err != nil { + if !ignoreErrors { + return err + } + + log.Printf("Could not create new client from token endpoints %+v: %s", endpoints, err) + } else { + prev := t.getClient() + if prev != nil { + prev.Close() + } + t.client.Store(c) + log.Printf("Using token endpoints %+v", endpoints) + } + } + + tokenFormat, _ := config.GetString("tokens", "keyformat") + if tokenFormat == "" { + tokenFormat = "/%s" + } + + t.tokenFormat.Store(tokenFormat) + log.Printf("Using %s as token format", tokenFormat) + return nil +} + +func (t *tokensEtcd) Reload(config *goconf.ConfigFile) { + t.load(config, true) +} + +func (t *tokensEtcd) Close() { + if client := t.getClient(); client != nil { + client.Close() + } +} diff --git a/src/proxy/proxy_tokens_static.go b/src/proxy/proxy_tokens_static.go index 3b8a984..8cbfa99 100644 --- a/src/proxy/proxy_tokens_static.go +++ b/src/proxy/proxy_tokens_static.go @@ -116,3 +116,7 @@ func (t *tokensStatic) load(config *goconf.ConfigFile, ignoreErrors bool) error func (t *tokensStatic) Reload(config *goconf.ConfigFile) { t.load(config, true) } + +func (t *tokensStatic) Close() { + t.setTokenKeys(map[string]*ProxyToken{}) +}