forked from hashicorp/consul
/
acl.go
1541 lines (1313 loc) · 45.9 KB
/
acl.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
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
package consul
import (
"fmt"
"log"
"os"
"sync"
"time"
"github.com/armon/go-metrics"
"github.com/hashicorp/consul/acl"
"github.com/hashicorp/consul/agent/structs"
"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/sentinel"
"golang.org/x/time/rate"
)
// These must be kept in sync with the constants in command/agent/acl.go.
const (
// anonymousToken is the token ID we re-write to if there is no token ID
// provided.
anonymousToken = "anonymous"
// redactedToken is shown in structures with embedded tokens when they
// are not allowed to be displayed.
redactedToken = "<hidden>"
// aclUpgradeBatchSize controls how many tokens we look at during each round of upgrading. Individual raft logs
// will be further capped using the aclBatchUpsertSize. This limit just prevents us from creating a single slice
// with all tokens in it.
aclUpgradeBatchSize = 128
// aclUpgradeRateLimit is the number of batch upgrade requests per second.
aclUpgradeRateLimit rate.Limit = 1.0
// aclBatchDeleteSize is the number of deletions to send in a single batch operation. 4096 should produce a batch that is <150KB
// in size but should be sufficiently large to handle 1 replication round in a single batch
aclBatchDeleteSize = 4096
// aclBatchUpsertSize is the target size in bytes we want to submit for a batch upsert request. We estimate the size at runtime
// due to the data being more variable in its size.
aclBatchUpsertSize = 256 * 1024
// DEPRECATED (ACL-Legacy-Compat) aclModeCheck* are all only for legacy usage
// aclModeCheckMinInterval is the minimum amount of time between checking if the
// agent should be using the new or legacy ACL system. All the places it is
// currently used will backoff as it detects that it is remaining in legacy mode.
// However the initial min value is kept small so that new cluster creation
// can enter into new ACL mode quickly.
aclModeCheckMinInterval = 50 * time.Millisecond
// aclModeCheckMaxInterval controls the maximum interval for how often the agent
// checks if it should be using the new or legacy ACL system.
aclModeCheckMaxInterval = 30 * time.Second
// Maximum number of re-resolution requests to be made if the token is modified between
// resolving the token and resolving its policies that would remove one of its policies.
tokenPolicyResolutionMaxRetries = 5
)
func minTTL(a time.Duration, b time.Duration) time.Duration {
if a < b {
return a
}
return b
}
type ACLRemoteError struct {
Err error
}
func (e ACLRemoteError) Error() string {
return fmt.Sprintf("Error communicating with the ACL Datacenter: %v", e.Err)
}
func IsACLRemoteError(err error) bool {
_, ok := err.(ACLRemoteError)
return ok
}
type ACLResolverDelegate interface {
ACLsEnabled() bool
ACLDatacenter(legacy bool) string
// UseLegacyACLs
UseLegacyACLs() bool
ResolveIdentityFromToken(token string) (bool, structs.ACLIdentity, error)
ResolvePolicyFromID(policyID string) (bool, *structs.ACLPolicy, error)
RPC(method string, args interface{}, reply interface{}) error
}
type remoteACLLegacyResult struct {
authorizer acl.Authorizer
err error
}
type remoteACLIdentityResult struct {
identity structs.ACLIdentity
err error
}
type remoteACLPolicyResult struct {
policy *structs.ACLPolicy
err error
}
type policyTokenError struct {
Err error
token string
}
func (e policyTokenError) Error() string {
return e.Err.Error()
}
// ACLResolverConfig holds all the configuration necessary to create an ACLResolver
type ACLResolverConfig struct {
Config *Config
Logger *log.Logger
// CacheConfig is a pass through configuration for ACL cache limits
CacheConfig *structs.ACLCachesConfig
// Delegate that implements some helper functionality that is server/client specific
Delegate ACLResolverDelegate
// AutoDisable indicates that RPC responses should be checked and if they indicate ACLs are disabled
// remotely then disable them locally as well. This is particularly useful for the client agent
// so that it can detect when the servers have gotten ACLs enabled.
AutoDisable bool
Sentinel sentinel.Evaluator
}
// ACLResolver is the type to handle all your token and policy resolution needs.
//
// Supports:
// - Resolving tokens locally via the ACLResolverDelegate
// - Resolving policies locally via the ACLResolverDelegate
// - Resolving legacy tokens remotely via a ACL.GetPolicy RPC
// - Resolving tokens remotely via an ACL.TokenRead RPC
// - Resolving policies remotely via an ACL.PolicyResolve RPC
//
// Remote Resolution:
// Remote resolution can be done syncrhonously or asynchronously depending
// on the ACLDownPolicy in the Config passed to the resolver.
//
// When the down policy is set to async-cache and we have already cached values
// then go routines will be spawned to perform the RPCs in the background
// and then will udpate the cache with either the positive or negative result.
//
// When the down policy is set to extend-cache or the token/policy is not already
// cached then the same go routines are spawned to do the RPCs in the background.
// However in this mode channels are created to receive the results of the RPC
// and are registered with the resolver. Those channels are immediately read/blocked
// upon.
//
type ACLResolver struct {
config *Config
logger *log.Logger
delegate ACLResolverDelegate
sentinel sentinel.Evaluator
cache *structs.ACLCaches
asyncIdentityResults map[string][]chan (*remoteACLIdentityResult)
asyncIdentityResultsMutex sync.RWMutex
asyncPolicyResults map[string][]chan (*remoteACLPolicyResult)
asyncPolicyResultsMutex sync.RWMutex
asyncLegacyResults map[string][]chan (*remoteACLLegacyResult)
asyncLegacyMutex sync.RWMutex
down acl.Authorizer
autoDisable bool
disabled time.Time
disabledLock sync.RWMutex
}
func NewACLResolver(config *ACLResolverConfig) (*ACLResolver, error) {
if config == nil {
return nil, fmt.Errorf("ACL Resolver must be initialized with a config")
}
if config.Config == nil {
return nil, fmt.Errorf("ACLResolverConfig.Config must not be nil")
}
if config.Delegate == nil {
return nil, fmt.Errorf("ACL Resolver must be initialized with a valid delegate")
}
if config.Logger == nil {
config.Logger = log.New(os.Stderr, "", log.LstdFlags)
}
cache, err := structs.NewACLCaches(config.CacheConfig)
if err != nil {
return nil, err
}
var down acl.Authorizer
switch config.Config.ACLDownPolicy {
case "allow":
down = acl.AllowAll()
case "deny":
down = acl.DenyAll()
case "async-cache", "extend-cache":
// Leave the down policy as nil to signal this.
default:
return nil, fmt.Errorf("invalid ACL down policy %q", config.Config.ACLDownPolicy)
}
return &ACLResolver{
config: config.Config,
logger: config.Logger,
delegate: config.Delegate,
sentinel: config.Sentinel,
cache: cache,
asyncIdentityResults: make(map[string][]chan (*remoteACLIdentityResult)),
asyncPolicyResults: make(map[string][]chan (*remoteACLPolicyResult)),
asyncLegacyResults: make(map[string][]chan (*remoteACLLegacyResult)),
autoDisable: config.AutoDisable,
down: down,
}, nil
}
// fireAsyncLegacyResult is used to notify any watchers that legacy resolution of a token is complete
func (r *ACLResolver) fireAsyncLegacyResult(token string, authorizer acl.Authorizer, ttl time.Duration, err error) {
// cache the result: positive or negative
r.cache.PutAuthorizerWithTTL(token, authorizer, ttl)
// get the list of channels to send the result to
r.asyncLegacyMutex.Lock()
channels := r.asyncLegacyResults[token]
delete(r.asyncLegacyResults, token)
r.asyncLegacyMutex.Unlock()
// notify all watchers of the RPC results
result := &remoteACLLegacyResult{authorizer, err}
for _, cx := range channels {
// only chans that are being blocked on will be in the list of channels so this cannot block
cx <- result
close(cx)
}
}
func (r *ACLResolver) resolveTokenLegacyAsync(token string, cached *structs.AuthorizerCacheEntry) {
req := structs.ACLPolicyResolveLegacyRequest{
Datacenter: r.delegate.ACLDatacenter(true),
ACL: token,
}
cacheTTL := r.config.ACLTokenTTL
if cached != nil {
cacheTTL = cached.TTL
}
var reply structs.ACLPolicyResolveLegacyResponse
err := r.delegate.RPC("ACL.GetPolicy", &req, &reply)
if err == nil {
parent := acl.RootAuthorizer(reply.Parent)
if parent == nil {
r.fireAsyncLegacyResult(token, cached.Authorizer, cacheTTL, acl.ErrInvalidParent)
return
}
var policies []*acl.Policy
policy := reply.Policy
if policy != nil {
policies = append(policies, policy.ConvertFromLegacy())
}
authorizer, err := acl.NewPolicyAuthorizer(parent, policies, r.sentinel)
r.fireAsyncLegacyResult(token, authorizer, reply.TTL, err)
return
}
if acl.IsErrNotFound(err) {
// Make sure to remove from the cache if it was deleted
r.fireAsyncLegacyResult(token, nil, cacheTTL, acl.ErrNotFound)
return
}
// some other RPC error
switch r.config.ACLDownPolicy {
case "allow":
r.fireAsyncLegacyResult(token, acl.AllowAll(), cacheTTL, nil)
return
case "async-cache", "extend-cache":
if cached != nil {
r.fireAsyncLegacyResult(token, cached.Authorizer, cacheTTL, nil)
return
}
fallthrough
default:
r.fireAsyncLegacyResult(token, acl.DenyAll(), cacheTTL, nil)
return
}
}
func (r *ACLResolver) resolveTokenLegacy(token string) (acl.Authorizer, error) {
defer metrics.MeasureSince([]string{"acl", "resolveTokenLegacy"}, time.Now())
// Attempt to resolve locally first (local results are not cached)
// This is only useful for servers where either legacy replication is being
// done or the server is within the primary datacenter.
if done, identity, err := r.delegate.ResolveIdentityFromToken(token); done {
if err == nil && identity != nil {
policies, err := r.resolvePoliciesForIdentity(identity)
if err != nil {
return nil, err
}
return policies.Compile(acl.RootAuthorizer(r.config.ACLDefaultPolicy), r.cache, r.sentinel)
}
return nil, err
}
// Look in the cache prior to making a RPC request
entry := r.cache.GetAuthorizer(token)
if entry != nil && entry.Age() <= minTTL(entry.TTL, r.config.ACLTokenTTL) {
metrics.IncrCounter([]string{"acl", "token", "cache_hit"}, 1)
if entry.Authorizer != nil {
return entry.Authorizer, nil
}
return nil, acl.ErrNotFound
}
metrics.IncrCounter([]string{"acl", "token", "cache_miss"}, 1)
// Resolve the token in the background and wait on the result if we must
var waitChan chan *remoteACLLegacyResult
waitForResult := entry == nil || r.config.ACLDownPolicy != "async-cache"
r.asyncLegacyMutex.Lock()
// check if resolution for this token is already happening
waiters, ok := r.asyncLegacyResults[token]
if !ok || waiters == nil {
// initialize the slice of waiters if not already done
waiters = make([]chan *remoteACLLegacyResult, 0)
}
if waitForResult {
// create the waitChan only if we are going to block waiting
// for the response and then append it to the list of waiters
// Because we will block (not select or discard this chan) we
// do not need to create it as buffered
waitChan = make(chan *remoteACLLegacyResult)
r.asyncLegacyResults[token] = append(waiters, waitChan)
}
r.asyncLegacyMutex.Unlock()
if !ok {
// start the async RPC if it wasn't already ongoing
go r.resolveTokenLegacyAsync(token, entry)
}
if !waitForResult {
// waitForResult being false requires the cacheEntry to not be nil
if entry.Authorizer != nil {
return entry.Authorizer, nil
}
return nil, acl.ErrNotFound
}
// block waiting for the async RPC to finish.
res := <-waitChan
return res.authorizer, res.err
}
// fireAsyncTokenResult is used to notify all waiters that the results of a token resolution is complete
func (r *ACLResolver) fireAsyncTokenResult(token string, identity structs.ACLIdentity, err error) {
// cache the result: positive or negative
r.cache.PutIdentity(token, identity)
// get the list of channels to send the result to
r.asyncIdentityResultsMutex.Lock()
channels := r.asyncIdentityResults[token]
delete(r.asyncIdentityResults, token)
r.asyncIdentityResultsMutex.Unlock()
// notify all watchers of the RPC results
result := &remoteACLIdentityResult{identity, err}
for _, cx := range channels {
// cannot block because all wait chans will have another goroutine blocked on the read
cx <- result
close(cx)
}
}
func (r *ACLResolver) resolveIdentityFromTokenAsync(token string, cached *structs.IdentityCacheEntry) {
req := structs.ACLTokenGetRequest{
Datacenter: r.delegate.ACLDatacenter(false),
TokenID: token,
TokenIDType: structs.ACLTokenSecret,
QueryOptions: structs.QueryOptions{
Token: token,
AllowStale: true,
},
}
var resp structs.ACLTokenResponse
err := r.delegate.RPC("ACL.TokenRead", &req, &resp)
if err == nil {
if resp.Token == nil {
r.fireAsyncTokenResult(token, nil, acl.ErrNotFound)
} else {
r.fireAsyncTokenResult(token, resp.Token, nil)
}
return
}
if acl.IsErrNotFound(err) {
// Make sure to remove from the cache if it was deleted
r.fireAsyncTokenResult(token, nil, acl.ErrNotFound)
return
}
// some other RPC error
if cached != nil && (r.config.ACLDownPolicy == "extend-cache" || r.config.ACLDownPolicy == "async-cache") {
// extend the cache
r.fireAsyncTokenResult(token, cached.Identity, nil)
}
r.fireAsyncTokenResult(token, nil, err)
return
}
func (r *ACLResolver) resolveIdentityFromToken(token string) (structs.ACLIdentity, error) {
// Attempt to resolve locally first (local results are not cached)
if done, identity, err := r.delegate.ResolveIdentityFromToken(token); done {
return identity, err
}
// Check the cache before making any RPC requests
cacheEntry := r.cache.GetIdentity(token)
if cacheEntry != nil && cacheEntry.Age() <= r.config.ACLTokenTTL {
metrics.IncrCounter([]string{"acl", "token", "cache_hit"}, 1)
return cacheEntry.Identity, nil
}
metrics.IncrCounter([]string{"acl", "token", "cache_miss"}, 1)
// Background a RPC request and wait on it if we must
var waitChan chan *remoteACLIdentityResult
waitForResult := cacheEntry == nil || r.config.ACLDownPolicy != "async-cache"
r.asyncIdentityResultsMutex.Lock()
// check if resolution of this token is already ongoing
waiters, ok := r.asyncIdentityResults[token]
if !ok || waiters == nil {
// only initialize the slice of waiters if need be (when this token resolution isn't ongoing)
waiters = make([]chan *remoteACLIdentityResult, 0)
}
if waitForResult {
// create the waitChan only if we are going to block waiting
// for the response and then append it to the list of waiters
// Because we will block (not select or discard this chan) we
// do not need to create it as buffered
waitChan = make(chan *remoteACLIdentityResult)
r.asyncIdentityResults[token] = append(waiters, waitChan)
}
r.asyncIdentityResultsMutex.Unlock()
if !ok {
// only start the RPC if one isn't in flight
go r.resolveIdentityFromTokenAsync(token, cacheEntry)
}
if !waitForResult {
// waitForResult being false requires the cacheEntry to not be nil
return cacheEntry.Identity, nil
}
// block on the read here, this is why we don't need chan buffering
res := <-waitChan
if res.err != nil && !acl.IsErrNotFound(res.err) {
return res.identity, ACLRemoteError{Err: res.err}
}
return res.identity, res.err
}
// fireAsyncPolicyResult is used to notify all waiters that policy resolution is complete.
func (r *ACLResolver) fireAsyncPolicyResult(policyID string, policy *structs.ACLPolicy, err error, updateCache bool) {
if updateCache {
// cache the result: positive or negative
r.cache.PutPolicy(policyID, policy)
}
// get the list of channels to send the result to
r.asyncPolicyResultsMutex.Lock()
channels := r.asyncPolicyResults[policyID]
delete(r.asyncPolicyResults, policyID)
r.asyncPolicyResultsMutex.Unlock()
// notify all watchers of the RPC results
result := &remoteACLPolicyResult{policy, err}
for _, cx := range channels {
// not closing the channel as there could be more events to be fired.
cx <- result
}
}
func (r *ACLResolver) resolvePoliciesAsyncForIdentity(identity structs.ACLIdentity, policyIDs []string, cached map[string]*structs.PolicyCacheEntry) {
req := structs.ACLPolicyBatchGetRequest{
Datacenter: r.delegate.ACLDatacenter(false),
PolicyIDs: policyIDs,
QueryOptions: structs.QueryOptions{
Token: identity.SecretToken(),
AllowStale: true,
},
}
found := make(map[string]struct{})
var resp structs.ACLPolicyBatchResponse
err := r.delegate.RPC("ACL.PolicyResolve", &req, &resp)
if err == nil {
for _, policy := range resp.Policies {
r.fireAsyncPolicyResult(policy.ID, policy, nil, true)
found[policy.ID] = struct{}{}
}
for _, policyID := range policyIDs {
if _, ok := found[policyID]; !ok {
r.fireAsyncPolicyResult(policyID, nil, acl.ErrNotFound, true)
}
}
return
}
if acl.IsErrNotFound(err) {
// make sure to indicate that this identity is no longer valid within
// the cache
//
// Note - This must be done before firing the results or else it would
// be possible for waiters to get woken up an get the cached identity
// again
r.cache.PutIdentity(identity.SecretToken(), nil)
for _, policyID := range policyIDs {
// Do not touch the cache. Getting a top level ACL not found error
// only indicates that the secret token used in the request
// no longer exists
r.fireAsyncPolicyResult(policyID, nil, &policyTokenError{acl.ErrNotFound, identity.SecretToken()}, false)
}
return
}
if acl.IsErrPermissionDenied(err) {
// invalidate our ID cache so that identity resolution will take place
// again in the future
//
// Note - This must be done before firing the results or else it would
// be possible for waiters to get woken up and get the cached identity
// again
r.cache.RemoveIdentity(identity.SecretToken())
for _, policyID := range policyIDs {
// Do not remove from the cache for permission denied
// what this does indicate is that our view of the token is out of date
r.fireAsyncPolicyResult(policyID, nil, &policyTokenError{acl.ErrPermissionDenied, identity.SecretToken()}, false)
}
return
}
// other RPC error - use cache if available
extendCache := r.config.ACLDownPolicy == "extend-cache" || r.config.ACLDownPolicy == "async-cache"
for _, policyID := range policyIDs {
if entry, ok := cached[policyID]; extendCache && ok {
r.fireAsyncPolicyResult(policyID, entry.Policy, nil, true)
} else {
r.fireAsyncPolicyResult(policyID, nil, ACLRemoteError{Err: err}, true)
}
}
return
}
func (r *ACLResolver) filterPoliciesByScope(policies structs.ACLPolicies) structs.ACLPolicies {
var out structs.ACLPolicies
for _, policy := range policies {
if len(policy.Datacenters) == 0 {
out = append(out, policy)
continue
}
for _, dc := range policy.Datacenters {
if dc == r.config.Datacenter {
out = append(out, policy)
continue
}
}
}
return out
}
func (r *ACLResolver) resolvePoliciesForIdentity(identity structs.ACLIdentity) (structs.ACLPolicies, error) {
policyIDs := identity.PolicyIDs()
if len(policyIDs) == 0 {
policy := identity.EmbeddedPolicy()
if policy != nil {
return []*structs.ACLPolicy{policy}, nil
}
// In this case the default policy will be all that is in effect.
return nil, nil
}
// For the new ACLs policy replication is mandatory for correct operation on servers. Therefore
// we only attempt to resolve policies locally
policies := make([]*structs.ACLPolicy, 0, len(policyIDs))
// Get all associated policies
var missing []string
var expired []*structs.ACLPolicy
expCacheMap := make(map[string]*structs.PolicyCacheEntry)
for _, policyID := range policyIDs {
if done, policy, err := r.delegate.ResolvePolicyFromID(policyID); done {
if err != nil && !acl.IsErrNotFound(err) {
return nil, err
}
if policy != nil {
policies = append(policies, policy)
} else {
r.logger.Printf("[WARN] acl: policy %q not found for identity %q", policyID, identity.ID())
}
continue
}
// create the missing list which we can execute an RPC to get all the missing policies at once
entry := r.cache.GetPolicy(policyID)
if entry == nil {
missing = append(missing, policyID)
continue
}
if entry.Policy == nil {
// this happens when we cache a negative response for the policies existence
continue
}
if entry.Age() >= r.config.ACLPolicyTTL {
expired = append(expired, entry.Policy)
expCacheMap[policyID] = entry
} else {
policies = append(policies, entry.Policy)
}
}
// Hot-path if we have no missing or expired policies
if len(missing)+len(expired) == 0 {
return r.filterPoliciesByScope(policies), nil
}
fetchIDs := missing
for _, policy := range expired {
fetchIDs = append(fetchIDs, policy.ID)
}
// Background a RPC request and wait on it if we must
var waitChan chan *remoteACLPolicyResult
waitForResult := len(missing) > 0 || r.config.ACLDownPolicy != "async-cache"
if waitForResult {
// buffered because there are going to be multiple go routines that send data to this chan
waitChan = make(chan *remoteACLPolicyResult, len(fetchIDs))
}
var newAsyncFetchIds []string
r.asyncPolicyResultsMutex.Lock()
for _, policyID := range fetchIDs {
clients, ok := r.asyncPolicyResults[policyID]
if !ok || clients == nil {
clients = make([]chan *remoteACLPolicyResult, 0)
}
if waitForResult {
r.asyncPolicyResults[policyID] = append(clients, waitChan)
}
if !ok {
newAsyncFetchIds = append(newAsyncFetchIds, policyID)
}
}
r.asyncPolicyResultsMutex.Unlock()
if len(newAsyncFetchIds) > 0 {
// only start the RPC if one isn't in flight
go r.resolvePoliciesAsyncForIdentity(identity, newAsyncFetchIds, expCacheMap)
}
if !waitForResult {
// waitForResult being false requires that all the policies were cached already
policies = append(policies, expired...)
return r.filterPoliciesByScope(policies), nil
}
for i := 0; i < len(fetchIDs); i++ {
res := <-waitChan
if res.err != nil {
if _, ok := res.err.(*policyTokenError); ok {
// always return token errors
return nil, res.err
} else if !acl.IsErrNotFound(res.err) {
// ignore regular not found errors for policies
return nil, res.err
}
}
// we probably could handle a special case where we
// get a permission denied error due to another requests
// issues and spawn the go routine to resolve it ourselves.
// however this should be exceedingly rare and in this case
// we can just kick the can down the road and retry the whole
// token/policy resolution. All the remaining good bits that
// we need will already be cached anyways.
if res.policy != nil {
policies = append(policies, res.policy)
}
}
return r.filterPoliciesByScope(policies), nil
}
func (r *ACLResolver) resolveTokenToPolicies(token string) (structs.ACLPolicies, error) {
_, policies, err := r.resolveTokenToIdentityAndPolicies(token)
return policies, err
}
func (r *ACLResolver) resolveTokenToIdentityAndPolicies(token string) (structs.ACLIdentity, structs.ACLPolicies, error) {
var lastErr error
var lastIdentity structs.ACLIdentity
for i := 0; i < tokenPolicyResolutionMaxRetries; i++ {
// Resolve the token to an ACLIdentity
identity, err := r.resolveIdentityFromToken(token)
if err != nil {
return nil, nil, err
} else if identity == nil {
return nil, nil, acl.ErrNotFound
}
lastIdentity = identity
policies, err := r.resolvePoliciesForIdentity(identity)
if err == nil {
return identity, policies, nil
}
lastErr = err
if tokenErr, ok := err.(*policyTokenError); ok {
if acl.IsErrNotFound(err) && tokenErr.token == identity.SecretToken() {
// token was deleted while resolving policies
return nil, nil, acl.ErrNotFound
}
// other types of policyTokenErrors should cause retrying the whole token
// resolution process
} else {
return identity, nil, err
}
}
return lastIdentity, nil, lastErr
}
func (r *ACLResolver) disableACLsWhenUpstreamDisabled(err error) error {
if !r.autoDisable || err == nil || !acl.IsErrDisabled(err) {
return err
}
r.logger.Printf("[DEBUG] acl: ACLs disabled on upstream servers, will check again after %s", r.config.ACLDisabledTTL)
r.disabledLock.Lock()
r.disabled = time.Now().Add(r.config.ACLDisabledTTL)
r.disabledLock.Unlock()
return err
}
func (r *ACLResolver) ResolveToken(token string) (acl.Authorizer, error) {
if !r.ACLsEnabled() {
return nil, nil
}
if acl.RootAuthorizer(token) != nil {
return nil, acl.ErrRootDenied
}
// handle the anonymous token
if token == "" {
token = anonymousToken
}
if r.delegate.UseLegacyACLs() {
authorizer, err := r.resolveTokenLegacy(token)
return authorizer, r.disableACLsWhenUpstreamDisabled(err)
}
defer metrics.MeasureSince([]string{"acl", "ResolveToken"}, time.Now())
policies, err := r.resolveTokenToPolicies(token)
if err != nil {
r.disableACLsWhenUpstreamDisabled(err)
if IsACLRemoteError(err) {
r.logger.Printf("[ERR] consul.acl: %v", err)
return r.down, nil
}
return nil, err
}
// Build the Authorizer
authorizer, err := policies.Compile(acl.RootAuthorizer(r.config.ACLDefaultPolicy), r.cache, r.sentinel)
return authorizer, err
}
func (r *ACLResolver) ACLsEnabled() bool {
// Whether we desire ACLs to be enabled according to configuration
if !r.delegate.ACLsEnabled() {
return false
}
if r.autoDisable {
// Whether ACLs are disabled according to RPCs failing with a ACLs Disabled error
r.disabledLock.RLock()
defer r.disabledLock.RUnlock()
return !time.Now().Before(r.disabled)
}
return true
}
func (r *ACLResolver) GetMergedPolicyForToken(token string) (*acl.Policy, error) {
policies, err := r.resolveTokenToPolicies(token)
if err != nil {
return nil, err
}
if len(policies) == 0 {
return nil, acl.ErrNotFound
}
return policies.Merge(r.cache, r.sentinel)
}
// aclFilter is used to filter results from our state store based on ACL rules
// configured for the provided token.
type aclFilter struct {
authorizer acl.Authorizer
logger *log.Logger
enforceVersion8 bool
}
// newACLFilter constructs a new aclFilter.
func newACLFilter(authorizer acl.Authorizer, logger *log.Logger, enforceVersion8 bool) *aclFilter {
if logger == nil {
logger = log.New(os.Stderr, "", log.LstdFlags)
}
return &aclFilter{
authorizer: authorizer,
logger: logger,
enforceVersion8: enforceVersion8,
}
}
// allowNode is used to determine if a node is accessible for an ACL.
func (f *aclFilter) allowNode(node string) bool {
if !f.enforceVersion8 {
return true
}
return f.authorizer.NodeRead(node)
}
// allowService is used to determine if a service is accessible for an ACL.
func (f *aclFilter) allowService(service string) bool {
if service == "" {
return true
}
if !f.enforceVersion8 && service == structs.ConsulServiceID {
return true
}
return f.authorizer.ServiceRead(service)
}
// allowSession is used to determine if a session for a node is accessible for
// an ACL.
func (f *aclFilter) allowSession(node string) bool {
if !f.enforceVersion8 {
return true
}
return f.authorizer.SessionRead(node)
}
// filterHealthChecks is used to filter a set of health checks down based on
// the configured ACL rules for a token.
func (f *aclFilter) filterHealthChecks(checks *structs.HealthChecks) {
hc := *checks
for i := 0; i < len(hc); i++ {
check := hc[i]
if f.allowNode(check.Node) && f.allowService(check.ServiceName) {
continue
}
f.logger.Printf("[DEBUG] consul: dropping check %q from result due to ACLs", check.CheckID)
hc = append(hc[:i], hc[i+1:]...)
i--
}
*checks = hc
}
// filterServices is used to filter a set of services based on ACLs.
func (f *aclFilter) filterServices(services structs.Services) {
for svc := range services {
if f.allowService(svc) {
continue
}
f.logger.Printf("[DEBUG] consul: dropping service %q from result due to ACLs", svc)
delete(services, svc)
}
}
// filterServiceNodes is used to filter a set of nodes for a given service
// based on the configured ACL rules.
func (f *aclFilter) filterServiceNodes(nodes *structs.ServiceNodes) {
sn := *nodes
for i := 0; i < len(sn); i++ {
node := sn[i]
if f.allowNode(node.Node) && f.allowService(node.ServiceName) {
continue
}
f.logger.Printf("[DEBUG] consul: dropping node %q from result due to ACLs", node.Node)
sn = append(sn[:i], sn[i+1:]...)
i--
}
*nodes = sn
}
// filterNodeServices is used to filter services on a given node base on ACLs.
func (f *aclFilter) filterNodeServices(services **structs.NodeServices) {
if *services == nil {
return
}
if !f.allowNode((*services).Node.Node) {
*services = nil
return
}
for svc := range (*services).Services {
if f.allowService(svc) {
continue
}
f.logger.Printf("[DEBUG] consul: dropping service %q from result due to ACLs", svc)
delete((*services).Services, svc)
}
}
// filterCheckServiceNodes is used to filter nodes based on ACL rules.
func (f *aclFilter) filterCheckServiceNodes(nodes *structs.CheckServiceNodes) {
csn := *nodes
for i := 0; i < len(csn); i++ {
node := csn[i]
if f.allowNode(node.Node.Node) && f.allowService(node.Service.Service) {
continue
}
f.logger.Printf("[DEBUG] consul: dropping node %q from result due to ACLs", node.Node.Node)
csn = append(csn[:i], csn[i+1:]...)
i--
}
*nodes = csn
}
// filterSessions is used to filter a set of sessions based on ACLs.
func (f *aclFilter) filterSessions(sessions *structs.Sessions) {
s := *sessions
for i := 0; i < len(s); i++ {
session := s[i]
if f.allowSession(session.Node) {
continue
}
f.logger.Printf("[DEBUG] consul: dropping session %q from result due to ACLs", session.ID)
s = append(s[:i], s[i+1:]...)
i--
}
*sessions = s
}
// filterCoordinates is used to filter nodes in a coordinate dump based on ACL
// rules.
func (f *aclFilter) filterCoordinates(coords *structs.Coordinates) {
c := *coords
for i := 0; i < len(c); i++ {
node := c[i].Node