istio

1
// Copyright Istio Authors
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14

15
package trustbundle
16

17
import (
18
	"crypto/x509"
19
	"encoding/pem"
20
	"fmt"
21
	"sort"
22
	"strings"
23
	"sync"
24
	"time"
25

26
	meshconfig "istio.io/api/mesh/v1alpha1"
27
	"istio.io/istio/pkg/log"
28
	"istio.io/istio/pkg/slices"
29
	"istio.io/istio/pkg/spiffe"
30
	"istio.io/istio/pkg/util/sets"
31
)
32

33
// Source is all possible sources of MeshConfig
34
type Source int
35

36
const (
37
	SourceIstioCA Source = iota
38
	SourceMeshConfig
39
	SourceIstioRA
40
	sourceSpiffeEndpoints
41

42
	RemoteDefaultPollPeriod = 30 * time.Minute
43
)
44

45
func (s Source) String() string {
46
	switch s {
47
	case SourceIstioCA:
48
		return "IstioCA"
49
	case SourceMeshConfig:
50
		return "MeshConfig"
51
	case SourceIstioRA:
52
		return "IstioRA"
53
	case sourceSpiffeEndpoints:
54
		return "SpiffeEndpoints"
55
	default:
56
		return "Unknown"
57
	}
58
}
59

60
type TrustAnchorConfig struct {
61
	Certs []string
62
}
63

64
type TrustAnchorUpdate struct {
65
	TrustAnchorConfig
66
	Source Source
67
}
68

69
type TrustBundle struct {
70
	sourceConfig       map[Source]TrustAnchorConfig
71
	mutex              sync.RWMutex
72
	mergedCerts        []string
73
	updatecb           func()
74
	endpointMutex      sync.RWMutex
75
	endpoints          []string
76
	endpointUpdateChan chan struct{}
77
	remoteCaCertPool   *x509.CertPool
78
}
79

80
var (
81
	trustBundleLog = log.RegisterScope("trustBundle", "Workload mTLS trust bundle logs")
82
	remoteTimeout  = 10 * time.Second
83
)
84

85
// NewTrustBundle returns a new trustbundle
86
func NewTrustBundle(remoteCaCertPool *x509.CertPool) *TrustBundle {
87
	var err error
88
	tb := &TrustBundle{
89
		sourceConfig: map[Source]TrustAnchorConfig{
90
			SourceIstioCA:         {Certs: []string{}},
91
			SourceMeshConfig:      {Certs: []string{}},
92
			SourceIstioRA:         {Certs: []string{}},
93
			sourceSpiffeEndpoints: {Certs: []string{}},
94
		},
95
		mergedCerts:        []string{},
96
		updatecb:           nil,
97
		endpointUpdateChan: make(chan struct{}, 1),
98
		endpoints:          []string{},
99
	}
100
	if remoteCaCertPool == nil {
101
		tb.remoteCaCertPool, err = x509.SystemCertPool()
102
		if err != nil {
103
			trustBundleLog.Errorf("failed to initialize remote Cert pool: %v", err)
104
		}
105
	} else {
106
		tb.remoteCaCertPool = remoteCaCertPool
107
	}
108
	return tb
109
}
110

111
func (tb *TrustBundle) UpdateCb(updatecb func()) {
112
	tb.updatecb = updatecb
113
}
114

115
// GetTrustBundle : Retrieves all the trustAnchors for current Spiffee Trust Domain
116
func (tb *TrustBundle) GetTrustBundle() []string {
117
	tb.mutex.RLock()
118
	defer tb.mutex.RUnlock()
119
	trustedCerts := make([]string, len(tb.mergedCerts))
120
	copy(trustedCerts, tb.mergedCerts)
121
	return trustedCerts
122
}
123

124
func verifyTrustAnchor(trustAnchor string) error {
125
	block, _ := pem.Decode([]byte(trustAnchor))
126
	if block == nil {
127
		return fmt.Errorf("failed to decode pem certificate")
128
	}
129
	cert, err := x509.ParseCertificate(block.Bytes)
130
	if err != nil {
131
		return fmt.Errorf("failed to parse X.509 certificate: %v", err)
132
	}
133
	if !cert.IsCA {
134
		return fmt.Errorf("certificate is not a CA certificate")
135
	}
136
	return nil
137
}
138

139
func (tb *TrustBundle) mergeInternal() {
140
	var mergeCerts []string
141
	certMap := sets.New[string]()
142

143
	tb.mutex.Lock()
144
	defer tb.mutex.Unlock()
145

146
	for _, configSource := range tb.sourceConfig {
147
		for _, cert := range configSource.Certs {
148
			if !certMap.InsertContains(cert) {
149
				mergeCerts = append(mergeCerts, cert)
150
			}
151
		}
152
	}
153
	tb.mergedCerts = mergeCerts
154
	sort.Strings(tb.mergedCerts)
155
}
156

