1package cert 2 3import ( 4 "crypto/tls" 5 "crypto/x509" 6 "errors" 7 "fmt" 8 "log" 9 "net/url" 10 "path" 11 "reflect" 12 "strings" 13 "time" 14 15 "github.com/hashicorp/consul/api" 16) 17 18// ConsulSource implements a certificate source which loads 19// TLS and client authentication certificates from the consul KV store. 20// The CertURL/ClientCAURL must point to the base path of the certificates. 21// The TLS certificates are updated automatically when the KV store 22// changes. 23type ConsulSource struct { 24 CertURL string 25 ClientCAURL string 26 CAUpgradeCN string 27} 28 29func parseConsulURL(rawurl string) (config *api.Config, key string, err error) { 30 if rawurl == "" || !strings.HasPrefix(rawurl, "http://") && !strings.HasPrefix(rawurl, "https://") { 31 return nil, "", errors.New("invalid url") 32 } 33 34 u, err := url.Parse(rawurl) 35 if err != nil { 36 return nil, "", err 37 } 38 39 config = &api.Config{Address: u.Host, Scheme: u.Scheme} 40 if len(u.Query()["token"]) > 0 { 41 config.Token = u.Query()["token"][0] 42 } 43 44 // path needs to point to kv store and we need 45 // to strip the prefix off to get the key 46 const prefix = "/v1/kv/" 47 key = u.Path 48 if !strings.HasPrefix(key, prefix) { 49 return nil, "", errors.New("missing prefix: " + prefix) 50 } 51 key = key[len(prefix):] 52 return 53} 54 55func (s ConsulSource) LoadClientCAs() (*x509.CertPool, error) { 56 if s.ClientCAURL == "" { 57 return nil, nil 58 } 59 60 config, key, err := parseConsulURL(s.ClientCAURL) 61 if err != nil { 62 return nil, err 63 } 64 65 client, err := api.NewClient(config) 66 if err != nil { 67 return nil, err 68 } 69 70 load := func(key string) (map[string][]byte, error) { 71 pemBlocks, _, err := getCerts(client, key, 0) 72 return pemBlocks, err 73 } 74 return newCertPool(key, s.CAUpgradeCN, load) 75} 76 77func (s ConsulSource) Certificates() chan []tls.Certificate { 78 if s.CertURL == "" { 79 return nil 80 } 81 82 config, key, err := parseConsulURL(s.CertURL) 83 if err != nil { 84 log.Printf("[ERROR] cert: Failed to parse consul url. %s", err) 85 } 86 87 client, err := api.NewClient(config) 88 if err != nil { 89 log.Printf("[ERROR] cert: Failed to create consul client. %s", err) 90 } 91 92 pemBlocksCh := make(chan map[string][]byte, 1) 93 go watchKV(client, key, pemBlocksCh) 94 95 ch := make(chan []tls.Certificate, 1) 96 go func() { 97 for pemBlocks := range pemBlocksCh { 98 certs, err := loadCertificates(pemBlocks) 99 if err != nil { 100 log.Printf("[ERROR] cert: Failed to load certificates. %s", err) 101 continue 102 } 103 ch <- certs 104 } 105 }() 106 return ch 107} 108 109// watchKV monitors a key in the KV store for changes. 110func watchKV(client *api.Client, key string, pemBlocks chan map[string][]byte) { 111 var lastIndex uint64 112 var lastValue map[string][]byte 113 114 for { 115 value, index, err := getCerts(client, key, lastIndex) 116 if err != nil { 117 log.Printf("[WARN] cert: Error fetching certificates from %s. %v", key, err) 118 time.Sleep(time.Second) 119 continue 120 } 121 122 if !reflect.DeepEqual(value, lastValue) || index != lastIndex { 123 log.Printf("[DEBUG] cert: Certificate index changed to #%d", index) 124 pemBlocks <- value 125 lastValue, lastIndex = value, index 126 } 127 } 128} 129 130func getCerts(client *api.Client, key string, waitIndex uint64) (pemBlocks map[string][]byte, lastIndex uint64, err error) { 131 q := &api.QueryOptions{RequireConsistent: true, WaitIndex: waitIndex} 132 kvpairs, meta, err := client.KV().List(key, q) 133 if err != nil { 134 return nil, 0, fmt.Errorf("consul: list: %s", err) 135 } 136 if len(kvpairs) == 0 { 137 return nil, meta.LastIndex, nil 138 } 139 pemBlocks = map[string][]byte{} 140 for _, kvpair := range kvpairs { 141 pemBlocks[path.Base(kvpair.Key)] = kvpair.Value 142 } 143 return pemBlocks, meta.LastIndex, nil 144} 145