diff --git a/.gitignore b/.gitignore index c09a315d05..0ec8e28e1c 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,7 @@ opa_* _release wasm/_obj _test +!*_test.go site.tar.gz policy.wasm .npm diff --git a/internal/wasm/sdk/README.md b/internal/wasm/sdk/README.md new file mode 100644 index 0000000000..af407db984 --- /dev/null +++ b/internal/wasm/sdk/README.md @@ -0,0 +1,7 @@ +**Work in Progress -- Contributions welcome!** + +# Open Policy Agent WebAssemby Go SDK +This is the source for the Open Policy Agent WebAssembly Go SDK which +is a small go library for using WebAssembly (wasm) compiled [Open +Policy Agent](https://www.openpolicyagent.org/) Rego policies. + diff --git a/internal/wasm/sdk/examples/basic/example-1.rego b/internal/wasm/sdk/examples/basic/example-1.rego new file mode 100644 index 0000000000..cbc29d0560 --- /dev/null +++ b/internal/wasm/sdk/examples/basic/example-1.rego @@ -0,0 +1,3 @@ +package example + +allow = input.foo diff --git a/internal/wasm/sdk/examples/basic/example-1.wasm b/internal/wasm/sdk/examples/basic/example-1.wasm new file mode 100644 index 0000000000..3045bd85a6 Binary files /dev/null and b/internal/wasm/sdk/examples/basic/example-1.wasm differ diff --git a/internal/wasm/sdk/examples/basic/example-2.rego b/internal/wasm/sdk/examples/basic/example-2.rego new file mode 100644 index 0000000000..f60b425834 --- /dev/null +++ b/internal/wasm/sdk/examples/basic/example-2.rego @@ -0,0 +1,3 @@ +package example + +allow = input.bar diff --git a/internal/wasm/sdk/examples/basic/example-2.wasm b/internal/wasm/sdk/examples/basic/example-2.wasm new file mode 100644 index 0000000000..665168720c Binary files /dev/null and b/internal/wasm/sdk/examples/basic/example-2.wasm differ diff --git a/internal/wasm/sdk/examples/basic/main.go b/internal/wasm/sdk/examples/basic/main.go new file mode 100644 index 0000000000..1066c32dee --- /dev/null +++ b/internal/wasm/sdk/examples/basic/main.go @@ -0,0 +1,97 @@ +// Copyright 2020 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 main + +import ( + "context" + "fmt" + "io/ioutil" + "os" + "path" + + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa" +) + +// main demonstrates the loading and executnig of OPA produced wasm +// policy binary. To execute run 'go run main.go .' in the directory +// of the main.go. +func main() { + if len(os.Args) < 2 { + fmt.Printf("%s: first argument must a path to a directory with example-1.wasm and example-2.wasm.\n", os.Args[0]) + return + } + + directory := os.Args[1] + + // Setup the SDK + + policy, err := ioutil.ReadFile(path.Join(directory, "example-1.wasm")) + if err != nil { + fmt.Printf("error: %v\n", err) + return + } + + rego, err := opa.New().WithPolicyBytes(policy).Init() + if err != nil { + fmt.Printf("error: %v\n", err) + return + } + + defer rego.Close() + + // Evaluate the policy once. + + var input interface{} = map[string]interface{}{ + "foo": true, + "bar": false, + } + + ctx := context.Background() + result, err := rego.Eval(ctx, &input) + if err != nil { + fmt.Printf("error: %v\n", err) + return + } + + fmt.Printf("Policy 1 result: %v\n", result) + + resultBool, err := opa.EvalBool(ctx, rego, &input) + if err != nil { + fmt.Printf("error: %v\n", err) + return + } + + fmt.Printf("Policy 1 boolean result: %v\n", resultBool) + + // Update the policy on the fly. + + policy, err = ioutil.ReadFile(path.Join(directory, "example-2.wasm")) + if err != nil { + fmt.Printf("error: %v\n", err) + return + } + + // Evaluate the new policy. + + if err := rego.SetPolicy(policy); err != nil { + fmt.Printf("error: %v\n", err) + return + } + + result, err = rego.Eval(ctx, &input) + if err != nil { + fmt.Printf("error: %v\n", err) + return + } + + fmt.Printf("Policy 2 result: %v\n", result) + + resultBool, err = opa.EvalBool(ctx, rego, &input) + if err != nil { + fmt.Printf("error: %v\n", err) + return + } + + fmt.Printf("Policy 2 boolean result: %v\n", resultBool) +} diff --git a/internal/wasm/sdk/examples/loaders/bundle.tgz b/internal/wasm/sdk/examples/loaders/bundle.tgz new file mode 100644 index 0000000000..014875f901 Binary files /dev/null and b/internal/wasm/sdk/examples/loaders/bundle.tgz differ diff --git a/internal/wasm/sdk/examples/loaders/main.go b/internal/wasm/sdk/examples/loaders/main.go new file mode 100644 index 0000000000..eebed83c9b --- /dev/null +++ b/internal/wasm/sdk/examples/loaders/main.go @@ -0,0 +1,114 @@ +// Copyright 2020 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 main + +import ( + "context" + "fmt" + gohttp "net/http" + "net/url" + "os" + "time" + + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa" + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa/file" + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa/http" +) + +var ( + loader opa.Loader + rego *opa.OPA +) + +// main loads a bundle either from a file or HTTP server. +// +// In the directory of the main.go, execute 'go run main.go +// bundle.tgz' to load the accompanied bundle file. Similarly, execute +// 'go run main.go http://url/to/bundle.tgz' to test the HTTP +// downloading from a HTTP server. +func main() { + if len(os.Args) < 2 { + fmt.Printf("provide URL or file\n") + return + } + + url := os.Args[1] + token := "" + if len(os.Args) >= 3 { + token = os.Args[2] + } + + // Setup the SDK, either with HTTP bundle loader or file bundle loader. + + if err := setup(url, token); err != nil { + fmt.Printf("error: %v\n", err) + return + } + + defer cleanup() + + // Evaluate the policy. + + var input interface{} = map[string]interface{}{ + "foo": true, + } + + ctx := context.Background() + result, err := rego.Eval(ctx, &input) + if err != nil { + fmt.Printf("error: %v\n", err) + return + } + + fmt.Printf("Policy result: %v\n", result) +} + +func setup(u string, token string) error { + r, err := opa.New().Init() + if err != nil { + return err + } + + url, err := url.Parse(u) + if err != nil { + return err + } + + var l opa.Loader + + switch url.Scheme { + case "http", "https": + l, err = http.New(r). + WithURL(url.String()). + WithPrepareRequest(func(req *gohttp.Request) error { + if token != "" { + req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", token)) + } + return nil + }). + WithInterval(30*time.Second, 60*time.Second). + Init() + case "file", "": + l, err = file.New(r). + WithFile(url.String()). + WithInterval(10 * time.Second). + Init() + } + + if err != nil { + return err + } + + if err := l.Start(context.Background()); err != nil { + return err + } + + rego, loader = r, l + return nil +} + +func cleanup() { + loader.Close() + rego.Close() +} diff --git a/internal/wasm/sdk/opa/bindings.go b/internal/wasm/sdk/opa/bindings.go new file mode 100644 index 0000000000..e6df389177 --- /dev/null +++ b/internal/wasm/sdk/opa/bindings.go @@ -0,0 +1,85 @@ +// Copyright 2020 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 opa + +// #include +// +// extern void opa_abort(void *context, int32_t addr); +// extern int32_t opa_builtin0(void *context, int32_t builtin_id, int32_t ctx); +// extern int32_t opa_builtin1(void *context, int32_t builtin_id, int32_t ctx, int32_t arg0); +// extern int32_t opa_builtin2(void *context, int32_t builtin_id, int32_t ctx, int32_t arg0, int32_t arg1); +// extern int32_t opa_builtin3(void *context, int32_t builtin_id, int32_t ctx, int32_t arg0, int32_t arg1, int32_t arg2); +// extern int32_t opa_builtin4(void *context, int32_t builtin_id, int32_t ctx, int32_t arg0, int32_t arg1, int32_t arg2, int32_t arg3); +import "C" + +import ( + "unsafe" + + wasm "github.com/wasmerio/go-ext-wasm/wasmer" +) + +func opaFunctions(imports *wasm.Imports) (*wasm.Imports, error) { + imports, err := imports.AppendFunction("opa_abort", opa_abort, C.opa_abort) + if err != nil { + return nil, err + } + + imports, err = imports.AppendFunction("opa_builtin0", opa_builtin0, C.opa_builtin0) + if err != nil { + return nil, err + } + + imports, err = imports.AppendFunction("opa_builtin1", opa_builtin1, C.opa_builtin1) + if err != nil { + return nil, err + } + + imports, err = imports.AppendFunction("opa_builtin2", opa_builtin2, C.opa_builtin2) + if err != nil { + return nil, err + } + + imports, err = imports.AppendFunction("opa_builtin3", opa_builtin3, C.opa_builtin3) + if err != nil { + return nil, err + } + + return imports.AppendFunction("opa_builtin4", opa_builtin4, C.opa_builtin4) +} + +//export opa_abort +func opa_abort(ctx unsafe.Pointer, addr int32) { + getVM(ctx).Abort(addr) +} + +//export opa_builtin0 +func opa_builtin0(ctx unsafe.Pointer, builtinID, context int32) int32 { + return getVM(ctx).Builtin(builtinID, context) +} + +//export opa_builtin1 +func opa_builtin1(ctx unsafe.Pointer, builtinID, context, arg0 int32) int32 { + return getVM(ctx).Builtin(builtinID, context, arg0) +} + +//export opa_builtin2 +func opa_builtin2(ctx unsafe.Pointer, builtinID, context, arg0, arg1 int32) int32 { + return getVM(ctx).Builtin(builtinID, context, arg0, arg1) +} + +//export opa_builtin3 +func opa_builtin3(ctx unsafe.Pointer, builtinID, context, arg0, arg1, arg2 int32) int32 { + return getVM(ctx).Builtin(builtinID, context, arg0, arg1, arg2) +} + +//export opa_builtin4 +func opa_builtin4(ctx unsafe.Pointer, builtinID, context, arg0, arg1, arg2, arg3 int32) int32 { + return getVM(ctx).Builtin(builtinID, context, arg0, arg1, arg2, arg3) +} + +func getVM(ctx unsafe.Pointer) *vm { + ictx := wasm.IntoInstanceContext(ctx) + return ictx.Data().(*vm) +} diff --git a/internal/wasm/sdk/opa/config.go b/internal/wasm/sdk/opa/config.go new file mode 100644 index 0000000000..517749687f --- /dev/null +++ b/internal/wasm/sdk/opa/config.go @@ -0,0 +1,108 @@ +// Copyright 2020 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 opa + +import ( + "encoding/json" + "fmt" + "io/ioutil" +) + +const wasmPageSize = 65535 + +// WithPolicyFile configures a policy file to load. +func (o *OPA) WithPolicyFile(fileName string) *OPA { + policy, err := ioutil.ReadFile(fileName) + if err != nil { + o.configErr = fmt.Errorf("%v: %w", err.Error(), ErrInvalidConfig) + return o + } + + o.policy = policy + return o +} + +// WithPolicyBytes configures the compiled policy to load. +func (o *OPA) WithPolicyBytes(policy []byte) *OPA { + o.policy = policy + return o +} + +// WithDataFile configures the JSON data file to load. +func (o *OPA) WithDataFile(fileName string) *OPA { + data, err := ioutil.ReadFile(fileName) + if err != nil { + o.configErr = fmt.Errorf("%v: %w", err.Error(), ErrInvalidConfig) + return o + } + + o.data = data + return o +} + +// WithDataBytes configures the JSON data to load. +func (o *OPA) WithDataBytes(data []byte) *OPA { + o.data = data + return o +} + +// WithDataJSON configures the JSON data to load. +func (o *OPA) WithDataJSON(data interface{}) *OPA { + v, err := json.Marshal(data) + if err != nil { + o.configErr = fmt.Errorf("%v: %w", err.Error(), ErrInvalidConfig) + return o + } + + o.data = v + return o +} + +// WithMemoryLimits configures the memory limits (in bytes) for a single policy +// evaluation. +func (o *OPA) WithMemoryLimits(min, max uint32) *OPA { + if min < 2*65535 { + o.configErr = fmt.Errorf("too low minimum memory limit: %w", ErrInvalidConfig) + return o + } + + if max != 0 && min > max { + o.configErr = fmt.Errorf("too low maximum memory limit: %w", ErrInvalidConfig) + return o + } + + o.memoryMinPages, o.memoryMaxPages = pages(min), pages(max) + return o +} + +// WithPoolSize configures the maximum number of simultaneous policy +// evaluations, i.e., the maximum number of underlying WASM instances +// active at any time. The default is the number of logical CPUs +// usable for the process as per runtime.NumCPU(). +func (o *OPA) WithPoolSize(size uint32) *OPA { + if size == 0 { + o.configErr = fmt.Errorf("pool size: %w", ErrInvalidConfig) + return o + } + + o.poolSize = size + return o +} + +// WithErrorLogger configures an error logger invoked with all the errors. +func (o *OPA) WithErrorLogger(logger func(error)) *OPA { + o.logError = logger + return o +} + +// pages converts a byte size to pages, rounding up as necessary. +func pages(n uint32) uint32 { + pages := n / wasmPageSize + if pages*wasmPageSize == n { + return pages + } + + return pages + 1 +} diff --git a/internal/wasm/sdk/opa/errors.go b/internal/wasm/sdk/opa/errors.go new file mode 100644 index 0000000000..76630d84f5 --- /dev/null +++ b/internal/wasm/sdk/opa/errors.go @@ -0,0 +1,26 @@ +// Copyright 2020 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 opa + +import ( + "errors" +) + +var ( + // ErrInvalidConfig is the error returned if the OPA initialization fails due to an invalid config. + ErrInvalidConfig = errors.New("invalid config") + // ErrInvalidPolicyOrData is the error returned if either policy or data is invalid. + ErrInvalidPolicyOrData = errors.New("invalid policy or data") + // ErrInvalidBundle is the error returned if the bundle loaded is corrupted. + ErrInvalidBundle = errors.New("invalid bundle") + // ErrNotReady is the error returned if the OPA instance is not initialized. + ErrNotReady = errors.New("not ready") + // ErrUndefined is the error returned if the evaluation result is undefined. + ErrUndefined = errors.New("undefined decision") + // ErrNonBoolean is the error returned if the evaluation result is not of boolean value. + ErrNonBoolean = errors.New("non-boolean decision") + // ErrInternal is the error returned if the evaluation fails due to an internal error. + ErrInternal = errors.New("internal error") +) diff --git a/internal/wasm/sdk/opa/file/config.go b/internal/wasm/sdk/opa/file/config.go new file mode 100644 index 0000000000..43d2038836 --- /dev/null +++ b/internal/wasm/sdk/opa/file/config.go @@ -0,0 +1,35 @@ +// Copyright 2020 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 file + +import ( + "fmt" + "time" + + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa" +) + +// WithFile configures the file to load the bundle from. +func (l *Loader) WithFile(filename string) *Loader { + l.filename = filename + return l +} + +// WithInterval configures the delay between bundle file reloading. +func (l *Loader) WithInterval(interval time.Duration) *Loader { + l.interval = interval + return l +} + +// WithErrorLogger configures an error logger invoked with all the errors. +func (l *Loader) WithErrorLogger(logger func(error)) *Loader { + if logger == nil { + l.configErr = fmt.Errorf("logger: %w", opa.ErrInvalidConfig) + return l + } + + l.logError = logger + return l +} diff --git a/internal/wasm/sdk/opa/file/loader.go b/internal/wasm/sdk/opa/file/loader.go new file mode 100644 index 0000000000..ce1a71df76 --- /dev/null +++ b/internal/wasm/sdk/opa/file/loader.go @@ -0,0 +1,163 @@ +// Copyright 2020 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 file + +import ( + "context" + "fmt" + "os" + "sync" + "time" + + "github.com/open-policy-agent/opa/bundle" + + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa" +) + +const ( + // DefaultInterval for re-loading the bundle file. + DefaultInterval = time.Minute +) + +// Loader loads a bundle from a file. If started, it loads the bundle +// periodically until closed. +type Loader struct { + configErr error // Delayed configuration error, if any. + initialized bool + pd policyData + filename string + interval time.Duration + closing chan struct{} // Signal the request to stop the poller. + closed chan struct{} // Signals the successful stopping of the poller. + logError func(error) + mutex sync.Mutex +} + +// policyData captures the functions used in setting the policy and data. +type policyData interface { + SetPolicyData(policy []byte, data *interface{}) error +} + +// New constructs a new file loader periodically reloading the bundle +// from a file. +func New(opa *opa.OPA) *Loader { + return new(opa) +} + +// new constucts a new file loader. This is for tests. +func new(pd policyData) *Loader { + return &Loader{ + pd: pd, + interval: DefaultInterval, + logError: func(error) {}, + } +} + +// Init initializes the loader after its construction and +// configuration. If invalid config, will return ErrInvalidConfig. +func (l *Loader) Init() (*Loader, error) { + if l.configErr != nil { + return nil, l.configErr + } + + if l.filename == "" { + return nil, fmt.Errorf("filename: %w", opa.ErrInvalidConfig) + } + + l.initialized = true + return l, nil +} + +// Start starts the periodic loading byt calling Load, failing if the +// bundle loading fails. +func (l *Loader) Start(ctx context.Context) error { + if !l.initialized { + return opa.ErrNotReady + } + + if err := l.Load(ctx); err != nil { + return err + } + + l.closing = make(chan struct{}) + l.closed = make(chan struct{}) + + go l.poller() + + return nil +} + +// Close stops the loading, releasing all resources. +func (l *Loader) Close() { + if !l.initialized { + return + } + + if l.closing == nil { + return + } + + close(l.closing) + <-l.closed + + l.closing = nil + l.closed = nil +} + +// Load loads the bundle from a file and installs it. The possible +// returned errors are ErrInvalidBundle (in case of an error in +// loading or opening the bundle) and the ones SetPolicyData of OPA +// returns. +func (l *Loader) Load(ctx context.Context) error { + if !l.initialized { + return opa.ErrNotReady + } + + l.mutex.Lock() + defer l.mutex.Unlock() + + f, err := os.Open(l.filename) + if err != nil { + return fmt.Errorf("%v: %w", err, opa.ErrInvalidBundle) + } + + defer f.Close() + + // TODO: Cut the dependency to the OPA bundle package. + + bundle, err := bundle.NewReader(f).Read() + if err != nil { + return fmt.Errorf("%v: %w", err, opa.ErrInvalidBundle) + } + + if bundle.Wasm == nil { + return fmt.Errorf("missing wasm: %w", opa.ErrInvalidBundle) + } + + var data *interface{} + if bundle.Data != nil { + var v interface{} = bundle.Data + data = &v + } + + return l.pd.SetPolicyData(bundle.Wasm, data) +} + +// poller periodically downloads the bundle. +func (l *Loader) poller() { + defer close(l.closed) + + for { + if err := l.Load(context.Background()); err != nil { + l.logError(err) + } + + select { + case <-time.After(l.interval): + case <-l.closing: + return + } + } +} diff --git a/internal/wasm/sdk/opa/file/loader_test.go b/internal/wasm/sdk/opa/file/loader_test.go new file mode 100644 index 0000000000..5af655d99a --- /dev/null +++ b/internal/wasm/sdk/opa/file/loader_test.go @@ -0,0 +1,128 @@ +// Copyright 2020 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 file + +import ( + "bytes" + "context" + "io/ioutil" + "os" + "reflect" + "sync" + "testing" + "time" + + "github.com/open-policy-agent/opa/bundle" +) + +func TestFileLoader(t *testing.T) { + // Assign a temp file. + + f, err := ioutil.TempFile("", "test-file-loader") + if err != nil { + panic(err) + } + + defer os.Remove(f.Name()) + + // Start loader, without having a file in place. + + var pd testPolicyData + loader, err := new(&pd).WithFile(f.Name()).WithInterval(10 * time.Millisecond).Init() + if err != nil { + t.Fatal(err.Error()) + } + + ctx := context.Background() + if err := loader.Start(ctx); err == nil { + t.Fatal("missing file not resulting in an error") + } + + policy := "wasm-policy" + var data interface{} = map[string]interface{}{ + "foo": "bar", + } + + // Start loader, with the file in place. + + writeBundle(f.Name(), policy, data) + + if err := loader.Start(ctx); err != nil { + t.Fatalf("unable to start loader: %v", err) + } + + pd.CheckEqual(t, policy, &data) + + // Reload with updated contents. + + policy = "wasm-policy-modified" + data = map[string]interface{}{ + "bar": "foo", + } + + writeBundle(f.Name(), policy, data) + + pd.WaitUpdate() + pd.CheckEqual(t, policy, &data) + + loader.Close() +} + +type testPolicyData struct { + sync.Mutex + policy []byte + data *interface{} + updated chan struct{} +} + +func (pd *testPolicyData) SetPolicyData(policy []byte, data *interface{}) error { + pd.Lock() + defer pd.Unlock() + + pd.policy = policy + pd.data = data + if pd.updated != nil { + close(pd.updated) + } + + return nil +} + +func (pd *testPolicyData) CheckEqual(t *testing.T, policy string, data *interface{}) { + pd.Lock() + defer pd.Unlock() + + if !bytes.Equal([]byte(policy), pd.policy) && reflect.DeepEqual(data, pd.data) { + t.Fatal("policy/data mismatch.") + } +} + +func (pd *testPolicyData) WaitUpdate() { + pd.Lock() + pd.updated = make(chan struct{}) + pd.Unlock() + + <-pd.updated + + pd.Lock() + pd.updated = nil + pd.Unlock() +} + +func writeBundle(name string, policy string, data interface{}) { + b := bundle.Bundle{ + Data: data.(map[string]interface{}), + Wasm: []byte(policy), + } + + var buf bytes.Buffer + if err := bundle.Write(&buf, b); err != nil { + panic(err) + } + + if err := ioutil.WriteFile(name, buf.Bytes(), 0644); err != nil { + panic(err) + } +} diff --git a/internal/wasm/sdk/opa/http/config.go b/internal/wasm/sdk/opa/http/config.go new file mode 100644 index 0000000000..9cb499618e --- /dev/null +++ b/internal/wasm/sdk/opa/http/config.go @@ -0,0 +1,66 @@ +// Copyright 2020 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 http + +import ( + "fmt" + "net/http" + "time" + + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa" +) + +// WithURL configures the URL to download the bundle from. +func (l *Loader) WithURL(url string) *Loader { + l.url = url + return l +} + +// WithClient configures the HTTP client to use. If not configured, +// http.DefaultClient is used. +func (l *Loader) WithClient(client *http.Client) *Loader { + if client == nil { + l.configErr = fmt.Errorf("client: %w", opa.ErrInvalidConfig) + return l + } + + l.client = client + return l +} + +// WithInterval configures the minimum and maximum delay between bundle downloads. +func (l *Loader) WithInterval(min, max time.Duration) *Loader { + if min > max { + l.configErr = fmt.Errorf("interval: %w", opa.ErrInvalidConfig) + return l + } + + l.minDelay = min + l.maxDelay = max + return l +} + +// WithPrepareRequest configures a handler to customize the HTTP requests before their sending. The +// HTTP request is not modified after the handle invocation. +func (l *Loader) WithPrepareRequest(prepare func(*http.Request) error) *Loader { + if prepare == nil { + l.configErr = fmt.Errorf("prepare request: %w", opa.ErrInvalidConfig) + return l + } + + l.prepareRequest = prepare + return l +} + +// WithErrorLogger configures an error logger invoked with all the errors. +func (l *Loader) WithErrorLogger(logger func(error)) *Loader { + if logger == nil { + l.configErr = fmt.Errorf("logger: %w", opa.ErrInvalidConfig) + return l + } + + l.logError = logger + return l +} diff --git a/internal/wasm/sdk/opa/http/loader.go b/internal/wasm/sdk/opa/http/loader.go new file mode 100644 index 0000000000..253d14b5c2 --- /dev/null +++ b/internal/wasm/sdk/opa/http/loader.go @@ -0,0 +1,257 @@ +// Copyright 2020 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 http + +import ( + "context" + "fmt" + "io" + "io/ioutil" + "math/rand" + "net/http" + "sync" + "time" + + "github.com/open-policy-agent/opa/bundle" + + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa" +) + +const ( + // MinRetryDelay determines the minimum retry interval in case + // of an error. + MinRetryDelay = 100 * time.Millisecond + + // DefaultMinDelay is the default minimum re-downloading + // interval in case of a previously successful download. + DefaultMinDelay = 60 * time.Second + + // DefaultMaxDelay is the default maximum re-downloading + // interval in case of a previously successful download. + DefaultMaxDelay = 120 * time.Second +) + +// Loader downloads a bundle over HTTP. If started, it downloads the +// bundle periodically until closed. +type Loader struct { + configErr error // Delayed configuration error, if any. + initialized bool + pd policyData + client *http.Client + url string + tag string + minDelay time.Duration + maxDelay time.Duration + closing chan struct{} // Signal the request to stop the poller. + closed chan struct{} // Signals the successful stopping of the poller. + logError func(error) + prepareRequest func(*http.Request) error + mutex sync.Mutex +} + +// policyData captures the functions used in setting the policy and data. +type policyData interface { + SetPolicyData(policy []byte, data *interface{}) error +} + +// New constructs a new HTTP loader periodically downloading a bundle +// over HTTP. +func New(o *opa.OPA) *Loader { + return new(o) +} + +// new constucts a new HTTP loader. This is for tests. +func new(pd policyData) *Loader { + return &Loader{ + pd: pd, + client: http.DefaultClient, + minDelay: DefaultMinDelay, + maxDelay: DefaultMaxDelay, + logError: func(error) {}, + prepareRequest: func(*http.Request) error { return nil }, + } +} + +// Init initializes the loader after its construction and +// configuration. If invalid config, will return ErrInvalidConfig. +func (l *Loader) Init() (*Loader, error) { + if l.configErr != nil { + return nil, l.configErr + } + + if l.url == "" { + return nil, fmt.Errorf("missing url: %w", opa.ErrInvalidConfig) + } + + l.initialized = true + return l, nil +} + +// Start starts the periodic downloads, blocking until the first +// successful download. If cancelled, will return context.Cancelled. +func (l *Loader) Start(ctx context.Context) error { + if !l.initialized { + return opa.ErrNotReady + } + + if err := l.download(ctx); err != nil { + return err + } + + l.closing = make(chan struct{}) + l.closed = make(chan struct{}) + + go l.poller() + + return nil +} + +// Close stops the downloading, releasing all resources. +func (l *Loader) Close() { + if !l.initialized { + return + } + + if l.closing == nil { + return + } + + close(l.closing) + <-l.closed + + l.closing = nil + l.closed = nil +} + +// poller periodically downloads the bundle. +func (l *Loader) poller() { + defer close(l.closed) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + <-l.closing + cancel() + }() + + for { + if err := l.download(ctx); err != nil { + break + } + + select { + case <-time.After(time.Duration(float64((l.maxDelay-l.minDelay))*rand.Float64()) + l.minDelay): + case <-l.closing: + return + } + } +} + +// download blocks until a bundle has been download successfully or +// the context is cancelled. No other error besides context.Canceled +// is ever returned. +func (l *Loader) download(ctx context.Context) error { + for retry := 0; true; retry++ { + if err := l.Load(ctx); err == context.Canceled { + return err + } else if err != nil { + l.logError(err) + } else if err == nil { + break + } + + select { + case <-time.After(defaultBackoff(float64(MinRetryDelay), float64(l.maxDelay), retry)): + case <-ctx.Done(): + return context.Canceled + } + } + + return nil +} + +// Load downloads the bundle from a remote location and installs +// it. The possible returned errors are ErrInvalidBundle (in case of +// an error in downloading or opening the bundle) and the ones +// SetPolicyData of OPA returns. +func (l *Loader) Load(ctx context.Context) error { + if !l.initialized { + return opa.ErrNotReady + } + + l.mutex.Lock() + defer l.mutex.Unlock() + + bundle, err := l.get(ctx, "") + if err != nil { + return fmt.Errorf("%v: %w", err, opa.ErrInvalidBundle) + } + + if bundle.Wasm == nil { + return opa.ErrInvalidBundle + } + + var data *interface{} + if bundle.Data != nil { + var v interface{} = bundle.Data + data = &v + } + + return l.pd.SetPolicyData(bundle.Wasm, data) +} + +// get executes HTTP GET. +func (l *Loader) get(ctx context.Context, tag string) (*bundle.Bundle, error) { + req, err := http.NewRequest(http.MethodGet, l.url, nil) + if err != nil { + return nil, err + } + + if tag != "" { + req.Header.Add("If-None-Match", tag) + } + + req = req.WithContext(ctx) + if err := l.prepareRequest(req); err != nil { + return nil, err + } + + resp, err := l.client.Do(req) + if err != nil { + return nil, err + } + + defer l.close(resp) + + switch resp.StatusCode { + case http.StatusOK: + // TODO: Cut the dependency to the OPA bundle package. + + b, err := bundle.NewReader(resp.Body).Read() + if err != nil { + return nil, err + } + + l.tag = resp.Header.Get("ETag") + return &b, nil + + case http.StatusNotModified: + return nil, nil + case http.StatusUnauthorized: + return nil, fmt.Errorf("not authorized (401)") + case http.StatusForbidden: + return nil, fmt.Errorf("forbidden (403)") + case http.StatusNotFound: + return nil, fmt.Errorf("not found (404)") + default: + return nil, fmt.Errorf("unknown HTTP status %v", resp.StatusCode) + } +} + +// close closes the HTTP response gracefully, first draining it, to +// avoid resource leaks. +func (l *Loader) close(resp *http.Response) { + io.Copy(ioutil.Discard, resp.Body) // Ignore errors. + resp.Body.Close() +} diff --git a/internal/wasm/sdk/opa/http/loader_test.go b/internal/wasm/sdk/opa/http/loader_test.go new file mode 100644 index 0000000000..0c92315886 --- /dev/null +++ b/internal/wasm/sdk/opa/http/loader_test.go @@ -0,0 +1,125 @@ +// Copyright 2020 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 http + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "reflect" + "sync" + "testing" + "time" + + "github.com/open-policy-agent/opa/bundle" +) + +func TestFileLoader(t *testing.T) { + // Start loader, without having the HTTP content in place. + + var pd testPolicyData + loader, err := new(&pd).WithURL("http://localhost:0").WithInterval(10*time.Millisecond, 20*time.Millisecond).Init() + if err != nil { + t.Fatal(err.Error()) + } + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(10 * time.Millisecond) + cancel() + }() + + if err := loader.Start(ctx); err != context.Canceled { + t.Fatalf("missing file not resulting in a correct error: %v", err) + } + + // Start again, with the HTTP content in place. + + var mutex sync.Mutex + policy := "wasm-policy" + var data interface{} = map[string]interface{}{ + "foo": "bar", + } + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mutex.Lock() + defer mutex.Unlock() + + if err := bundle.Write(w, bundle.Bundle{ + Data: data.(map[string]interface{}), + Wasm: []byte(policy), + }); err != nil { + panic(err) + } + })) + defer ts.Close() + + loader, err = new(&pd).WithURL(ts.URL).WithInterval(10*time.Millisecond, 20*time.Millisecond).Init() + if err != nil { + t.Fatal(err.Error()) + } + + ctx = context.Background() + if err := loader.Start(ctx); err != nil { + t.Fatalf("unable to start loader: %v", err) + } + + pd.CheckEqual(t, policy, &data) + + // Reload with updated contents. + + mutex.Lock() + policy = "wasm-policy-modified" + data = map[string]interface{}{ + "bar": "foo", + } + mutex.Unlock() + + pd.WaitUpdate() + pd.CheckEqual(t, policy, &data) + + loader.Close() +} + +type testPolicyData struct { + sync.Mutex + policy []byte + data *interface{} + updated chan struct{} +} + +func (pd *testPolicyData) SetPolicyData(policy []byte, data *interface{}) error { + pd.Lock() + defer pd.Unlock() + + pd.policy = policy + pd.data = data + if pd.updated != nil { + close(pd.updated) + } + + return nil +} + +func (pd *testPolicyData) CheckEqual(t *testing.T, policy string, data *interface{}) { + pd.Lock() + defer pd.Unlock() + + if !bytes.Equal([]byte(policy), pd.policy) && reflect.DeepEqual(data, pd.data) { + t.Fatal("policy/data mismatch.") + } +} + +func (pd *testPolicyData) WaitUpdate() { + pd.Lock() + pd.updated = make(chan struct{}) + pd.Unlock() + + <-pd.updated + + pd.Lock() + pd.updated = nil + pd.Unlock() +} diff --git a/internal/wasm/sdk/opa/http/util.go b/internal/wasm/sdk/opa/http/util.go new file mode 100644 index 0000000000..6585f5c040 --- /dev/null +++ b/internal/wasm/sdk/opa/http/util.go @@ -0,0 +1,42 @@ +// Copyright 2020 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 http + +import ( + "math/rand" + "time" +) + +// defaultBackoff returns a delay with an exponential backoff based on the +// number of retries. +func defaultBackoff(base, max float64, retries int) time.Duration { + return backoff(base, max, .2, 1.6, retries) +} + +// backoff returns a delay with an exponential backoff based on the number of +// retries. Same algorithm used in gRPC. +func backoff(base, max, jitter, factor float64, retries int) time.Duration { + if retries == 0 { + return 0 + } + + backoff, max := float64(base), float64(max) + for backoff < max && retries > 0 { + backoff *= factor + retries-- + } + if backoff > max { + backoff = max + } + + // Randomize backoff delays so that if a cluster of requests start at + // the same time, they won't operate in lockstep. + backoff *= 1 + jitter*(rand.Float64()*2-1) + if backoff < 0 { + return 0 + } + + return time.Duration(backoff) +} diff --git a/internal/wasm/sdk/opa/loader.go b/internal/wasm/sdk/opa/loader.go new file mode 100644 index 0000000000..8d9f5958b7 --- /dev/null +++ b/internal/wasm/sdk/opa/loader.go @@ -0,0 +1,21 @@ +// Copyright 2020 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 opa + +import ( + "context" +) + +// Loader is the interface all bundle loaders implement. +type Loader interface { + // Load loads a bundle. This can be invoked without starting the polling. + Load(ctx context.Context) error + + // Start starts the bundle polling. + Start(ctx context.Context) error + + // Close stops the polling. + Close() +} diff --git a/internal/wasm/sdk/opa/opa.go b/internal/wasm/sdk/opa/opa.go new file mode 100644 index 0000000000..2588ab3145 --- /dev/null +++ b/internal/wasm/sdk/opa/opa.go @@ -0,0 +1,203 @@ +// Copyright 2020 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 opa + +import ( + "context" + "encoding/json" + "fmt" + "runtime" + "sync" +) + +// OPA executes WebAssembly compiled Rego policies. +type OPA struct { + configErr error // Delayed configuration error, if any. + memoryMinPages uint32 + memoryMaxPages uint32 // 0 means no limit. + poolSize uint32 + pool *pool + mutex sync.Mutex // To serialize access to SetPolicy, SetData and Close. + policy []byte // Current policy. + data []byte // Current data. + logError func(error) +} + +// Result holds the evaluation result. +type Result struct { + Result interface{} +} + +// New constructs a new OPA SDK instance, ready to be configured with +// With functions. If no policy is provided as a part of +// configuration, policy (and data) needs to be set before invoking +// Eval. Once constructed and configured, the instance needs to be +// initialized before invoking the Eval. +func New() *OPA { + opa := &OPA{ + memoryMinPages: 2, + memoryMaxPages: 0, + poolSize: uint32(runtime.GOMAXPROCS(0)), + logError: func(error) {}, + } + + return opa +} + +// Init initializes the SDK instance after the construction and +// configuration. If the configuration is invalid, it returns +// ErrInvalidConfig. +func (o *OPA) Init() (*OPA, error) { + if o.configErr != nil { + return nil, o.configErr + } + + o.pool = newPool(o.poolSize, o.memoryMinPages, o.memoryMaxPages) + + if len(o.policy) != 0 { + if err := o.pool.SetPolicyData(o.policy, o.data); err != nil { + return nil, err + } + } + + return o, nil +} + +// SetData updates the data for the subsequent Eval calls. Returns +// either ErrNotReady, ErrInvalidPolicyOrData, or ErrInternal if an +// error occurs. +func (o *OPA) SetData(v interface{}) error { + if o.pool == nil { + return ErrNotReady + } + + raw, err := json.Marshal(v) + if err != nil { + return fmt.Errorf("%v: %w", err, ErrInvalidPolicyOrData) + } + + o.mutex.Lock() + defer o.mutex.Unlock() + + return o.setPolicyData(o.policy, raw) +} + +// SetPolicy updates the policy for the subsequent Eval calls. +// Returns either ErrNotReady, ErrInvalidPolicy or ErrInternal if an +// error occurs. +func (o *OPA) SetPolicy(p []byte) error { + if o.pool == nil { + return ErrNotReady + } + + o.mutex.Lock() + defer o.mutex.Unlock() + + return o.setPolicyData(p, o.data) +} + +// SetPolicyData updates both the policy and data for the subsequent +// Eval calls. Returns either ErrNotReady, ErrInvalidPolicyOrData, or +// ErrInternal if an error occurs. +func (o *OPA) SetPolicyData(policy []byte, data *interface{}) error { + if o.pool == nil { + return ErrNotReady + } + + var raw []byte + if data != nil { + var err error + raw, err = json.Marshal(*data) + if err != nil { + return fmt.Errorf("%v: %w", err, ErrInvalidPolicyOrData) + } + } + + o.mutex.Lock() + defer o.mutex.Unlock() + + return o.setPolicyData(policy, raw) +} + +func (o *OPA) setPolicyData(policy []byte, data []byte) error { + if err := o.pool.SetPolicyData(policy, data); err != nil { + return err + } + + o.policy = policy + o.data = data + return nil +} + +// Eval evaluates the policy with the given input, returning the +// evaluation results. If no policy was configured at construction +// time nor set after, the function returns ErrNotReady. It returns +// ErrInternal if any other error occurs. +func (o *OPA) Eval(ctx context.Context, input *interface{}) (*Result, error) { + if o.pool == nil { + return nil, ErrNotReady + } + + instance, err := o.pool.Acquire(ctx) + if err != nil { + return nil, err + } + + defer o.pool.Release(instance) + + result, err := instance.Eval(ctx, input) + if err != nil { + return nil, fmt.Errorf("%v: %w", err, ErrInternal) + } + + return &Result{result}, nil +} + +// Close waits until all the pending evaluations complete and then +// releases all the resources allocated. Eval will return ErrClosed +// afterwards. +func (o *OPA) Close() { + if o.pool == nil { + return + } + + o.mutex.Lock() + defer o.mutex.Unlock() + + o.pool.Close() +} + +// EvalBool evaluates the boolean policy with the given input. The +// possible error values returned are as with Eval with addition of +// ErrUndefined indicating an undefined policy decision and +// ErrNonBoolean indicating a non-boolean policy decision. +func EvalBool(ctx context.Context, o *OPA, input *interface{}) (bool, error) { + rs, err := o.Eval(ctx, input) + if err != nil { + return false, err + } + + r, ok := rs.Result.([]interface{}) + if !ok || len(r) == 0 { + return false, ErrUndefined + } + + m, ok := r[0].(map[string]interface{}) + if !ok || len(m) != 1 { + return false, ErrNonBoolean + } + + var b bool + for _, v := range m { + b, ok = v.(bool) + break + } + + if !ok { + return false, ErrNonBoolean + } + + return b, nil +} diff --git a/internal/wasm/sdk/opa/opa_test.go b/internal/wasm/sdk/opa/opa_test.go new file mode 100644 index 0000000000..0a913d8288 --- /dev/null +++ b/internal/wasm/sdk/opa/opa_test.go @@ -0,0 +1,226 @@ +// Copyright 2020 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 opa_test + +import ( + "context" + "fmt" + "reflect" + "testing" + + "github.com/open-policy-agent/opa/internal/wasm/sdk/opa" + "github.com/open-policy-agent/opa/rego" + "github.com/open-policy-agent/opa/util" +) + +func TestOPA(t *testing.T) { + type Eval struct { + NewPolicy string + NewData string + Input string + Result string + } + + tests := []struct { + Description string + Policy string + Query string + Data string + Evals []Eval + }{ + { + Description: "No input, no data, static policy", + Policy: `a = true`, + Query: "data.p.a = x", + Evals: []Eval{ + Eval{Result: `[{"x": true}]`}, + Eval{Result: `[{"x": true}]`}, + }, + }, + { + Description: "Only input changing", + Policy: `a = input`, + Query: "data.p.a = x", + Evals: []Eval{ + Eval{Input: "false", Result: `[{"x": false}]`}, + Eval{Input: "true", Result: `[{"x": true}]`}, + }, + }, + { + Description: "Only data changing", + Policy: `a = data.q`, + Query: "data.p.a = x", + Data: `{"q": false}`, + Evals: []Eval{ + Eval{Result: `[{"x": false}]`}, + Eval{NewData: `{"q": true}`, Result: `[{"x": true}]`}, + }, + }, + { + Description: "Only policy changing", + Policy: `a = data.q`, + Query: "data.p.a = x", + Data: `{"q": false, "r": true}`, + Evals: []Eval{ + Eval{Result: `[{"x": false}]`}, + Eval{NewPolicy: `a = data.r`, Result: `[{"x": true}]`}, + }, + }, + { + Description: "Policy and data changing", + Policy: `a = data.q`, + Query: "data.p.a = x", + Data: `{"q": 0, "r": 1}`, + Evals: []Eval{ + Eval{Result: `[{"x": 0}]`}, + Eval{NewPolicy: `a = data.r`, NewData: `{"q": 2, "r": 3}`, Result: `[{"x": 3}]`}, + }, + }, + { + Description: "Builtins", + Policy: `a = count(data.q) + sum(data.q)`, + Query: "data.p.a = x", + Evals: []Eval{ + Eval{NewData: `{"q": []}`, Result: `[{"x": 0}]`}, + Eval{NewData: `{"q": [1, 2]}`, Result: `[{"x": 5}]`}, + }, + }, + { + Description: "Undefined decision", + Policy: `a = true`, + Query: "data.p.b = x", + Evals: []Eval{ + Eval{Result: `[]`}, + }, + }, + } + + for _, test := range tests { + t.Run(test.Description, func(t *testing.T) { + policy := compileRegoToWasm(test.Policy, test.Query) + data := []byte(test.Data) + if len(data) == 0 { + data = nil + } + opa, err := opa.New(). + WithPolicyBytes(policy). + WithDataBytes(data). + WithMemoryLimits(131070, 0). + WithPoolSize(1). // Minimal pool size to test pooling. + Init() + if err != nil { + t.Fatal(err) + } + + // Execute each requested policy evaluation, with their inputs and updating data if requested. + + for _, eval := range test.Evals { + switch { + case eval.NewPolicy != "" && eval.NewData != "": + policy := compileRegoToWasm(eval.NewPolicy, test.Query) + data := parseJSON(eval.NewData) + if err := opa.SetPolicyData(policy, data); err != nil { + t.Errorf(err.Error()) + } + + case eval.NewPolicy != "": + policy := compileRegoToWasm(eval.NewPolicy, test.Query) + if err := opa.SetPolicy(policy); err != nil { + t.Errorf(err.Error()) + } + + case eval.NewData != "": + data := parseJSON(eval.NewData) + if err := opa.SetData(*data); err != nil { + t.Errorf(err.Error()) + } + } + + result, err := opa.Eval(context.Background(), parseJSON(eval.Input)) + if err != nil { + t.Errorf(err.Error()) + } + + if !reflect.DeepEqual(*parseJSON(eval.Result), result.Result) { + t.Errorf("Incorrect result: %v", result.Result) + } + } + + opa.Close() + }) + } +} + +func BenchmarkWasmRego(b *testing.B) { + policy := compileRegoToWasm("a = true", "data.p.a = x") + opa, _ := opa.New(). + WithPolicyBytes(policy). + WithMemoryLimits(131070, 2*131070). // TODO: For some reason unlimited memory slows down the eval_ctx_new(). + WithPoolSize(1). + Init() + + b.ReportAllocs() + b.ResetTimer() + + ctx := context.Background() + var input interface{} = make(map[string]interface{}) + + for i := 0; i < b.N; i++ { + if _, err := opa.Eval(ctx, &input); err != nil { + panic(err) + } + } +} + +func BenchmarkGoRego(b *testing.B) { + pq := compileRego(`package p + +a = true`, "data.p.a = x") + + b.ReportAllocs() + b.ResetTimer() + + input := make(map[string]interface{}) + + for i := 0; i < b.N; i++ { + if _, err := pq.Eval(context.Background(), rego.EvalInput(input)); err != nil { + panic(err) + } + } +} + +func compileRegoToWasm(policy string, query string) []byte { + module := fmt.Sprintf("package p\n%s", policy) + cr, err := rego.New( + rego.Query(query), + rego.Module("module.rego", module), + ).Compile(context.Background(), rego.CompilePartial(false)) + if err != nil { + panic(err) + } + + return cr.Bytes +} + +func compileRego(module string, query string) rego.PreparedEvalQuery { + rego := rego.New( + rego.Query(query), + rego.Module("module.rego", module), + ) + pq, err := rego.PrepareForEval(context.Background()) + if err != nil { + panic(err) + } + + return pq +} + +func parseJSON(s string) *interface{} { + if s == "" { + return nil + } + + v := util.MustUnmarshalJSON([]byte(s)) + return &v +} diff --git a/internal/wasm/sdk/opa/pool.go b/internal/wasm/sdk/opa/pool.go new file mode 100644 index 0000000000..819a193cbd --- /dev/null +++ b/internal/wasm/sdk/opa/pool.go @@ -0,0 +1,262 @@ +// Copyright 2020 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 opa + +import ( + "bytes" + "context" + "fmt" + "sync" +) + +// pool maintains a pool of WebAssemly VM instances. +type pool struct { + available chan struct{} + mutex sync.Mutex + initialized bool + closed bool + policy []byte + data []byte + memoryMinPages uint32 + memoryMaxPages uint32 + vms []*vm // All current VM instances, acquired or not. + acquired []bool + pendingReinit *vm + blockedReinit chan struct{} +} + +// newPool constructs a new pool with the pool and VM configuration provided. +func newPool(poolSize, memoryMinPages, memoryMaxPages uint32) *pool { + available := make(chan struct{}, poolSize) + for i := uint32(0); i < poolSize; i++ { + available <- struct{}{} + } + + return &pool{ + memoryMinPages: memoryMinPages, + memoryMaxPages: memoryMaxPages, + available: available, + vms: make([]*vm, 0), + acquired: make([]bool, 0), + } +} + +// Acquire obtains a VM from the pool, waiting if all VMms are in use +// and building one as necessary. Returns either ErrNotReady or +// ErrInternal if an error. +func (p *pool) Acquire(ctx context.Context) (*vm, error) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-p.available: + } + + p.mutex.Lock() + defer p.mutex.Unlock() + + if !p.initialized || p.closed { + return nil, ErrNotReady + } + + for i, vm := range p.vms { + if !p.acquired[i] { + p.acquired[i] = true + return vm, nil + } + } + + policy, data := p.policy, p.data + + p.mutex.Unlock() + vm, err := newVM(policy, data, p.memoryMinPages, p.memoryMaxPages) + p.mutex.Lock() + + if err != nil { + p.available <- struct{}{} + return nil, fmt.Errorf("%v: %w", err, ErrInternal) + } + + p.acquired = append(p.acquired, true) + p.vms = append(p.vms, vm) + return vm, nil +} + +// Release releases the VM back to the pool. +func (p *pool) Release(vm *vm) { + p.mutex.Lock() + + // If the policy data setting is waiting for this one, don't release it back to the general consumption. + // Note the reinit is responsible for pushing to available channel once done with the VM. + if vm == p.pendingReinit { + p.mutex.Unlock() + p.blockedReinit <- struct{}{} + return + } + + for i := range p.vms { + if p.vms[i] == vm { + p.acquired[i] = false + p.mutex.Unlock() + p.available <- struct{}{} + return + } + } + + // VM instance not found anymore, hence pool reconfigured and can release the VM. + + p.mutex.Unlock() + p.available <- struct{}{} + + vm.Close() +} + +// Reset re-initializes the vms within the pool with the new policy +// and data. The re-initialization takes place atomically: all new vms +// are constructed in advance before touching the pool. Returns +// either ErrNotReady, ErrInvalidPolicy or ErrInternal if an error +// occurs. +func (p *pool) SetPolicyData(policy []byte, data []byte) error { + p.mutex.Lock() + + if !p.initialized { + vm, err := newVM(policy, data, p.memoryMinPages, p.memoryMaxPages) + if err == nil { + p.initialized = true + p.vms = append(p.vms, vm) + p.acquired = append(p.acquired, false) + p.policy, p.data = policy, data + } else { + err = fmt.Errorf("%v: %w", err, ErrInvalidPolicyOrData) + } + + p.mutex.Unlock() + return err + } + + if p.closed { + p.mutex.Unlock() + return ErrNotReady + } + + currentPolicy, currentData := p.policy, p.data + p.mutex.Unlock() + + if bytes.Equal(policy, currentPolicy) && bytes.Equal(data, currentData) { + return nil + + } + + err := p.setPolicyData(policy, data) + if err != nil { + return fmt.Errorf("%v: %w", err, ErrInternal) + } + + return nil +} + +// setPolicyData reinitializes the VMs one at a time. +func (p *pool) setPolicyData(policy []byte, data []byte) error { + for i, activations := 0, 0; true; i++ { + vm := p.wait(i) + if vm == nil { + // All have been converted. + return nil + } + + if err := vm.SetPolicyData(policy, data); err != nil { + // No guarantee about the VM state after an error; hence, remove. + p.remove(i) + p.Release(vm) + + // After the first successful activation, proceed through all the VMs, ignoring the remaining errors. + if activations == 0 { + return err + } + } else { + p.Release(vm) + } + + // Activate the policy and data, now that a single VM has been reset without errors. + + if activations == 0 { + p.activate(policy, data) + } + + activations++ + } + + return nil +} + +// Close waits for all the evaluations to finish and then releases the VMs. +func (p *pool) Close() { + for range p.vms { + <-p.available + } + + p.mutex.Lock() + defer p.mutex.Unlock() + + for _, vm := range p.vms { + if vm != nil { + vm.Close() + } + } + + p.closed = true + p.vms = nil +} + +// wait steals the i'th VM instance. The VM has to be released afterwards. +func (p *pool) wait(i int) *vm { + p.mutex.Lock() + defer p.mutex.Unlock() + + if i == len(p.vms) { + return nil + } + + vm := p.vms[i] + isActive := p.acquired[i] + p.acquired[i] = true + + if isActive { + p.blockedReinit = make(chan struct{}, 1) + p.pendingReinit = vm + } + + p.mutex.Unlock() + + if isActive { + <-p.blockedReinit + } else { + <-p.available + } + + p.mutex.Lock() + p.pendingReinit = nil + return vm +} + +// remove removes the i'th vm. +func (p *pool) remove(i int) { + p.mutex.Lock() + defer p.mutex.Unlock() + + n := len(p.vms) + if n > 1 { + p.vms[i] = p.vms[n-1] + } + + p.vms = p.vms[0 : n-1] + p.acquired = p.acquired[0 : n-1] +} + +func (p *pool) activate(policy []byte, data []byte) { + p.mutex.Lock() + defer p.mutex.Unlock() + + p.policy, p.data = policy, data +} diff --git a/internal/wasm/sdk/opa/vm.go b/internal/wasm/sdk/opa/vm.go new file mode 100644 index 0000000000..a029b0eb18 --- /dev/null +++ b/internal/wasm/sdk/opa/vm.go @@ -0,0 +1,416 @@ +// Copyright 2020 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 opa + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "time" + + "github.com/open-policy-agent/opa/ast" + "github.com/open-policy-agent/opa/metrics" + "github.com/open-policy-agent/opa/topdown" + "github.com/open-policy-agent/opa/topdown/builtins" + wasm "github.com/wasmerio/go-ext-wasm/wasmer" +) + +type vm struct { + instance *wasm.Instance // Pointer to avoid unintented destruction (triggering finalizers within). + policy []byte + data []byte + memory *wasm.Memory + memoryMin uint32 + memoryMax uint32 + bctx *topdown.BuiltinContext + builtins map[int32]topdown.BuiltinFunc + builtinResult *ast.Term + baseHeapPtr int32 + dataAddr int32 + evalHeapPtr int32 + evalHeapTop int32 + eval func(...interface{}) (wasm.Value, error) + evalCtxGetResult func(...interface{}) (wasm.Value, error) + evalCtxNew func(...interface{}) (wasm.Value, error) + evalCtxSetData func(...interface{}) (wasm.Value, error) + evalCtxSetInput func(...interface{}) (wasm.Value, error) + free func(...interface{}) (wasm.Value, error) + heapPtrGet func(...interface{}) (wasm.Value, error) + heapPtrSet func(...interface{}) (wasm.Value, error) + heapTopGet func(...interface{}) (wasm.Value, error) + heapTopSet func(...interface{}) (wasm.Value, error) + jsonDump func(...interface{}) (wasm.Value, error) + jsonParse func(...interface{}) (wasm.Value, error) + malloc func(...interface{}) (wasm.Value, error) +} + +func newVM(policy []byte, data []byte, memoryMin, memoryMax uint32) (*vm, error) { + memory, err := wasm.NewMemory(memoryMin, memoryMax) + if err != nil { + return nil, err + } + + imports, err := opaFunctions(wasm.NewImports()) + if err != nil { + return nil, err + } + + imports, err = imports.AppendMemory("memory", memory) + if err != nil { + panic(err) + } + + i, err := wasm.NewInstanceWithImports(policy, imports) + if err != nil { + return nil, err + } + + v := &vm{ + instance: &i, + policy: policy, + data: data, + memory: memory, + memoryMin: memoryMin, + memoryMax: memoryMax, + builtins: make(map[int32]topdown.BuiltinFunc), + dataAddr: 0, + eval: i.Exports["eval"], + evalCtxGetResult: i.Exports["opa_eval_ctx_get_result"], + evalCtxNew: i.Exports["opa_eval_ctx_new"], + evalCtxSetData: i.Exports["opa_eval_ctx_set_data"], + evalCtxSetInput: i.Exports["opa_eval_ctx_set_input"], + free: i.Exports["opa_free"], + heapPtrGet: i.Exports["opa_heap_ptr_get"], + heapPtrSet: i.Exports["opa_heap_ptr_set"], + heapTopGet: i.Exports["opa_heap_top_get"], + heapTopSet: i.Exports["opa_heap_top_set"], + jsonDump: i.Exports["opa_json_dump"], + jsonParse: i.Exports["opa_json_parse"], + malloc: i.Exports["opa_malloc"], + } + + // Initialize the heap. + + if _, err := v.malloc(0); err != nil { + return nil, err + } + + if v.baseHeapPtr, err = v.getHeapState(); err != nil { + return nil, err + } + + if data != nil { + if v.dataAddr, err = v.toRegoJSON(data, true); err != nil { + return nil, err + } + } + + if v.evalHeapPtr, err = v.getHeapState(); err != nil { + return nil, err + } + + // For the opa builtin functions to access the instance. + i.SetContextData(v) + + // Construct the builtin id to name mappings. + + val, err := i.Exports["builtins"]() + if err != nil { + return nil, err + } + + builtins, err := v.fromRegoJSON(val.ToI32(), true) + if err != nil { + return nil, err + } + + for name, id := range builtins.(map[string]interface{}) { + f := topdown.GetBuiltin(name) + if f == nil { + return nil, fmt.Errorf("builtin '%s' not found", name) + } + + n, err := id.(json.Number).Int64() + if err != nil { + panic(err) + } + + v.builtins[int32(n)] = f + } + + return v, nil +} + +func (i *vm) Eval(ctx context.Context, input *interface{}) (interface{}, error) { + if err := i.setHeapState(i.evalHeapPtr); err != nil { + return nil, err + } + + defer func() { + i.bctx = nil + }() + + // Parse the input JSON and activate it with the data. + + addr, err := i.evalCtxNew() + if err != nil { + return nil, err + } + + ctxAddr := addr.ToI32() + + if i.dataAddr != 0 { + if _, err := i.evalCtxSetData(ctxAddr, i.dataAddr); err != nil { + return nil, err + } + } + + if input != nil { + inputAddr, err := i.toRegoJSON(*input, false) + if err != nil { + return nil, err + } + + if _, err := i.evalCtxSetInput(ctxAddr, inputAddr); err != nil { + return nil, err + } + } + + // Evaluate the policy. + func() { + defer func() { + if e := recover(); e != nil { + switch e := e.(type) { + case abortError: + err = errors.New(e.message) + case builtinError: + err = e.err + default: + panic(e) + } + + } + }() + _, err = i.eval(ctxAddr) + }() + + if err != nil { + return nil, err + } + + resultAddr, err := i.evalCtxGetResult(ctxAddr) + if err != nil { + return nil, err + } + + result, err := i.fromRegoJSON(resultAddr.ToI32(), false) + + // Skip free'ing input and result JSON as the heap will be reset next round anyway. + + return result, err +} + +func (i *vm) SetPolicyData(policy []byte, data []byte) error { + if !bytes.Equal(policy, i.policy) { + // Swap the instance to a new one, with new policy. + + n, err := newVM(policy, data, i.memoryMin, i.memoryMax) + if err != nil { + return err + } + + i.Close() + + *i = *n + return nil + } + + i.data = data + i.dataAddr = 0 + + var err error + if err = i.setHeapState(i.baseHeapPtr); err != nil { + return err + } + + if data != nil { + if i.dataAddr, err = i.toRegoJSON(data, true); err != nil { + return err + } + } + + if i.evalHeapPtr, err = i.getHeapState(); err != nil { + return err + } + + return nil +} + +func (i *vm) Close() { + i.memory.Close() + i.instance.Close() +} + +type abortError struct { + message string +} + +// Abort is invoked by the policy if an internal error occurs during +// the policy execution. +func (i *vm) Abort(arg int32) { + data := i.memory.Data()[arg:] + n := bytes.IndexByte(data, 0) + if n == -1 { + panic("invalid abort argument") + } + + panic(abortError{message: string(data[0:n])}) +} + +type builtinError struct { + err error +} + +// Builtin executes a builtin for the policy. +func (i *vm) Builtin(builtinID, ctx int32, args ...int32) int32 { + + // TODO: Returning proper errors instead of panicing. + // TODO: To avoid growing the heap with every built-in call, recycle the JSON buffers since the free implementation is no-op. + + convertedArgs := make([]*ast.Term, len(args)) + for j, arg := range args { + x, err := i.fromRegoJSON(arg, true) + if err != nil { + panic(builtinError{err: err}) + } + + y, err := ast.InterfaceToValue(x) + if err != nil { + panic(builtinError{err: err}) + } + + convertedArgs[j] = ast.NewTerm(y) + } + + if i.bctx == nil { + i.bctx = &topdown.BuiltinContext{ + Context: context.Background(), + Cancel: nil, + Runtime: nil, + Time: ast.NumberTerm(json.Number(strconv.FormatInt(time.Now().UnixNano(), 10))), + Metrics: metrics.New(), + Cache: make(builtins.Cache), + Location: nil, + Tracers: nil, + QueryID: 0, + ParentID: 0, + } + } + + err := i.builtins[builtinID](*i.bctx, convertedArgs, i.iter) + if err != nil { + panic(builtinError{err: err}) + } + + result, err := ast.JSON(i.builtinResult.Value) + if err != nil { + panic(builtinError{err: err}) + } + + addr, err := i.toRegoJSON(result, true) + if err != nil { + panic(builtinError{err: err}) + } + + return addr +} + +func (i *vm) iter(result *ast.Term) error { + i.builtinResult = result + return nil +} + +// fromRegoJSON converts Rego JSON to go native JSON. +func (i *vm) fromRegoJSON(addr int32, free bool) (interface{}, error) { + serialized, err := i.jsonDump(addr) + if err != nil { + return nil, err + } + + data := i.memory.Data()[serialized.ToI32():] + n := bytes.IndexByte(data, 0) + if n < 0 { + n = 0 + } + + // Parse the result into go types. + + decoder := json.NewDecoder(bytes.NewReader(data[0:n])) + decoder.UseNumber() + + var result interface{} + if err := decoder.Decode(&result); err != nil { + return nil, err + } + + if free { + if _, err := i.free(serialized.ToI32()); err != nil { + return nil, err + } + } + + return result, nil +} + +// toRegoJSON converts go native JSON to Rego JSON. +func (i *vm) toRegoJSON(v interface{}, free bool) (int32, error) { + raw, ok := v.([]byte) + if !ok { + var err error + raw, err = json.Marshal(v) + if err != nil { + return 0, err + } + } + + n := int32(len(raw)) + pos, err := i.malloc(n) + if err != nil { + return 0, err + } + + p := pos.ToI32() + copy(i.memory.Data()[p:p+n], raw) + + addr, err := i.jsonParse(p, n) + if err != nil { + return 0, err + } + + if free { + if _, err := i.free(p); err != nil { + return 0, err + } + } + + return addr.ToI32(), nil +} + +func (i *vm) getHeapState() (int32, error) { + ptr, err := i.heapPtrGet() + if err != nil { + return 0, err + } + + return ptr.ToI32(), nil +} + +func (i *vm) setHeapState(ptr int32) error { + _, err := i.heapPtrSet(ptr) + return err +}