forked from kubernetes/kubernetes
-
Notifications
You must be signed in to change notification settings - Fork 1
/
aws_credentials.go
311 lines (271 loc) · 9.31 KB
/
aws_credentials.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
/*
Copyright 2014 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package credentials
import (
"encoding/base64"
"errors"
"fmt"
"net/url"
"regexp"
"strings"
"sync"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/request"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/ecr"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/tools/cache"
"k8s.io/klog"
"k8s.io/kubernetes/pkg/credentialprovider"
"k8s.io/kubernetes/pkg/version"
)
var ecrPattern = regexp.MustCompile(`^(\d{12})\.dkr\.ecr(\-fips)?\.([a-zA-Z0-9][a-zA-Z0-9-_]*)\.amazonaws\.com(\.cn)?$`)
// init registers a credential provider for each registryURLTemplate and creates
// an ECR token getter factory with a new cache to store token getters
func init() {
credentialprovider.RegisterCredentialProvider("amazon-ecr",
newECRProvider(&ecrTokenGetterFactory{cache: make(map[string]tokenGetter)}))
}
// ecrProvider is a DockerConfigProvider that gets and refreshes tokens
// from AWS to access ECR.
type ecrProvider struct {
cache cache.Store
getterFactory tokenGetterFactory
}
var _ credentialprovider.DockerConfigProvider = &ecrProvider{}
func newECRProvider(getterFactory tokenGetterFactory) *ecrProvider {
return &ecrProvider{
cache: cache.NewExpirationStore(stringKeyFunc, &ecrExpirationPolicy{}),
getterFactory: getterFactory,
}
}
// Enabled implements DockerConfigProvider.Enabled. Enabled is true if AWS
// credentials are found.
func (p *ecrProvider) Enabled() bool {
sess, err := session.NewSessionWithOptions(session.Options{
SharedConfigState: session.SharedConfigEnable,
})
if err != nil {
klog.Errorf("while validating AWS credentials %v", err)
return false
}
if _, err := sess.Config.Credentials.Get(); err != nil {
klog.Errorf("while getting AWS credentials %v", err)
return false
}
return true
}
// LazyProvide is lazy
// TODO: the LazyProvide methods will be removed in a future PR
func (p *ecrProvider) LazyProvide(image string) *credentialprovider.DockerConfigEntry {
return nil
}
// Provide returns a DockerConfig with credentials from the cache if they are
// found, or from ECR
func (p *ecrProvider) Provide(image string) credentialprovider.DockerConfig {
parsed, err := parseRepoURL(image)
if err != nil {
klog.V(3).Info(err)
return credentialprovider.DockerConfig{}
}
if cfg, exists := p.getFromCache(parsed); exists {
klog.V(6).Infof("Got ECR credentials from cache for %s", parsed.registry)
return cfg
}
klog.V(3).Info("unable to get ECR credentials from cache, checking ECR API")
cfg, err := p.getFromECR(parsed)
if err != nil {
klog.Errorf("error getting credentials from ECR for %s %v", parsed.registry, err)
return credentialprovider.DockerConfig{}
}
klog.V(3).Infof("Got ECR credentials from ECR API for %s", parsed.registry)
return cfg
}
// getFromCache attempts to get credentials from the cache
func (p *ecrProvider) getFromCache(parsed *parsedURL) (credentialprovider.DockerConfig, bool) {
cfg := credentialprovider.DockerConfig{}
obj, exists, err := p.cache.GetByKey(parsed.registry)
if err != nil {
klog.Errorf("error getting ECR credentials from cache: %v", err)
return cfg, false
}
if !exists {
return cfg, false
}
entry := obj.(*cacheEntry)
cfg[entry.registry] = entry.credentials
return cfg, true
}
// getFromECR gets credentials from ECR since they are not in the cache
func (p *ecrProvider) getFromECR(parsed *parsedURL) (credentialprovider.DockerConfig, error) {
cfg := credentialprovider.DockerConfig{}
getter, err := p.getterFactory.GetTokenGetterForRegion(parsed.region)
if err != nil {
return cfg, err
}
params := &ecr.GetAuthorizationTokenInput{RegistryIds: []*string{aws.String(parsed.registryID)}}
output, err := getter.GetAuthorizationToken(params)
if err != nil {
return cfg, err
}
if output == nil {
return cfg, errors.New("authorization token is nil")
}
if len(output.AuthorizationData) == 0 {
return cfg, errors.New("authorization data from response is empty")
}
data := output.AuthorizationData[0]
if data.AuthorizationToken == nil {
return cfg, errors.New("authorization token in response is nil")
}
entry, err := makeCacheEntry(data, parsed.registry)
if err != nil {
return cfg, err
}
if err := p.cache.Add(entry); err != nil {
return cfg, err
}
cfg[entry.registry] = entry.credentials
return cfg, nil
}
type parsedURL struct {
registryID string
region string
registry string
}
// parseRepoURL parses and splits the registry URL into the registry ID,
// region, and registry.
// <registryID>.dkr.ecr(-fips).<region>.amazonaws.com(.cn)
func parseRepoURL(image string) (*parsedURL, error) {
parsed, err := url.Parse("https://" + image)
if err != nil {
return nil, fmt.Errorf("error parsing image %s %v", image, err)
}
splitURL := ecrPattern.FindStringSubmatch(parsed.Hostname())
if len(splitURL) == 0 {
return nil, fmt.Errorf("%s is not a valid ECR repository URL", parsed.Hostname())
}
return &parsedURL{
registryID: splitURL[1],
region: splitURL[3],
registry: parsed.Hostname(),
}, nil
}
// tokenGetter is for testing purposes
type tokenGetter interface {
GetAuthorizationToken(input *ecr.GetAuthorizationTokenInput) (*ecr.GetAuthorizationTokenOutput, error)
}
// tokenGetterFactory is for testing purposes
type tokenGetterFactory interface {
GetTokenGetterForRegion(string) (tokenGetter, error)
}
// ecrTokenGetterFactory stores a token getter per region
type ecrTokenGetterFactory struct {
cache map[string]tokenGetter
mutex sync.Mutex
}
// awsHandlerLogger is a handler that logs all AWS SDK requests
// Copied from pkg/cloudprovider/providers/aws/log_handler.go
func awsHandlerLogger(req *request.Request) {
service := req.ClientInfo.ServiceName
region := req.Config.Region
name := "?"
if req.Operation != nil {
name = req.Operation.Name
}
klog.V(3).Infof("AWS request: %s:%s in %s", service, name, *region)
}
func newECRTokenGetter(region string) (tokenGetter, error) {
sess, err := session.NewSessionWithOptions(session.Options{
Config: aws.Config{Region: aws.String(region)},
SharedConfigState: session.SharedConfigEnable,
})
if err != nil {
return nil, err
}
getter := &ecrTokenGetter{svc: ecr.New(sess)}
getter.svc.Handlers.Build.PushFrontNamed(request.NamedHandler{
Name: "k8s/user-agent",
Fn: request.MakeAddToUserAgentHandler("kubernetes", version.Get().String()),
})
getter.svc.Handlers.Sign.PushFrontNamed(request.NamedHandler{
Name: "k8s/logger",
Fn: awsHandlerLogger,
})
return getter, nil
}
// GetTokenGetterForRegion gets the token getter for the requested region. If it
// doesn't exist, it creates a new ECR token getter
func (f *ecrTokenGetterFactory) GetTokenGetterForRegion(region string) (tokenGetter, error) {
f.mutex.Lock()
defer f.mutex.Unlock()
if getter, ok := f.cache[region]; ok {
return getter, nil
}
getter, err := newECRTokenGetter(region)
if err != nil {
return nil, fmt.Errorf("unable to create token getter for region %v %v", region, err)
}
f.cache[region] = getter
return getter, nil
}
// The canonical implementation
type ecrTokenGetter struct {
svc *ecr.ECR
}
// GetAuthorizationToken gets the ECR authorization token using the ECR API
func (p *ecrTokenGetter) GetAuthorizationToken(input *ecr.GetAuthorizationTokenInput) (*ecr.GetAuthorizationTokenOutput, error) {
return p.svc.GetAuthorizationToken(input)
}
type cacheEntry struct {
expiresAt time.Time
credentials credentialprovider.DockerConfigEntry
registry string
}
// makeCacheEntry decodes the ECR authorization entry and re-packages it into a
// cacheEntry.
func makeCacheEntry(data *ecr.AuthorizationData, registry string) (*cacheEntry, error) {
decodedToken, err := base64.StdEncoding.DecodeString(aws.StringValue(data.AuthorizationToken))
if err != nil {
return nil, fmt.Errorf("error decoding ECR authorization token: %v", err)
}
parts := strings.SplitN(string(decodedToken), ":", 2)
if len(parts) < 2 {
return nil, errors.New("error getting username and password from authorization token")
}
creds := credentialprovider.DockerConfigEntry{
Username: parts[0],
Password: parts[1],
Email: "not@val.id", // ECR doesn't care and Docker is about to obsolete it
}
if data.ExpiresAt == nil {
return nil, errors.New("authorization data expiresAt is nil")
}
return &cacheEntry{
expiresAt: data.ExpiresAt.Add(-1 * wait.Jitter(30*time.Minute, 0.2)),
credentials: creds,
registry: registry,
}, nil
}
// ecrExpirationPolicy implements ExpirationPolicy from client-go.
type ecrExpirationPolicy struct{}
// stringKeyFunc returns the cache key as a string
func stringKeyFunc(obj interface{}) (string, error) {
key := obj.(*cacheEntry).registry
return key, nil
}
// IsExpired checks if the ECR credentials are expired.
func (p *ecrExpirationPolicy) IsExpired(entry *cache.TimestampedEntry) bool {
return time.Now().After(entry.Obj.(*cacheEntry).expiresAt)
}