157
// UpdateTrustAnchor : External Function to merge a TrustAnchor config with the existing TrustBundle
158
func (tb *TrustBundle) UpdateTrustAnchor(anchorConfig *TrustAnchorUpdate) error {
159
	var ok bool
160
	var err error
161

162
	tb.mutex.RLock()
163
	cachedConfig, ok := tb.sourceConfig[anchorConfig.Source]
164
	tb.mutex.RUnlock()
165
	if !ok {
166
		return fmt.Errorf("invalid source of TrustBundle configuration %v", anchorConfig.Source)
167
	}
168

169
	// Check if anything needs to be changed at all
170
	if slices.Equal(anchorConfig.Certs, cachedConfig.Certs) {
171
		trustBundleLog.Debugf("no change to trustAnchor configuration after recent update")
172
		return nil
173
	}
174

175
	for _, cert := range anchorConfig.Certs {
176
		err = verifyTrustAnchor(cert)
177
		if err != nil {
178
			return err
179
		}
180
	}
181
	tb.mutex.Lock()
182
	tb.sourceConfig[anchorConfig.Source] = anchorConfig.TrustAnchorConfig
183
	tb.mutex.Unlock()
184
	tb.mergeInternal()
185

186
	trustBundleLog.Infof("updating Source %v with certs %v",
187
		anchorConfig.Source,
188
		strings.Join(anchorConfig.TrustAnchorConfig.Certs, "\n"))
189

190
	if tb.updatecb != nil {
191
		tb.updatecb()
192
	}
193
	return nil
194
}
195

196
func (tb *TrustBundle) updateRemoteEndpoint(spiffeEndpoints []string) {
197
	tb.endpointMutex.RLock()
198
	remoteEndpoints := tb.endpoints
199
	tb.endpointMutex.RUnlock()
200

201
	if slices.Equal(spiffeEndpoints, remoteEndpoints) {
202
		return
203
	}
204
	trustBundleLog.Infof("updated remote endpoints  :%v", spiffeEndpoints)
205
	tb.endpointMutex.Lock()
206
	tb.endpoints = spiffeEndpoints
207
	tb.endpointMutex.Unlock()
208
	tb.endpointUpdateChan <- struct{}{}
209
}
210

211
// AddMeshConfigUpdate : Update trustAnchor configurations from meshConfig
212
func (tb *TrustBundle) AddMeshConfigUpdate(cfg *meshconfig.MeshConfig) error {
213
	var err error
214
	if cfg != nil {
215
		certs := []string{}
216
		endpoints := []string{}
217
		for _, pemCert := range cfg.GetCaCertificates() {
218
			cert := pemCert.GetPem()
219
			if cert != "" {
220
				certs = append(certs, cert)
221
			} else if pemCert.GetSpiffeBundleUrl() != "" {
222
				endpoints = append(endpoints, pemCert.GetSpiffeBundleUrl())
223
			}
224
		}
225

226
		err = tb.UpdateTrustAnchor(&TrustAnchorUpdate{
227
			TrustAnchorConfig: TrustAnchorConfig{Certs: certs},
228
			Source:            SourceMeshConfig,
229
		})
230
		if err != nil {
231
			trustBundleLog.Errorf("failed to update meshConfig PEM trustAnchors: %v", err)
232
			return err
233
		}
234

235
		tb.updateRemoteEndpoint(endpoints)
236
	}
237
	return nil
238
}
239

240
func (tb *TrustBundle) fetchRemoteTrustAnchors() {
241
	var err error
242

243
	tb.endpointMutex.RLock()
244
	remoteEndpoints := tb.endpoints
245
	tb.endpointMutex.RUnlock()
246
	remoteCerts := []string{}
247

248
	currentTrustDomain := spiffe.GetTrustDomain()
249
	for _, endpoint := range remoteEndpoints {
250
		trustDomainAnchorMap, err := spiffe.RetrieveSpiffeBundleRootCerts(
251
			map[string]string{currentTrustDomain: endpoint}, tb.remoteCaCertPool, remoteTimeout)
252
		if err != nil {
253
			trustBundleLog.Errorf("unable to fetch trust Anchors from endpoint %s: %s", endpoint, err)
254
			continue
255
		}
256
		certs := trustDomainAnchorMap[currentTrustDomain]
257
		for _, cert := range certs {
258
			certStr := string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert.Raw}))
259
			trustBundleLog.Debugf("from endpoint %v, fetched trust anchor cert: %v", endpoint, certStr)
260
			remoteCerts = append(remoteCerts, certStr)
261
		}
262
	}
263
	err = tb.UpdateTrustAnchor(&TrustAnchorUpdate{
264
		TrustAnchorConfig: TrustAnchorConfig{Certs: remoteCerts},
265
		Source:            sourceSpiffeEndpoints,
266
	})
267
	if err != nil {
268
		trustBundleLog.Errorf("failed to update meshConfig Spiffe trustAnchors: %v", err)
269
	}
270
}
271

272
func (tb *TrustBundle) ProcessRemoteTrustAnchors(stop <-chan struct{}, pollInterval time.Duration) {
273
	ticker := time.NewTicker(pollInterval)
274
	defer ticker.Stop()
275
	for {
276
		select {
277
		case <-ticker.C:
278
			trustBundleLog.Infof("waking up to perform periodic checks")
279
			tb.fetchRemoteTrustAnchors()
280
		case <-stop:
281
			trustBundleLog.Infof("stop processing endpoint trustAnchor updates")
282
			return
283
		case <-tb.endpointUpdateChan:
284
			tb.fetchRemoteTrustAnchors()
285
			trustBundleLog.Infof("processing endpoint trustAnchor Updates for config change")
286
		}
287
	}
288
}
289