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