forked from open-policy-agent/opa
/
opa.go
588 lines (500 loc) · 15.5 KB
/
opa.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
// Copyright 2021 The OPA Authors. All rights reserved.
// Use of this source code is governed by an Apache2
// license that can be found in the LICENSE file.
// Package sdk contains a high-level API for embedding OPA inside of Go programs.
package sdk
import (
"bytes"
"context"
"crypto/rand"
"fmt"
"sync"
"time"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/internal/ref"
"github.com/open-policy-agent/opa/internal/runtime"
"github.com/open-policy-agent/opa/internal/uuid"
"github.com/open-policy-agent/opa/logging"
"github.com/open-policy-agent/opa/metrics"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/plugins/discovery"
"github.com/open-policy-agent/opa/plugins/logs"
"github.com/open-policy-agent/opa/rego"
"github.com/open-policy-agent/opa/server"
"github.com/open-policy-agent/opa/storage"
"github.com/open-policy-agent/opa/storage/inmem"
"github.com/open-policy-agent/opa/topdown/cache"
"github.com/open-policy-agent/opa/topdown/print"
)
// OPA represents an instance of the policy engine. OPA can be started with
// several options that control configuration, logging, and lifecycle.
type OPA struct {
id string
state *state
mtx sync.Mutex
logger logging.Logger
console logging.Logger
plugins map[string]plugins.Factory
config []byte
}
type state struct {
manager *plugins.Manager
interQueryBuiltinCache cache.InterQueryCache
queryCache *queryCache
}
// New returns a new OPA object. This function should minimally be called with
// options that specify an OPA configuration file.
func New(ctx context.Context, opts Options) (*OPA, error) {
id, err := uuid.New(rand.Reader)
if err != nil {
return nil, err
}
if err := opts.init(); err != nil {
return nil, err
}
opa := &OPA{
id: id,
state: &state{
queryCache: newQueryCache(),
},
}
opa.config = opts.config
opa.logger = opts.Logger
opa.console = opts.ConsoleLogger
opa.plugins = opts.Plugins
return opa, opa.configure(ctx, opa.config, opts.Ready, opts.block)
}
// Plugin returns the named plugin. If the plugin does not exist, this function
// returns nil.
func (opa *OPA) Plugin(name string) plugins.Plugin {
opa.mtx.Lock()
defer opa.mtx.Unlock()
return opa.state.manager.Plugin(name)
}
// Configure updates the configuration of the OPA in-place. This function should
// be called in response to configuration updates in the environment. This
// function is atomic. If the configuration update cannot be successfully
// applied, the old configuration will remain intact.
func (opa *OPA) Configure(ctx context.Context, opts ConfigOptions) error {
if err := opts.init(); err != nil {
return err
}
// NOTE(tsandall): In future we could be more intelligent about
// re-configuration and avoid expensive background processing.
opa.mtx.Lock()
equal := bytes.Equal(opts.config, opa.config)
opa.mtx.Unlock()
if equal {
close(opts.Ready)
return nil
}
return opa.configure(ctx, opts.config, opts.Ready, opts.block)
}
func (opa *OPA) configure(ctx context.Context, bs []byte, ready chan struct{}, block bool) error {
info, err := runtime.Term(runtime.Params{Config: opa.config})
if err != nil {
return err
}
manager, err := plugins.New(
bs,
opa.id,
inmem.New(),
plugins.Info(info),
plugins.Logger(opa.logger),
plugins.ConsoleLogger(opa.console),
plugins.EnablePrintStatements(opa.logger.GetLevel() >= logging.Info),
plugins.PrintHook(loggingPrintHook{logger: opa.logger}))
if err != nil {
return err
}
manager.RegisterCompilerTrigger(func(_ storage.Transaction) {
opa.mtx.Lock()
opa.state.queryCache.Clear()
opa.mtx.Unlock()
})
manager.RegisterPluginStatusListener("sdk", func(status map[string]*plugins.Status) {
select {
case <-ready:
return
default:
}
// NOTE(tsandall): we do not include a special case for the discovery
// plugin. If the discovery plugin is the only plugin and it goes ready,
// then OPA will be considered ready. The discovery plugin only goes ready
// _after_ it has successfully processed a discovery bundle. During
// discovery bundle processing, other plugins will register so their states
// will be accounted for. If a discovery bundle did not enable any other
// plugins (bundles, etc.) the OPA will still be operational.
for _, s := range status {
if s.State != plugins.StateOK {
return
}
}
close(ready)
})
d, err := discovery.New(manager, discovery.Factories(opa.plugins))
if err != nil {
return err
}
manager.Register(discovery.Name, d)
if err := manager.Start(ctx); err != nil {
return err
}
if block {
select {
case <-ctx.Done():
return ctx.Err()
case <-ready:
}
}
opa.mtx.Lock()
defer opa.mtx.Unlock()
// NOTE(tsandall): there is no return value from Stop() and it could block
// on async operations (e.g., decision log uploading) so defer the call to
// another goroutine.
//
// TODO(tsandall): if we need to block on operations like decision log
// uploading, perhaps we could rely on a manual trigger.
previousManager := opa.state.manager
go func() {
if previousManager != nil {
previousManager.Stop(ctx)
}
}()
opa.state.manager = manager
opa.state.queryCache.Clear()
opa.state.interQueryBuiltinCache = cache.NewInterQueryCache(manager.InterQueryBuiltinCacheConfig())
opa.config = bs
return nil
}
// Stop closes the OPA. The OPA cannot be restarted.
func (opa *OPA) Stop(ctx context.Context) {
opa.mtx.Lock()
mgr := opa.state.manager
opa.mtx.Unlock()
if mgr != nil {
mgr.Stop(ctx)
}
}
// Decision returns a named decision. This function is threadsafe.
func (opa *OPA) Decision(ctx context.Context, options DecisionOptions) (*DecisionResult, error) {
record := server.Info{
Timestamp: options.Now,
Path: options.Path,
Input: &options.Input,
}
result, err := opa.executeTransaction(
ctx,
&record,
func(result *DecisionResult) {
result.Result, record.InputAST, record.Bundles, record.Error = evaluate(ctx, evalArgs{
runtime: opa.state.manager.Info,
printHook: opa.state.manager.PrintHook(),
compiler: opa.state.manager.GetCompiler(),
store: opa.state.manager.Store,
txn: record.Txn,
queryCache: opa.state.queryCache,
interQueryCache: opa.state.interQueryBuiltinCache,
now: record.Timestamp,
path: record.Path,
input: *record.Input,
m: record.Metrics,
})
if record.Error == nil {
record.Results = &result.Result
}
},
)
if err != nil {
return nil, err
}
return result, record.Error
}
// DecisionOptions contains parameters for query evaluation.
type DecisionOptions struct {
Now time.Time // specifies wallclock time used for time.now_ns(), decision log timestamp, etc.
Path string // specifies name of policy decision to evaluate (e.g., example/allow)
Input interface{} // specifies value of the input document to evaluate policy with
}
// DecisionResult contains the output of query evaluation.
type DecisionResult struct {
ID string // provides a globally unique identifier for this decision (which is included in the decision log.)
Result interface{} // provides the output of query evaluation.
}
func newDecisionResult() (*DecisionResult, error) {
id, err := uuid.New(rand.Reader)
if err != nil {
return nil, err
}
result := &DecisionResult{ID: id}
return result, nil
}
func (opa *OPA) executeTransaction(ctx context.Context, record *server.Info, work func(result *DecisionResult)) (*DecisionResult, error) {
m := metrics.New()
m.Timer(metrics.SDKDecisionEval).Start()
result, err := newDecisionResult()
if err != nil {
return nil, err
}
opa.mtx.Lock()
s := *opa.state
opa.mtx.Unlock()
record.DecisionID = result.ID
record.Metrics = m
if record.Timestamp.IsZero() {
record.Timestamp = time.Now().UTC()
}
if record.Path == "" {
record.Path = *s.manager.Config.DefaultDecision
}
record.Txn, record.Error = s.manager.Store.NewTransaction(ctx, storage.TransactionParams{})
if record.Error == nil {
defer s.manager.Store.Abort(ctx, record.Txn)
work(result)
}
m.Timer(metrics.SDKDecisionEval).Stop()
if logger := logs.Lookup(s.manager); logger != nil {
if err := logger.Log(ctx, record); err != nil {
return result, fmt.Errorf("decision log: %w", err)
}
}
return result, nil
}
// Partial returns a named decision. This function is threadsafe.
func (opa *OPA) Partial(ctx context.Context, options PartialOptions) (*PartialResult, error) {
if options.Mapper == nil {
options.Mapper = &RawMapper{}
}
record := server.Info{
Timestamp: options.Now,
Input: &options.Input,
Query: options.Query,
}
var pq *rego.PartialQueries
decision, err := opa.executeTransaction(
ctx,
&record,
func(result *DecisionResult) {
pq, record.InputAST, record.Bundles, record.Error = partial(ctx, partialEvalArgs{
runtime: opa.state.manager.Info,
printHook: opa.state.manager.PrintHook(),
compiler: opa.state.manager.GetCompiler(),
store: opa.state.manager.Store,
txn: record.Txn,
now: record.Timestamp,
query: record.Query,
unknowns: options.Unknowns,
input: *record.Input,
m: record.Metrics,
})
if record.Error == nil {
result.Result, record.Error = options.Mapper.MapResults(pq)
var pqAst interface{}
if record.Error == nil {
var mappedResults interface{}
mappedResults, record.Error = options.Mapper.ResultToJSON(result.Result)
record.MappedResults = &mappedResults
pqAst = pq
record.Results = &pqAst
}
}
},
)
if err != nil {
return nil, err
}
return &PartialResult{
ID: decision.ID,
Result: decision.Result,
AST: pq,
}, record.Error
}
type PartialQueryMapper interface {
// The first interface being returned is the type that will be used for further processing
MapResults(pq *rego.PartialQueries) (interface{}, error)
// This should be able to take the Result object from MapResults and return a type that can be logged as JSON
ResultToJSON(result interface{}) (interface{}, error)
}
// PartialOptions contains parameters for partial query evaluation.
type PartialOptions struct {
Now time.Time // specifies wallclock time used for time.now_ns(), decision log timestamp, etc.
Input interface{} // specifies value of the input document to evaluate policy with
Query string // specifies the query to be partially evaluated
Unknowns []string // specifies the unknown elements of the policy
Mapper PartialQueryMapper // specifies the mapper to use when processing results
}
type PartialResult struct {
ID string // decision ID
Result interface{} // mapped result
AST *rego.PartialQueries // raw result
}
// Error represents an internal error in the SDK.
type Error struct {
Code string `json:"code"`
Message string `json:"message,omitempty"`
}
func (err *Error) Error() string {
return fmt.Sprintf("%v: %v", err.Code, err.Message)
}
const (
// UndefinedErr indicates that the queried decision was undefined.
UndefinedErr = "opa_undefined_error"
)
func undefinedDecisionErr(path string) *Error {
return &Error{
Code: UndefinedErr,
Message: fmt.Sprintf("%v decision was undefined", path),
}
}
// IsUndefinedErr returns true of the err represents an undefined decision error.
func IsUndefinedErr(err error) bool {
actual, ok := err.(*Error)
return ok && actual.Code == UndefinedErr
}
type evalArgs struct {
runtime *ast.Term
printHook print.Hook
compiler *ast.Compiler
store storage.Store
txn storage.Transaction
queryCache *queryCache
interQueryCache cache.InterQueryCache
now time.Time
path string
input interface{}
m metrics.Metrics
}
func evaluate(ctx context.Context, args evalArgs) (interface{}, ast.Value, map[string]server.BundleInfo, error) {
bundles, err := bundles(ctx, args.store, args.txn)
if err != nil {
return nil, nil, nil, err
}
r, err := ref.ParseDataPath(args.path)
if err != nil {
return nil, nil, bundles, err
}
pq, err := args.queryCache.Get(r.String(), func(query string) (*rego.PreparedEvalQuery, error) {
pq, err := rego.New(
rego.Time(args.now),
rego.Metrics(args.m),
rego.Query(query),
rego.Compiler(args.compiler),
rego.Store(args.store),
rego.Transaction(args.txn),
rego.PrintHook(args.printHook),
rego.Runtime(args.runtime)).PrepareForEval(ctx)
if err != nil {
return nil, err
}
return &pq, err
})
if err != nil {
return nil, nil, bundles, err
}
inputAST, err := ast.InterfaceToValue(args.input)
if err != nil {
return nil, nil, bundles, err
}
rs, err := pq.Eval(
ctx,
rego.EvalTime(args.now),
rego.EvalParsedInput(inputAST),
rego.EvalTransaction(args.txn),
rego.EvalMetrics(args.m),
rego.EvalInterQueryBuiltinCache(args.interQueryCache),
)
if err != nil {
return nil, inputAST, bundles, err
} else if len(rs) == 0 {
return nil, inputAST, bundles, undefinedDecisionErr(args.path)
}
return rs[0].Expressions[0].Value, inputAST, bundles, nil
}
type partialEvalArgs struct {
runtime *ast.Term
compiler *ast.Compiler
printHook print.Hook
store storage.Store
txn storage.Transaction
unknowns []string
query string
now time.Time
input interface{}
m metrics.Metrics
}
func partial(ctx context.Context, args partialEvalArgs) (*rego.PartialQueries, ast.Value, map[string]server.BundleInfo, error) {
bundles, err := bundles(ctx, args.store, args.txn)
if err != nil {
return nil, nil, nil, err
}
inputAST, err := ast.InterfaceToValue(args.input)
if err != nil {
return nil, nil, bundles, err
}
re := rego.New(
rego.Time(args.now),
rego.Metrics(args.m),
rego.Store(args.store),
rego.Compiler(args.compiler),
rego.Transaction(args.txn),
rego.Runtime(args.runtime),
rego.Input(args.input),
rego.Query(args.query),
rego.Unknowns(args.unknowns),
rego.PrintHook(args.printHook),
)
pq, err := re.Partial(ctx)
if err != nil {
return nil, nil, bundles, err
}
return pq, inputAST, bundles, err
}
type queryCache struct {
sync.Mutex
cache map[string]*rego.PreparedEvalQuery
}
func newQueryCache() *queryCache {
return &queryCache{cache: map[string]*rego.PreparedEvalQuery{}}
}
func (qc *queryCache) Get(key string, orElse func(string) (*rego.PreparedEvalQuery, error)) (*rego.PreparedEvalQuery, error) {
qc.Lock()
defer qc.Unlock()
result, ok := qc.cache[key]
if ok {
return result, nil
}
result, err := orElse(key)
if err != nil {
return nil, err
}
qc.cache[key] = result
return result, nil
}
func (qc *queryCache) Clear() {
qc.Lock()
defer qc.Unlock()
qc.cache = make(map[string]*rego.PreparedEvalQuery)
}
func bundles(ctx context.Context, store storage.Store, txn storage.Transaction) (map[string]server.BundleInfo, error) {
bundles := map[string]server.BundleInfo{}
names, err := bundle.ReadBundleNamesFromStore(ctx, store, txn)
if err != nil && !storage.IsNotFound(err) {
return nil, fmt.Errorf("failed to read bundle names: %w", err)
}
for _, name := range names {
r, err := bundle.ReadBundleRevisionFromStore(ctx, store, txn, name)
if err != nil {
return nil, fmt.Errorf("failed to read bundle revisions: %w", err)
}
bundles[name] = server.BundleInfo{Revision: r}
}
return bundles, nil
}
type loggingPrintHook struct {
logger logging.Logger
}
func (h loggingPrintHook) Print(pctx print.Context, msg string) error {
h.logger.WithFields(map[string]interface{}{"line": pctx.Location.String()}).Info(msg)
return nil
}