mirror of
https://github.com/open-policy-agent/opa.git
synced 2026-08-12 19:32:48 -06:00
Preparing for v1 API
Moving (most) source to v1 root package to prepare for v0/v1 API separation. Signed-off-by: Johan Fylling <johan.dev@fylling.se>
This commit is contained in:
Johan Fylling
@@ -1,84 +0,0 @@
|
||||
// Copyright 2018 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 download
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/open-policy-agent/opa/plugins"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultMinDelaySeconds = int64(60)
|
||||
defaultMaxDelaySeconds = int64(120)
|
||||
|
||||
// deltaBundleMode indicates that OPA supports delta bundle processing
|
||||
deltaBundleMode = "delta"
|
||||
|
||||
// defaultBundleMode indicates that OPA supports snapshot bundle processing
|
||||
defaultBundleMode = "snapshot"
|
||||
)
|
||||
|
||||
// PollingConfig represents polling configuration for the downloader.
|
||||
type PollingConfig struct {
|
||||
MinDelaySeconds *int64 `json:"min_delay_seconds,omitempty"` // min amount of time to wait between successful poll attempts
|
||||
MaxDelaySeconds *int64 `json:"max_delay_seconds,omitempty"` // max amount of time to wait between poll attempts
|
||||
LongPollingTimeoutSeconds *int64 `json:"long_polling_timeout_seconds,omitempty"` // max amount of time the server should wait before issuing a timeout if there's no update available
|
||||
}
|
||||
|
||||
// Config represents the configuration for the downloader.
|
||||
type Config struct {
|
||||
Trigger *plugins.TriggerMode `json:"trigger,omitempty"`
|
||||
Polling PollingConfig `json:"polling"`
|
||||
}
|
||||
|
||||
// ValidateAndInjectDefaults checks for configuration errors and ensures all
|
||||
// values are set on the Config object.
|
||||
func (c *Config) ValidateAndInjectDefaults() error {
|
||||
|
||||
if c.Trigger == nil {
|
||||
t := plugins.DefaultTriggerMode
|
||||
c.Trigger = &t
|
||||
}
|
||||
|
||||
switch *c.Trigger {
|
||||
case plugins.TriggerPeriodic, plugins.TriggerManual:
|
||||
break
|
||||
default:
|
||||
return fmt.Errorf("invalid trigger mode %q (want %q or %q)", *c.Trigger, plugins.TriggerPeriodic, plugins.TriggerManual)
|
||||
}
|
||||
|
||||
min := defaultMinDelaySeconds
|
||||
max := defaultMaxDelaySeconds
|
||||
|
||||
// reject bad min/max values
|
||||
if c.Polling.MaxDelaySeconds != nil && c.Polling.MinDelaySeconds != nil {
|
||||
if *c.Polling.MaxDelaySeconds < *c.Polling.MinDelaySeconds {
|
||||
return fmt.Errorf("max polling delay must be >= min polling delay")
|
||||
}
|
||||
min = *c.Polling.MinDelaySeconds
|
||||
max = *c.Polling.MaxDelaySeconds
|
||||
} else if c.Polling.MaxDelaySeconds == nil && c.Polling.MinDelaySeconds != nil {
|
||||
return fmt.Errorf("polling configuration missing 'max_delay_seconds'")
|
||||
} else if c.Polling.MinDelaySeconds == nil && c.Polling.MaxDelaySeconds != nil {
|
||||
return fmt.Errorf("polling configuration missing 'min_delay_seconds'")
|
||||
}
|
||||
|
||||
// scale to seconds
|
||||
minSeconds := int64(time.Duration(min) * time.Second)
|
||||
c.Polling.MinDelaySeconds = &minSeconds
|
||||
|
||||
maxSeconds := int64(time.Duration(max) * time.Second)
|
||||
c.Polling.MaxDelaySeconds = &maxSeconds
|
||||
|
||||
if c.Polling.LongPollingTimeoutSeconds != nil {
|
||||
if *c.Polling.LongPollingTimeoutSeconds < 1 {
|
||||
return fmt.Errorf("'long_polling_timeout_seconds' must be at least 1")
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -1,100 +0,0 @@
|
||||
// Copyright 2018 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 download
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestConfigValidation(t *testing.T) {
|
||||
|
||||
tests := []struct {
|
||||
note string
|
||||
input string
|
||||
wantErr bool
|
||||
expMin time.Duration
|
||||
expMax time.Duration
|
||||
}{
|
||||
{
|
||||
note: "min > max",
|
||||
input: `{
|
||||
"polling": {
|
||||
"min_delay_seconds": 10,
|
||||
"max_delay_seconds": 1
|
||||
}
|
||||
}`,
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
note: "empty",
|
||||
input: `{}`,
|
||||
expMin: time.Second * time.Duration(defaultMinDelaySeconds),
|
||||
expMax: time.Second * time.Duration(defaultMaxDelaySeconds),
|
||||
},
|
||||
{
|
||||
note: "min missing",
|
||||
input: `{
|
||||
"polling": {
|
||||
"max_delay_seconds": 10
|
||||
}
|
||||
}`,
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
note: "max missing",
|
||||
input: `{
|
||||
"polling": {
|
||||
"min_delay_seconds": 1
|
||||
}
|
||||
}`,
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
note: "user supplied",
|
||||
input: `{
|
||||
"polling": {
|
||||
"min_delay_seconds": 10,
|
||||
"max_delay_seconds": 30
|
||||
}
|
||||
}`,
|
||||
expMin: time.Second * 10,
|
||||
expMax: time.Second * 30,
|
||||
},
|
||||
{
|
||||
note: "long polling timeout < 1",
|
||||
input: `{
|
||||
"polling": {
|
||||
"long_polling_timeout_seconds": 0
|
||||
}
|
||||
}`,
|
||||
wantErr: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
|
||||
var config Config
|
||||
|
||||
if err := json.Unmarshal([]byte(test.input), &config); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err := config.ValidateAndInjectDefaults()
|
||||
if err != nil && !test.wantErr {
|
||||
t.Errorf("Unexpected error on: %v, err: %v", test.input, err)
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
if time.Duration(*config.Polling.MinDelaySeconds) != test.expMin {
|
||||
t.Errorf("For %q expected min %v but got %v", test.note, test.expMin, time.Duration(*config.Polling.MinDelaySeconds))
|
||||
}
|
||||
if time.Duration(*config.Polling.MaxDelaySeconds) != test.expMax {
|
||||
t.Errorf("For %q expected min %v but got %v", test.note, test.expMax, time.Duration(*config.Polling.MaxDelaySeconds))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,454 +0,0 @@
|
||||
// Copyright 2018 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 download implements low-level OPA bundle downloading.
|
||||
package download
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/rand"
|
||||
"net/http"
|
||||
"path"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/open-policy-agent/opa/ast"
|
||||
"github.com/open-policy-agent/opa/bundle"
|
||||
"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/rest"
|
||||
"github.com/open-policy-agent/opa/util"
|
||||
)
|
||||
|
||||
const (
|
||||
minRetryDelay = time.Millisecond * 100
|
||||
)
|
||||
|
||||
// Update contains the result of a download. If an error occurred, the Error
|
||||
// field will be non-nil. If a new bundle is available, the Bundle field will
|
||||
// be non-nil.
|
||||
type Update struct {
|
||||
ETag string
|
||||
Bundle *bundle.Bundle
|
||||
Error error
|
||||
Metrics metrics.Metrics
|
||||
Raw io.Reader
|
||||
Size int
|
||||
}
|
||||
|
||||
// Downloader implements low-level OPA bundle downloading. Downloader can be
|
||||
// started and stopped. After starting, the downloader will request bundle
|
||||
// updates from the remote HTTP endpoint that the client is configured to
|
||||
// connect to.
|
||||
type Downloader struct {
|
||||
config Config // downloader configuration for tuning polling and other downloader behaviour
|
||||
client rest.Client // HTTP client to use for bundle downloading
|
||||
path string // path to use in bundle download request
|
||||
trigger chan chan struct{} // channel to signal downloads when manual triggering is enabled
|
||||
stop chan chan struct{} // used to signal plugin to stop running
|
||||
f func(context.Context, Update) // callback function invoked when download updates occur
|
||||
etag string // HTTP Etag for caching purposes
|
||||
sizeLimitBytes *int64 // max bundle file size in bytes (passed to reader)
|
||||
bvc *bundle.VerificationConfig
|
||||
respHdrTimeoutSec int64
|
||||
wg sync.WaitGroup
|
||||
logger logging.Logger
|
||||
mtx sync.Mutex
|
||||
stopped bool
|
||||
persist bool
|
||||
longPollingEnabled bool
|
||||
lazyLoadingMode bool
|
||||
bundleName string
|
||||
bundleParserOpts ast.ParserOptions
|
||||
}
|
||||
|
||||
type downloaderResponse struct {
|
||||
b *bundle.Bundle
|
||||
raw io.Reader
|
||||
etag string
|
||||
longPoll bool
|
||||
size int
|
||||
}
|
||||
|
||||
// New returns a new Downloader that can be started.
|
||||
func New(config Config, client rest.Client, path string) *Downloader {
|
||||
return &Downloader{
|
||||
config: config,
|
||||
client: client,
|
||||
path: path,
|
||||
trigger: make(chan chan struct{}),
|
||||
stop: make(chan chan struct{}),
|
||||
logger: client.Logger(),
|
||||
longPollingEnabled: config.Polling.LongPollingTimeoutSeconds != nil,
|
||||
}
|
||||
}
|
||||
|
||||
// WithCallback registers a function f to be called when download updates occur.
|
||||
func (d *Downloader) WithCallback(f func(context.Context, Update)) *Downloader {
|
||||
d.f = f
|
||||
return d
|
||||
}
|
||||
|
||||
// WithLogAttrs sets an optional set of key/value pair attributes to include in
|
||||
// log messages emitted by the downloader.
|
||||
func (d *Downloader) WithLogAttrs(attrs map[string]interface{}) *Downloader {
|
||||
d.logger = d.logger.WithFields(attrs)
|
||||
return d
|
||||
}
|
||||
|
||||
// WithBundleVerificationConfig sets the key configuration used to verify a signed bundle
|
||||
func (d *Downloader) WithBundleVerificationConfig(config *bundle.VerificationConfig) *Downloader {
|
||||
d.bvc = config
|
||||
return d
|
||||
}
|
||||
|
||||
// WithSizeLimitBytes sets the file size limit for bundles read by this downloader.
|
||||
func (d *Downloader) WithSizeLimitBytes(n int64) *Downloader {
|
||||
d.sizeLimitBytes = &n
|
||||
return d
|
||||
}
|
||||
|
||||
// WithBundlePersistence specifies if the downloaded bundle will eventually be persisted to disk.
|
||||
func (d *Downloader) WithBundlePersistence(persist bool) *Downloader {
|
||||
d.persist = persist
|
||||
return d
|
||||
}
|
||||
|
||||
// WithLazyLoadingMode specifies how the downloaded bundle should be read.
|
||||
// If true, data files in the bundle will not be deserialized
|
||||
// and the check to validate that the bundle data does not contain paths
|
||||
// outside the bundle's roots will not be performed while reading the bundle.
|
||||
func (d *Downloader) WithLazyLoadingMode(yes bool) *Downloader {
|
||||
d.lazyLoadingMode = yes
|
||||
return d
|
||||
}
|
||||
|
||||
// WithBundleName specifies the name of the downloaded bundle.
|
||||
func (d *Downloader) WithBundleName(bundleName string) *Downloader {
|
||||
d.bundleName = bundleName
|
||||
return d
|
||||
}
|
||||
|
||||
// WithBundleParserOpts specifies the parser options to use when parsing downloaded bundles.
|
||||
func (d *Downloader) WithBundleParserOpts(opts ast.ParserOptions) *Downloader {
|
||||
d.bundleParserOpts = opts
|
||||
return d
|
||||
}
|
||||
|
||||
// ClearCache is deprecated. Use SetCache instead.
|
||||
func (d *Downloader) ClearCache() {
|
||||
d.etag = ""
|
||||
}
|
||||
|
||||
// SetCache sets the given etag value on the downloader.
|
||||
func (d *Downloader) SetCache(etag string) {
|
||||
d.etag = etag
|
||||
}
|
||||
|
||||
// Trigger can be used to control when the downloader attempts to download
|
||||
// a new bundle in manual triggering mode.
|
||||
func (d *Downloader) Trigger(ctx context.Context) error {
|
||||
done := make(chan error)
|
||||
|
||||
go func() {
|
||||
err := d.oneShot(ctx)
|
||||
if err != nil {
|
||||
d.logger.Error("Bundle download failed: %v.", err)
|
||||
if ctx.Err() == nil {
|
||||
done <- err
|
||||
}
|
||||
}
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// Start tells the Downloader to begin downloading bundles.
|
||||
func (d *Downloader) Start(ctx context.Context) {
|
||||
if *d.config.Trigger == plugins.TriggerPeriodic {
|
||||
go d.doStart(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Downloader) doStart(context.Context) {
|
||||
// We'll revisit context passing/usage later.
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
d.wg.Add(1)
|
||||
go d.loop(ctx)
|
||||
|
||||
done := <-d.stop // blocks until there's something to read
|
||||
cancel()
|
||||
d.wg.Wait()
|
||||
d.stopped = true
|
||||
close(done)
|
||||
}
|
||||
|
||||
// Stop tells the Downloader to stop downloading bundles.
|
||||
func (d *Downloader) Stop(context.Context) {
|
||||
if *d.config.Trigger == plugins.TriggerManual {
|
||||
return
|
||||
}
|
||||
|
||||
d.mtx.Lock()
|
||||
defer d.mtx.Unlock()
|
||||
|
||||
if d.stopped {
|
||||
return
|
||||
}
|
||||
|
||||
done := make(chan struct{})
|
||||
d.stop <- done
|
||||
<-done
|
||||
}
|
||||
|
||||
func (d *Downloader) loop(ctx context.Context) {
|
||||
defer d.wg.Done()
|
||||
|
||||
var retry int
|
||||
|
||||
for {
|
||||
|
||||
var delay time.Duration
|
||||
|
||||
err := d.oneShot(ctx)
|
||||
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
delay = util.DefaultBackoff(float64(minRetryDelay), float64(*d.config.Polling.MaxDelaySeconds), retry)
|
||||
} else {
|
||||
if !d.longPollingEnabled || d.config.Polling.LongPollingTimeoutSeconds == nil {
|
||||
// revert the response header timeout value on the http client's transport
|
||||
if *d.client.Config().ResponseHeaderTimeoutSeconds == 0 {
|
||||
d.client = d.client.SetResponseHeaderTimeout(&d.respHdrTimeoutSec)
|
||||
}
|
||||
min := float64(*d.config.Polling.MinDelaySeconds)
|
||||
max := float64(*d.config.Polling.MaxDelaySeconds)
|
||||
delay = time.Duration(((max - min) * rand.Float64()) + min)
|
||||
}
|
||||
}
|
||||
|
||||
d.logger.Debug("Waiting %v before next download/retry.", delay)
|
||||
|
||||
timer, timerCancel := util.TimerWithCancel(delay)
|
||||
select {
|
||||
case <-timer.C:
|
||||
if err != nil {
|
||||
retry++
|
||||
} else {
|
||||
retry = 0
|
||||
}
|
||||
case <-ctx.Done():
|
||||
timerCancel() // explicitly cancel the timer.
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Downloader) oneShot(ctx context.Context) error {
|
||||
m := metrics.New()
|
||||
resp, err := d.download(ctx, m)
|
||||
|
||||
if err != nil {
|
||||
d.etag = ""
|
||||
|
||||
if d.f != nil {
|
||||
d.f(ctx, Update{ETag: "", Bundle: nil, Error: err, Metrics: m, Raw: nil})
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
d.etag = resp.etag
|
||||
d.longPollingEnabled = resp.longPoll
|
||||
|
||||
if d.f != nil {
|
||||
d.f(ctx, Update{ETag: resp.etag, Bundle: resp.b, Error: nil, Metrics: m, Raw: resp.raw, Size: resp.size})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Downloader) download(ctx context.Context, m metrics.Metrics) (*downloaderResponse, error) {
|
||||
d.logger.Debug("Download starting.")
|
||||
|
||||
d.client = d.client.WithHeader("If-None-Match", d.etag)
|
||||
|
||||
preferences := []string{fmt.Sprintf("modes=%v,%v", defaultBundleMode, deltaBundleMode)}
|
||||
|
||||
if d.longPollingEnabled && d.config.Polling.LongPollingTimeoutSeconds != nil {
|
||||
wait := fmt.Sprintf("wait=%s", strconv.FormatInt(*d.config.Polling.LongPollingTimeoutSeconds, 10))
|
||||
preferences = append(preferences, wait)
|
||||
|
||||
// fetch existing response header timeout value on the http client's transport and
|
||||
// clear it for the long poll request
|
||||
current := d.client.Config().ResponseHeaderTimeoutSeconds
|
||||
if *current != 0 {
|
||||
d.respHdrTimeoutSec = *current
|
||||
t := int64(0)
|
||||
d.client = d.client.SetResponseHeaderTimeout(&t)
|
||||
}
|
||||
}
|
||||
|
||||
preferValue := fmt.Sprintf("%v", strings.Join(preferences, ";"))
|
||||
d.client = d.client.WithHeader("Prefer", preferValue)
|
||||
|
||||
m.Timer(metrics.BundleRequest).Start()
|
||||
resp, err := d.client.Do(ctx, "GET", d.path)
|
||||
m.Timer(metrics.BundleRequest).Stop()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("request failed: %w", err)
|
||||
}
|
||||
|
||||
defer util.Close(resp)
|
||||
|
||||
switch resp.StatusCode {
|
||||
case http.StatusOK:
|
||||
var buf bytes.Buffer
|
||||
if resp.Body != nil {
|
||||
d.logger.Debug("Download in progress.")
|
||||
m.Timer(metrics.RegoLoadBundles).Start()
|
||||
defer m.Timer(metrics.RegoLoadBundles).Stop()
|
||||
baseURL := path.Join(d.client.Config().URL, d.path)
|
||||
|
||||
cnt := &count{}
|
||||
r := io.TeeReader(resp.Body, cnt)
|
||||
|
||||
var loader bundle.DirectoryLoader
|
||||
if d.persist {
|
||||
tee := io.TeeReader(r, &buf)
|
||||
loader = bundle.NewTarballLoaderWithBaseURL(tee, baseURL)
|
||||
} else {
|
||||
loader = bundle.NewTarballLoaderWithBaseURL(r, baseURL)
|
||||
}
|
||||
|
||||
// Setting the size limit on the loader allows early exit in the case
|
||||
// of any file exceeding the limit, without the file getting loaded
|
||||
if d.sizeLimitBytes != nil {
|
||||
loader = loader.WithSizeLimitBytes(*d.sizeLimitBytes)
|
||||
}
|
||||
|
||||
etag := resp.Header.Get("ETag")
|
||||
|
||||
reader := bundle.NewCustomReader(loader).
|
||||
WithRegoVersion(d.bundleParserOpts.RegoVersion).
|
||||
WithMetrics(m).
|
||||
WithBundleVerificationConfig(d.bvc).
|
||||
WithBundleEtag(etag).
|
||||
WithLazyLoadingMode(d.lazyLoadingMode).
|
||||
WithBundleName(d.bundleName).
|
||||
WithBundlePersistence(d.persist)
|
||||
|
||||
if d.sizeLimitBytes != nil {
|
||||
reader = reader.WithSizeLimitBytes(*d.sizeLimitBytes)
|
||||
}
|
||||
|
||||
if d.logger.GetLevel() >= logging.Debug {
|
||||
expectedBundleContentType := []string{
|
||||
"application/gzip",
|
||||
"application/octet-stream",
|
||||
"application/vnd.openpolicyagent.bundles",
|
||||
}
|
||||
|
||||
contentType := resp.Header.Get("content-type")
|
||||
if !contains(contentType, expectedBundleContentType) {
|
||||
d.logger.Debug("Content-Type response header set to %v. Expected one of %v. "+
|
||||
"Possibly not a bundle being downloaded.",
|
||||
contentType,
|
||||
expectedBundleContentType,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
b, err := reader.Read()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &downloaderResponse{
|
||||
b: &b,
|
||||
raw: &buf,
|
||||
etag: etag,
|
||||
longPoll: isLongPollSupported(resp.Header),
|
||||
size: cnt.Bytes(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
d.logger.Debug("Server replied with empty body.")
|
||||
return &downloaderResponse{
|
||||
b: nil,
|
||||
raw: nil,
|
||||
etag: "",
|
||||
longPoll: isLongPollSupported(resp.Header),
|
||||
}, nil
|
||||
case http.StatusNotModified:
|
||||
etag := resp.Header.Get("ETag")
|
||||
if etag == "" {
|
||||
etag = d.etag
|
||||
}
|
||||
return &downloaderResponse{
|
||||
b: nil,
|
||||
raw: nil,
|
||||
etag: etag,
|
||||
longPoll: d.longPollingEnabled,
|
||||
}, nil
|
||||
default:
|
||||
if d.logger.GetLevel() == logging.Debug && resp.Body != nil {
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err == nil {
|
||||
d.logger.Debug("bundle download error response with response body: %s", body)
|
||||
}
|
||||
}
|
||||
|
||||
return nil, HTTPError{StatusCode: resp.StatusCode}
|
||||
}
|
||||
}
|
||||
|
||||
type count struct {
|
||||
total int
|
||||
}
|
||||
|
||||
func (c *count) Write(p []byte) (n int, err error) {
|
||||
n = len(p)
|
||||
c.total += n
|
||||
return
|
||||
}
|
||||
|
||||
func (c *count) Bytes() int {
|
||||
return c.total
|
||||
}
|
||||
|
||||
func isLongPollSupported(header http.Header) bool {
|
||||
return header.Get("Content-Type") == "application/vnd.openpolicyagent.bundles"
|
||||
}
|
||||
|
||||
type HTTPError struct {
|
||||
StatusCode int
|
||||
}
|
||||
|
||||
func (e HTTPError) Error() string {
|
||||
return fmt.Sprintf("server replied with %s", http.StatusText(e.StatusCode))
|
||||
}
|
||||
|
||||
func contains(s string, strings []string) bool {
|
||||
for _, str := range strings {
|
||||
if s == str {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,460 +0,0 @@
|
||||
//go:build !opa_no_oci
|
||||
|
||||
package download
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/rand"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/containerd/containerd/remotes"
|
||||
"github.com/containerd/containerd/remotes/docker"
|
||||
"github.com/containerd/errdefs"
|
||||
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
oraslib "oras.land/oras-go/v2"
|
||||
"oras.land/oras-go/v2/content/oci"
|
||||
|
||||
"github.com/open-policy-agent/opa/ast"
|
||||
"github.com/open-policy-agent/opa/bundle"
|
||||
"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/rest"
|
||||
"github.com/open-policy-agent/opa/util"
|
||||
)
|
||||
|
||||
// NewOCI returns a new Downloader that can be started.
|
||||
func NewOCI(config Config, client rest.Client, path, storePath string) *OCIDownloader {
|
||||
localstore, err := oci.New(storePath)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return &OCIDownloader{
|
||||
config: config,
|
||||
path: path,
|
||||
localStorePath: storePath,
|
||||
client: client,
|
||||
trigger: make(chan chan struct{}),
|
||||
stop: make(chan chan struct{}),
|
||||
logger: client.Logger(),
|
||||
store: localstore,
|
||||
}
|
||||
}
|
||||
|
||||
// WithCallback registers a function f to be called when download updates occur.
|
||||
func (d *OCIDownloader) WithCallback(f func(context.Context, Update)) *OCIDownloader {
|
||||
d.f = f
|
||||
return d
|
||||
}
|
||||
|
||||
// WithLogAttrs sets an optional set of key/value pair attributes to include in
|
||||
// log messages emitted by the downloader.
|
||||
func (d *OCIDownloader) WithLogAttrs(attrs map[string]interface{}) *OCIDownloader {
|
||||
d.logger = d.logger.WithFields(attrs)
|
||||
return d
|
||||
}
|
||||
|
||||
// WithBundleVerificationConfig sets the key configuration used to verify a signed bundle
|
||||
func (d *OCIDownloader) WithBundleVerificationConfig(config *bundle.VerificationConfig) *OCIDownloader {
|
||||
d.bvc = config
|
||||
return d
|
||||
}
|
||||
|
||||
// WithSizeLimitBytes sets the file size limit for bundles read by this downloader.
|
||||
func (d *OCIDownloader) WithSizeLimitBytes(n int64) *OCIDownloader {
|
||||
d.sizeLimitBytes = &n
|
||||
return d
|
||||
}
|
||||
|
||||
// WithBundlePersistence specifies if the downloaded bundle will eventually be persisted to disk.
|
||||
func (d *OCIDownloader) WithBundlePersistence(persist bool) *OCIDownloader {
|
||||
d.persist = persist
|
||||
return d
|
||||
}
|
||||
|
||||
// WithBundleParserOpts specifies the parser options to use when parsing downloaded bundles.
|
||||
func (d *OCIDownloader) WithBundleParserOpts(opts ast.ParserOptions) *OCIDownloader {
|
||||
d.bundleParserOpts = opts
|
||||
return d
|
||||
}
|
||||
|
||||
// ClearCache is deprecated. Use SetCache instead.
|
||||
func (d *OCIDownloader) ClearCache() {
|
||||
}
|
||||
|
||||
// SetCache sets the etag value to the SHA of the loaded bundle
|
||||
func (d *OCIDownloader) SetCache(etag string) {
|
||||
d.etag = etag
|
||||
}
|
||||
|
||||
// Trigger can be used to control when the downloader attempts to download
|
||||
// a new bundle in manual triggering mode.
|
||||
func (d *OCIDownloader) Trigger(ctx context.Context) error {
|
||||
done := make(chan error)
|
||||
|
||||
go func() {
|
||||
err := d.oneShot(ctx)
|
||||
if err != nil {
|
||||
d.logger.Error("OCI - Bundle download failed: %v.", err)
|
||||
if ctx.Err() == nil {
|
||||
done <- err
|
||||
}
|
||||
}
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// Start tells the Downloader to begin downloading bundles.
|
||||
func (d *OCIDownloader) Start(ctx context.Context) {
|
||||
if *d.config.Trigger == plugins.TriggerPeriodic {
|
||||
go d.doStart(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
// Stop tells the Downloader to stop downloading bundles.
|
||||
func (d *OCIDownloader) Stop(context.Context) {
|
||||
if *d.config.Trigger == plugins.TriggerManual {
|
||||
return
|
||||
}
|
||||
|
||||
d.mtx.Lock()
|
||||
defer d.mtx.Unlock()
|
||||
|
||||
if d.stopped {
|
||||
return
|
||||
}
|
||||
|
||||
done := make(chan struct{})
|
||||
d.stop <- done
|
||||
<-done
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) doStart(context.Context) {
|
||||
// We'll revisit context passing/usage later.
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
d.wg.Add(1)
|
||||
go d.loop(ctx)
|
||||
|
||||
done := <-d.stop // blocks until there's something to read
|
||||
cancel()
|
||||
d.wg.Wait()
|
||||
d.stopped = true
|
||||
close(done)
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) loop(ctx context.Context) {
|
||||
defer d.wg.Done()
|
||||
|
||||
var retry int
|
||||
|
||||
for {
|
||||
|
||||
var delay time.Duration
|
||||
|
||||
err := d.oneShot(ctx)
|
||||
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
delay = util.DefaultBackoff(float64(minRetryDelay), float64(*d.config.Polling.MaxDelaySeconds), retry)
|
||||
} else {
|
||||
// revert the response header timeout value on the http client's transport
|
||||
min := float64(*d.config.Polling.MinDelaySeconds)
|
||||
max := float64(*d.config.Polling.MaxDelaySeconds)
|
||||
delay = time.Duration(((max - min) * rand.Float64()) + min)
|
||||
}
|
||||
|
||||
d.logger.Debug("OCI - Waiting %v before next download/retry.", delay)
|
||||
|
||||
timer, timerCancel := util.TimerWithCancel(delay)
|
||||
select {
|
||||
case <-timer.C:
|
||||
if err != nil {
|
||||
retry++
|
||||
} else {
|
||||
retry = 0
|
||||
}
|
||||
case <-ctx.Done():
|
||||
timerCancel() // explicitly cancel the timer.
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) oneShot(ctx context.Context) error {
|
||||
m := metrics.New()
|
||||
resp, err := d.download(ctx, m)
|
||||
if err != nil {
|
||||
if d.f != nil {
|
||||
d.f(ctx, Update{ETag: "", Bundle: nil, Error: err, Metrics: m, Raw: nil})
|
||||
}
|
||||
return err
|
||||
}
|
||||
d.SetCache(resp.etag) // set the current etag sha to the cache
|
||||
|
||||
if d.f != nil {
|
||||
d.f(ctx, Update{ETag: resp.etag, Bundle: resp.b, Error: nil, Metrics: m, Raw: resp.raw, Size: resp.size})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) download(ctx context.Context, m metrics.Metrics) (*downloaderResponse, error) {
|
||||
d.logger.Debug("OCI - Download starting.")
|
||||
var buf bytes.Buffer
|
||||
|
||||
preferences := []string{fmt.Sprintf("modes=%v,%v", defaultBundleMode, deltaBundleMode)}
|
||||
|
||||
preferValue := fmt.Sprintf("%v", strings.Join(preferences, ";"))
|
||||
d.client = d.client.WithHeader("Prefer", preferValue)
|
||||
|
||||
m.Timer(metrics.BundleRequest).Start()
|
||||
desc, err := d.pull(ctx, d.path)
|
||||
if err != nil {
|
||||
return &downloaderResponse{}, fmt.Errorf("failed to pull %s: %w", d.path, err)
|
||||
}
|
||||
|
||||
manifest, err := manifestFromDesc(ctx, d.store, desc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tarballDescriptor := ocispec.Descriptor{}
|
||||
for _, descriptor := range manifest.Layers {
|
||||
if descriptor.MediaType == "application/vnd.oci.image.layer.v1.tar+gzip" {
|
||||
tarballDescriptor = descriptor
|
||||
break
|
||||
}
|
||||
}
|
||||
if tarballDescriptor.MediaType == "" {
|
||||
return nil, fmt.Errorf("no tarball descriptor found in the layers")
|
||||
}
|
||||
etag := tarballDescriptor.Digest.Hex()
|
||||
bundleFilePath := filepath.Join(d.localStorePath, "blobs", "sha256", etag)
|
||||
// if the downloader etag sha is the same with digest of the tarball it was already loaded
|
||||
if d.etag == etag {
|
||||
return &downloaderResponse{
|
||||
b: nil,
|
||||
raw: nil,
|
||||
etag: etag,
|
||||
longPoll: false,
|
||||
}, nil
|
||||
}
|
||||
fileReader, err := os.Open(bundleFilePath)
|
||||
|
||||
cnt := &count{}
|
||||
r := io.TeeReader(fileReader, cnt)
|
||||
tee := io.TeeReader(r, &buf)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
loader := bundle.NewTarballLoaderWithBaseURL(tee, d.localStorePath)
|
||||
reader := bundle.NewCustomReader(loader).
|
||||
WithMetrics(m).
|
||||
WithBundleVerificationConfig(d.bvc).
|
||||
WithBundleEtag(etag).
|
||||
WithRegoVersion(d.bundleParserOpts.RegoVersion)
|
||||
bundleInfo, err := reader.Read()
|
||||
if err != nil {
|
||||
return &downloaderResponse{}, fmt.Errorf("unexpected error %w", err)
|
||||
}
|
||||
|
||||
m.Timer(metrics.BundleRequest).Stop()
|
||||
|
||||
return &downloaderResponse{
|
||||
b: &bundleInfo,
|
||||
raw: &buf,
|
||||
etag: etag,
|
||||
longPoll: false,
|
||||
size: cnt.Bytes(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) pull(ctx context.Context, ref string) (*ocispec.Descriptor, error) {
|
||||
lookup := d.client.AuthPluginLookup()
|
||||
|
||||
plugin, err := d.client.Config().AuthPlugin(lookup)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to look up auth plugin: %w", err)
|
||||
}
|
||||
|
||||
d.logger.Debug("OCIDownloader: using auth plugin: %T", plugin)
|
||||
|
||||
resolver, err := dockerResolver(plugin, d.client.Config(), d.logger)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid host url %s: %w", d.client.Config().URL, err)
|
||||
}
|
||||
|
||||
target := remoteManager{
|
||||
resolver: resolver,
|
||||
srcRef: ref,
|
||||
}
|
||||
|
||||
manifestDescriptor, err := oraslib.Copy(ctx, &target, ref, d.store, "", oraslib.DefaultCopyOptions)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("download for '%s' failed: %w", ref, err)
|
||||
}
|
||||
|
||||
return &manifestDescriptor, nil
|
||||
}
|
||||
|
||||
func dockerResolver(plugin rest.HTTPAuthPlugin, config *rest.Config, logger logging.Logger) (remotes.Resolver, error) {
|
||||
client, err := plugin.NewClient(*config)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create auth client: %w", err)
|
||||
}
|
||||
|
||||
urlInfo, err := url.Parse(config.URL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse url: %w", err)
|
||||
}
|
||||
|
||||
authorizer := pluginAuthorizer{
|
||||
plugin: plugin,
|
||||
client: client,
|
||||
logger: logger,
|
||||
}
|
||||
|
||||
registryHost := docker.RegistryHost{
|
||||
Host: urlInfo.Host,
|
||||
Scheme: urlInfo.Scheme,
|
||||
Capabilities: docker.HostCapabilityPull | docker.HostCapabilityResolve | docker.HostCapabilityPush,
|
||||
Client: client,
|
||||
Path: "/v2",
|
||||
Authorizer: &authorizer,
|
||||
}
|
||||
|
||||
opts := docker.ResolverOptions{
|
||||
Hosts: func(string) ([]docker.RegistryHost, error) {
|
||||
return []docker.RegistryHost{registryHost}, nil
|
||||
},
|
||||
}
|
||||
|
||||
return docker.NewResolver(opts), nil
|
||||
}
|
||||
|
||||
type pluginAuthorizer struct {
|
||||
plugin rest.HTTPAuthPlugin
|
||||
client *http.Client
|
||||
|
||||
// authorizer will be populated by the first call to pluginAuthorizer.Prepare
|
||||
// since it requires a first pass through the plugin.Prepare method.
|
||||
authorizer docker.Authorizer
|
||||
|
||||
logger logging.Logger
|
||||
}
|
||||
|
||||
var _ docker.Authorizer = &pluginAuthorizer{}
|
||||
|
||||
func (a *pluginAuthorizer) AddResponses(ctx context.Context, responses []*http.Response) error {
|
||||
return a.authorizer.AddResponses(ctx, responses)
|
||||
}
|
||||
|
||||
// Authorize uses a rest.HTTPAuthPlugin to Prepare a request before passing it on
|
||||
// to the docker.Authorizer.
|
||||
func (a *pluginAuthorizer) Authorize(ctx context.Context, req *http.Request) error {
|
||||
if err := a.plugin.Prepare(req); err != nil {
|
||||
err = fmt.Errorf("failed to prepare docker request: %w", err)
|
||||
|
||||
// Make sure to log this before passing the error back to docker
|
||||
a.logger.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
if a.authorizer == nil {
|
||||
// Some registry authentication implementations require a token fetch from
|
||||
// a separate authenticated token server. This flow is described in the
|
||||
// docker token auth spec:
|
||||
// https://docs.docker.com/registry/spec/auth/token/#requesting-a-token
|
||||
//
|
||||
// Unfortunately, the containerd implementation does not use the Prepare
|
||||
// mechanism to authenticate these token requests and we need to add
|
||||
// auth information in form of a static docker.WithAuthHeader.
|
||||
//
|
||||
// Since rest.HTTPAuthPlugins will set the auth header on the request
|
||||
// passed to HTTPAuthPlugin.Prepare, we can use it afterwards to build
|
||||
// our docker.Authorizer.
|
||||
a.authorizer = docker.NewDockerAuthorizer(
|
||||
docker.WithAuthHeader(req.Header),
|
||||
docker.WithAuthClient(a.client),
|
||||
)
|
||||
}
|
||||
|
||||
return a.authorizer.Authorize(ctx, req)
|
||||
}
|
||||
|
||||
func manifestFromDesc(ctx context.Context, target oraslib.Target, desc *ocispec.Descriptor) (*ocispec.Manifest, error) {
|
||||
var manifest ocispec.Manifest
|
||||
|
||||
descReader, err := target.Fetch(ctx, *desc)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("unable to fetch descriptor with digest %q: %w", desc.Digest, err)
|
||||
}
|
||||
defer descReader.Close()
|
||||
|
||||
descBytes, err := io.ReadAll(descReader)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("unable to read bytes from descriptor: %w", err)
|
||||
}
|
||||
|
||||
if err = json.Unmarshal(descBytes, &manifest); err != nil {
|
||||
return nil, fmt.Errorf("unable to unmarshal manifest: %w", err)
|
||||
}
|
||||
|
||||
if len(manifest.Layers) < 1 {
|
||||
return nil, fmt.Errorf("no layers in manifest")
|
||||
}
|
||||
|
||||
return &manifest, nil
|
||||
}
|
||||
|
||||
type remoteManager struct {
|
||||
resolver remotes.Resolver
|
||||
srcRef string
|
||||
}
|
||||
|
||||
func (r *remoteManager) Resolve(ctx context.Context, ref string) (ocispec.Descriptor, error) {
|
||||
_, desc, err := r.resolver.Resolve(ctx, ref)
|
||||
if err != nil {
|
||||
return ocispec.Descriptor{}, err
|
||||
}
|
||||
return desc, nil
|
||||
}
|
||||
|
||||
func (r *remoteManager) Fetch(ctx context.Context, target ocispec.Descriptor) (io.ReadCloser, error) {
|
||||
fetcher, err := r.resolver.Fetcher(ctx, r.srcRef)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return fetcher.Fetch(ctx, target)
|
||||
}
|
||||
|
||||
func (r *remoteManager) Exists(ctx context.Context, target ocispec.Descriptor) (bool, error) {
|
||||
_, err := r.Fetch(ctx, target)
|
||||
if err == nil {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
return !errdefs.IsNotFound(err), err
|
||||
}
|
||||
@@ -1,461 +0,0 @@
|
||||
//go:build slow
|
||||
// +build slow
|
||||
|
||||
package download
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/open-policy-agent/opa/ast"
|
||||
"github.com/open-policy-agent/opa/bundle"
|
||||
|
||||
"github.com/open-policy-agent/opa/keys"
|
||||
"github.com/open-policy-agent/opa/plugins/rest"
|
||||
)
|
||||
|
||||
// when changed the layer hash & size should be updated in .manifest files
|
||||
//go:generate go run github.com/open-policy-agent/opa build -b --signing-alg HS256 testdata/latest_bundle_data --output testdata/latest.tar.gz
|
||||
//go:generate go run github.com/open-policy-agent/opa build -b --signing-alg HS256 --signing-key secret testdata/signed_bundle_data --output testdata/signed.tar.gz
|
||||
//go:generate go run github.com/open-policy-agent/opa build --v1-compatible -b --signing-alg HS256 --signing-key secret testdata/rego_v1_bundle_data --output testdata/rego_v1.tar.gz
|
||||
|
||||
func TestOCIDownloaderWithBundleVerificationConfig(t *testing.T) {
|
||||
vc := bundle.NewVerificationConfig(map[string]*bundle.KeyConfig{"default": {Key: "secret", Algorithm: "HS256"}}, "", "", nil)
|
||||
ctx := context.Background()
|
||||
fixture := newTestFixture(t)
|
||||
fixture.server.expEtag = "sha256:c5834dbce332cabe6ae68a364de171a50bf5b08024c27d7c08cc72878b4df7ff"
|
||||
|
||||
updates := make(chan *Update)
|
||||
|
||||
config := Config{}
|
||||
if err := config.ValidateAndInjectDefaults(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
d := NewOCI(config, fixture.client, "ghcr.io/org/repo:signed", "/tmp/opa/").WithCallback(func(_ context.Context, u Update) {
|
||||
if u.Error != nil {
|
||||
t.Fatalf("expected no error but got: %v", u.Error)
|
||||
}
|
||||
updates <- &u
|
||||
}).WithBundleVerificationConfig(vc)
|
||||
|
||||
d.Start(ctx)
|
||||
|
||||
// Give time for some download events to occur
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
u1 := <-updates
|
||||
|
||||
if u1.Bundle == nil || len(u1.Bundle.Modules) == 0 {
|
||||
t.Fatal("expected bundle with at least one module but got:", u1)
|
||||
}
|
||||
|
||||
if !strings.HasSuffix(u1.Bundle.Modules[0].URL, u1.Bundle.Modules[0].Path) {
|
||||
t.Fatalf("expected URL to have path as suffix but got %v and %v", u1.Bundle.Modules[0].URL, u1.Bundle.Modules[0].Path)
|
||||
}
|
||||
|
||||
d.Stop(ctx)
|
||||
|
||||
}
|
||||
|
||||
func TestOCIDownloaderWithRegoV1Bundle(t *testing.T) {
|
||||
tests := []struct {
|
||||
note string
|
||||
regoVersion ast.RegoVersion
|
||||
expErr string
|
||||
}{
|
||||
// The bundle contains a v1 rego_version attr, so we expect no errors regardless of parser regoVersion.
|
||||
{
|
||||
note: "non-1.0 compatible OCI downloader",
|
||||
},
|
||||
{
|
||||
note: "1.0 compatible OCI downloader",
|
||||
regoVersion: ast.RegoV1,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.note, func(t *testing.T) {
|
||||
vc := bundle.NewVerificationConfig(map[string]*bundle.KeyConfig{"default": {Key: "secret", Algorithm: "HS256"}}, "", "", nil)
|
||||
ctx := context.Background()
|
||||
fixture := newTestFixture(t)
|
||||
fixture.server.expEtag = "sha256:c5834dbce332cabe6ae68a364de171a50bf5b08024c27d7c08cc72878b4df7ff"
|
||||
|
||||
// We might get multiple updates, a buffered channel will make sure we save the first one.
|
||||
updates := make(chan *Update, 1)
|
||||
|
||||
config := Config{}
|
||||
if err := config.ValidateAndInjectDefaults(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
d := NewOCI(config, fixture.client, "ghcr.io/org/repo:rego_v1", "/tmp/opa/").
|
||||
WithBundleParserOpts(ast.ParserOptions{RegoVersion: tc.regoVersion}).
|
||||
WithCallback(func(_ context.Context, u Update) {
|
||||
// We might get multiple updates before the test ends, and we don't want to block indefinitely.
|
||||
select {
|
||||
case updates <- &u:
|
||||
}
|
||||
}).WithBundleVerificationConfig(vc)
|
||||
|
||||
d.Start(ctx)
|
||||
|
||||
// Give time for some download events to occur
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
// We only care about the first update
|
||||
u1 := <-updates
|
||||
|
||||
if tc.expErr != "" {
|
||||
if u1.Error == nil {
|
||||
t.Fatalf("expected error but got: %v", u1)
|
||||
} else {
|
||||
if !strings.Contains(u1.Error.Error(), tc.expErr) {
|
||||
t.Fatalf("expected error:\n\n%v\n\nbut got:\n\n%v", tc.expErr, u1.Error)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if u1.Error != nil {
|
||||
t.Fatalf("expected no error but got: %v", u1.Error)
|
||||
}
|
||||
|
||||
if u1.Bundle == nil || len(u1.Bundle.Modules) == 0 {
|
||||
t.Fatal("expected bundle with at least one module but got:", u1)
|
||||
}
|
||||
|
||||
if !strings.HasSuffix(u1.Bundle.Modules[0].URL, u1.Bundle.Modules[0].Path) {
|
||||
t.Fatalf("expected URL to have path as suffix but got %v and %v", u1.Bundle.Modules[0].URL, u1.Bundle.Modules[0].Path)
|
||||
}
|
||||
}
|
||||
|
||||
d.Stop(ctx)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestOCIStartStop(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newTestFixture(t)
|
||||
fixture.server.expEtag = "sha256:cc09b0f5ac97b11637c96ff1b0fbbc287c5ba0169813edaa71fe58424e95f0b7"
|
||||
|
||||
updates := make(chan *Update)
|
||||
|
||||
config := Config{}
|
||||
if err := config.ValidateAndInjectDefaults(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
d := NewOCI(config, fixture.client, "ghcr.io/org/repo:latest", "/tmp/opa/").WithCallback(func(_ context.Context, u Update) {
|
||||
updates <- &u
|
||||
})
|
||||
|
||||
d.Start(ctx)
|
||||
|
||||
// Give time for some download events to occur
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
u1 := <-updates
|
||||
|
||||
if u1.Bundle == nil || len(u1.Bundle.Modules) == 0 {
|
||||
t.Fatal("expected bundle with at least one module but got:", u1)
|
||||
}
|
||||
|
||||
if !strings.HasSuffix(u1.Bundle.Modules[0].URL, u1.Bundle.Modules[0].Path) {
|
||||
t.Fatalf("expected URL to have path as suffix but got %v and %v", u1.Bundle.Modules[0].URL, u1.Bundle.Modules[0].Path)
|
||||
}
|
||||
|
||||
d.Stop(ctx)
|
||||
}
|
||||
|
||||
func TestOCIBearerAuthPlugin(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newTestFixture(t)
|
||||
plainToken := "secret"
|
||||
token := base64.StdEncoding.EncodeToString([]byte(plainToken)) // token should be base64 encoded
|
||||
fixture.server.expAuth = fmt.Sprintf("Bearer %s", token) // test on private repository
|
||||
fixture.server.expEtag = "sha256:c5834dbce332cabe6ae68a364de171a50bf5b08024c27d7c08cc72878b4df7ff"
|
||||
|
||||
restConf := fmt.Sprintf(`{
|
||||
"url": %q,
|
||||
"type": "oci",
|
||||
"credentials": {
|
||||
"bearer": {
|
||||
"token": %q
|
||||
}
|
||||
}
|
||||
}`, fixture.server.server.URL, plainToken)
|
||||
|
||||
client, err := rest.New([]byte(restConf), map[string]*keys.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
fixture.setClient(client)
|
||||
|
||||
config := Config{}
|
||||
if err := config.ValidateAndInjectDefaults(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
d := NewOCI(config, fixture.client, "ghcr.io/org/repo:latest", "/tmp/oci")
|
||||
|
||||
if err := d.oneShot(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOCIFailureAuthn(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newTestFixture(t)
|
||||
fixture.server.expAuth = "Bearer badsecret"
|
||||
defer fixture.server.stop()
|
||||
|
||||
d := NewOCI(Config{}, fixture.client, "ghcr.io/org/repo:latest", "/tmp/oci")
|
||||
|
||||
err := d.oneShot(ctx)
|
||||
if err == nil {
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "401 Unauthorized") {
|
||||
t.Fatal("expected 401 Unauthorized message")
|
||||
}
|
||||
}
|
||||
|
||||
func TestOCIEtag(t *testing.T) {
|
||||
fixture := newTestFixture(t)
|
||||
token := base64.StdEncoding.EncodeToString([]byte("secret")) // token should be base64 encoded
|
||||
fixture.server.expAuth = fmt.Sprintf("Bearer %s", token) // test on private repository
|
||||
fixture.server.expEtag = "sha256:c5834dbce332cabe6ae68a364de171a50bf5b08024c27d7c08cc72878b4df7ff"
|
||||
|
||||
restConfig := []byte(fmt.Sprintf(`{
|
||||
"url": %q,
|
||||
"type": "oci",
|
||||
"credentials": {
|
||||
"bearer": {
|
||||
"token": "secret"
|
||||
}
|
||||
}
|
||||
}`, fixture.server.server.URL))
|
||||
|
||||
client, err := rest.New(restConfig, map[string]*keys.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
fixture.setClient(client)
|
||||
|
||||
config := Config{}
|
||||
if err := config.ValidateAndInjectDefaults(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
firstResponse := Update{ETag: ""}
|
||||
d := NewOCI(config, fixture.client, "ghcr.io/org/repo:latest", "/tmp/oci").WithCallback(func(_ context.Context, u Update) {
|
||||
if firstResponse.ETag == "" {
|
||||
firstResponse = u
|
||||
return
|
||||
}
|
||||
|
||||
if u.ETag != firstResponse.ETag || u.Bundle != nil {
|
||||
t.Fatal("expected nil bundle and same etag but got:", u)
|
||||
}
|
||||
})
|
||||
|
||||
// fill firstResponse
|
||||
if err := d.oneShot(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Give time for some download events to occur
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
// second call to verify if nil bundle is returned and same etag
|
||||
err = d.oneShot(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestOCIPublicRegistryAuth tests the registry `token` auth
|
||||
// that is implemented by public registries (more details are
|
||||
// in the doc comment of withPublicRegistryAuth).
|
||||
//
|
||||
// Other tests that don't explicitly set an authentication method
|
||||
// implicitly test no authentication - this is different from
|
||||
// the mechanism used by public registries.
|
||||
func TestOCIPublicRegistryAuth(t *testing.T) {
|
||||
fixture := newTestFixture(t, withPublicRegistryAuth())
|
||||
|
||||
restConfig := []byte(fmt.Sprintf(`{
|
||||
"url": %q,
|
||||
"type": "oci"
|
||||
}`, fixture.server.server.URL))
|
||||
|
||||
client, err := rest.New(restConfig, map[string]*keys.Config{})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create rest client: %s", err)
|
||||
}
|
||||
fixture.client = client
|
||||
|
||||
d := NewOCI(Config{}, fixture.client, "ghcr.io/org/repo:latest", t.TempDir())
|
||||
|
||||
if err := d.oneShot(context.Background()); err != nil {
|
||||
t.Fatalf("Unexpected error: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestOCITokenAuth tests the registry `token` auth that is used for some registries (f.e. gitlab).
|
||||
// After the initial fetch the token has to be added to the request that fetches the temporary token.
|
||||
// This test verifies that the token is added to the second token request.
|
||||
func TestOCITokenAuth(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newTestFixture(t, withAuthenticatedTokenAuth())
|
||||
plainToken := "secret"
|
||||
token := base64.StdEncoding.EncodeToString([]byte(plainToken)) // token should be base64 encoded
|
||||
fixture.server.expAuth = fmt.Sprintf("Bearer %s", token) // test on private repository
|
||||
fixture.server.expEtag = "sha256:c5834dbce332cabe6ae68a364de171a50bf5b08024c27d7c08cc72878b4df7ff"
|
||||
|
||||
restConf := fmt.Sprintf(`{
|
||||
"url": %q,
|
||||
"type": "oci",
|
||||
"credentials": {
|
||||
"bearer": {
|
||||
"token": %q
|
||||
}
|
||||
}
|
||||
}`, fixture.server.server.URL, plainToken)
|
||||
|
||||
client, err := rest.New([]byte(restConf), map[string]*keys.Config{})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create rest client: %s", err)
|
||||
}
|
||||
fixture.setClient(client)
|
||||
|
||||
config := Config{}
|
||||
if err := config.ValidateAndInjectDefaults(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
d := NewOCI(Config{}, fixture.client, "ghcr.io/org/repo:latest", t.TempDir())
|
||||
|
||||
if err := d.oneShot(ctx); err != nil {
|
||||
t.Fatalf("Unexpected error: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOCICustomAuthPlugin(t *testing.T) {
|
||||
fixture := newTestFixture(t)
|
||||
defer fixture.server.stop()
|
||||
|
||||
restConfig := []byte(fmt.Sprintf(`{
|
||||
"url": %q,
|
||||
"credentials": {
|
||||
"plugin": "my_plugin"
|
||||
}
|
||||
}`, fixture.server.server.URL))
|
||||
|
||||
client, err := rest.New(
|
||||
restConfig,
|
||||
map[string]*keys.Config{},
|
||||
rest.AuthPluginLookup(mockAuthPluginLookup),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
fixture.setClient(client)
|
||||
|
||||
config := Config{}
|
||||
if err := config.ValidateAndInjectDefaults(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
tmpDir := t.TempDir()
|
||||
|
||||
d := NewOCI(config, fixture.client, "ghcr.io/org/repo:latest", tmpDir)
|
||||
|
||||
if err := d.oneShot(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOCIValidateAndInjectDefaults(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newTestFixture(t)
|
||||
fixture.server.expEtag = "sha256:c5834dbce332cabe6ae68a364de171a50bf5b08024c27d7c08cc72878b4df7ff"
|
||||
|
||||
updates := make(chan *Update)
|
||||
|
||||
config := Config{}
|
||||
if err := config.ValidateAndInjectDefaults(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
d := NewOCI(config, fixture.client, "ghcr.io/org/repo:latest", t.TempDir()).WithCallback(func(_ context.Context, u Update) {
|
||||
updates <- &u
|
||||
}).WithBundlePersistence(true)
|
||||
|
||||
d.Start(ctx)
|
||||
|
||||
// Give time for some download events to occur
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
u1 := <-updates
|
||||
|
||||
if u1.Size == 0 {
|
||||
t.Fatal("expected non-0 size")
|
||||
}
|
||||
|
||||
if u1.Raw == nil {
|
||||
t.Fatal("expected bundle reader to be non-nil")
|
||||
}
|
||||
|
||||
r := bundle.NewReader(u1.Raw)
|
||||
|
||||
b, err := r.Read()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(b.Data, u1.Bundle.Data) {
|
||||
t.Fatal("expected the bundle object and reader to have the same data")
|
||||
}
|
||||
|
||||
if len(b.Modules) != len(u1.Bundle.Modules) {
|
||||
t.Fatal("expected the bundle object and reader to have the same number of bundle modules")
|
||||
}
|
||||
|
||||
d.Stop(ctx)
|
||||
}
|
||||
|
||||
func mockAuthPluginLookup(string) rest.HTTPAuthPlugin {
|
||||
return &mockAuthPlugin{}
|
||||
}
|
||||
|
||||
type mockAuthPlugin struct{}
|
||||
|
||||
func (p *mockAuthPlugin) NewClient(c rest.Config) (*http.Client, error) {
|
||||
tlsConfig, err := rest.DefaultTLSConfig(c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
timeoutSec := 10
|
||||
|
||||
client := rest.DefaultRoundTripperClient(
|
||||
tlsConfig,
|
||||
int64(timeoutSec),
|
||||
)
|
||||
|
||||
return client, nil
|
||||
}
|
||||
|
||||
func (*mockAuthPlugin) Prepare(r *http.Request) error {
|
||||
r.Header.Set("Authorization", "Bearer secret")
|
||||
return nil
|
||||
}
|
||||
@@ -1,59 +0,0 @@
|
||||
//go:build opa_no_oci
|
||||
|
||||
package download
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/open-policy-agent/opa/ast"
|
||||
"github.com/open-policy-agent/opa/bundle"
|
||||
"github.com/open-policy-agent/opa/plugins/rest"
|
||||
)
|
||||
|
||||
func NewOCI(Config, rest.Client, string, string) *OCIDownloader {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) WithCallback(f func(context.Context, Update)) *OCIDownloader {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) WithLogAttrs(map[string]interface{}) *OCIDownloader {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) WithBundleVerificationConfig(*bundle.VerificationConfig) *OCIDownloader {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) WithSizeLimitBytes(int64) *OCIDownloader {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) WithBundlePersistence(bool) *OCIDownloader {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) ClearCache() {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) SetCache(string) {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) Trigger(context.Context) error {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) Start(context.Context) {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (d *OCIDownloader) Stop(context.Context) {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
|
||||
func (*OCIDownloader) WithBundleParserOpts(ast.ParserOptions) *OCIDownloader {
|
||||
panic("built without OCI support")
|
||||
}
|
||||
@@ -1,32 +0,0 @@
|
||||
package download
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/open-policy-agent/opa/ast"
|
||||
"github.com/open-policy-agent/opa/bundle"
|
||||
"github.com/open-policy-agent/opa/logging"
|
||||
"github.com/open-policy-agent/opa/plugins/rest"
|
||||
"oras.land/oras-go/v2/content/oci"
|
||||
)
|
||||
|
||||
type OCIDownloader struct {
|
||||
config Config // downloader configuration for tuning polling and other downloader behaviour
|
||||
client rest.Client // HTTP client to use for bundle downloading
|
||||
path string // path for OCI image as <registry>/<org>/<repo>:<tag>
|
||||
localStorePath string // path for the local OCI storage
|
||||
trigger chan chan struct{} // channel to signal downloads when manual triggering is enabled
|
||||
stop chan chan struct{} // used to signal plugin to stop running
|
||||
f func(context.Context, Update) // callback function invoked when download updates occur
|
||||
sizeLimitBytes *int64 // max bundle file size in bytes (passed to reader)
|
||||
bvc *bundle.VerificationConfig
|
||||
wg sync.WaitGroup
|
||||
logger logging.Logger
|
||||
mtx sync.Mutex
|
||||
stopped bool
|
||||
persist bool
|
||||
store *oci.Store
|
||||
etag string
|
||||
bundleParserOpts ast.ParserOptions
|
||||
}
|
||||
Vendored
-1
@@ -1 +0,0 @@
|
||||
{}
|
||||
Vendored
-19
@@ -1,19 +0,0 @@
|
||||
{
|
||||
"schemaVersion":2,
|
||||
"config":{
|
||||
"mediaType":"application/vnd.oci.image.config.v1+json",
|
||||
"digest":"sha256:44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a",
|
||||
"size":2
|
||||
},
|
||||
"layers":[
|
||||
{
|
||||
"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip",
|
||||
"digest":"sha256:d85a3b7072e295a091f4ec50e85fefcd5285a1e2c60c298c0b87c498f1cb0613",
|
||||
"size":610,
|
||||
"annotations":{
|
||||
"org.opencontainers.image.created":"2022-02-11T09:00:07Z",
|
||||
"org.opencontainers.image.title":"dani/testpol"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
Vendored
BIN
Binary file not shown.
@@ -1 +0,0 @@
|
||||
{"revision":"","roots":["peoplefinder"]}
|
||||
@@ -1 +0,0 @@
|
||||
{}
|
||||
@@ -1,23 +0,0 @@
|
||||
package peoplefinder.DELETE.api.users.__id
|
||||
|
||||
import rego.v1
|
||||
import input.user.attributes.properties as user_props
|
||||
|
||||
default allowed = false
|
||||
|
||||
default visible = false
|
||||
|
||||
default enabled = false
|
||||
|
||||
allowed if {
|
||||
user_props.department == "Operations"
|
||||
user_props.title == "IT Manager"
|
||||
}
|
||||
|
||||
visible if {
|
||||
user_props.department == "Operations"
|
||||
}
|
||||
|
||||
enabled if {
|
||||
allowed
|
||||
}
|
||||
@@ -1,7 +0,0 @@
|
||||
package peoplefinder.GET.api.users.__id
|
||||
|
||||
default allowed = true
|
||||
|
||||
default visible = true
|
||||
|
||||
default enabled = true
|
||||
@@ -1,18 +0,0 @@
|
||||
package peoplefinder.POST.api.users.__id
|
||||
|
||||
import rego.v1
|
||||
import input.user.attributes.properties as user_props
|
||||
|
||||
default allowed = false
|
||||
|
||||
default visible = true
|
||||
|
||||
default enabled = false
|
||||
|
||||
allowed if {
|
||||
user_props.department == "Operations"
|
||||
}
|
||||
|
||||
enabled if {
|
||||
allowed
|
||||
}
|
||||
@@ -1,18 +0,0 @@
|
||||
package peoplefinder.PUT.api.users.__id
|
||||
|
||||
import rego.v1
|
||||
import input.user.attributes.properties as user_props
|
||||
|
||||
default allowed = false
|
||||
|
||||
default visible = true
|
||||
|
||||
default enabled = true
|
||||
|
||||
allowed if {
|
||||
user_props.department == "Operations"
|
||||
}
|
||||
|
||||
allowed if {
|
||||
input.user.id == input.resource.id
|
||||
}
|
||||
@@ -1,7 +0,0 @@
|
||||
package peoplefinder.GET.api.users
|
||||
|
||||
default allowed = true
|
||||
|
||||
default visible = true
|
||||
|
||||
default enabled = true
|
||||
@@ -1,23 +0,0 @@
|
||||
package peoplefinder.POST.api.users
|
||||
|
||||
import rego.v1
|
||||
import input.user.attributes.properties as user_props
|
||||
|
||||
default allowed = false
|
||||
|
||||
default visible = false
|
||||
|
||||
default enabled = false
|
||||
|
||||
allowed if {
|
||||
user_props.department == "Operations"
|
||||
user_props.title == "IT Manager"
|
||||
}
|
||||
|
||||
visible if {
|
||||
allowed
|
||||
}
|
||||
|
||||
enabled if {
|
||||
allowed
|
||||
}
|
||||
Vendored
-19
@@ -1,19 +0,0 @@
|
||||
{
|
||||
"schemaVersion":2,
|
||||
"config":{
|
||||
"mediaType":"application/vnd.oci.image.config.v1+json",
|
||||
"digest":"sha256:44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a",
|
||||
"size":2
|
||||
},
|
||||
"layers":[
|
||||
{
|
||||
"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip",
|
||||
"digest":"sha256:0f93a2c5964d7c8b676e3b507b6bc3b771c086428dece6f84256b9948cb3f256",
|
||||
"size":830,
|
||||
"annotations":{
|
||||
"org.opencontainers.image.created":"2022-02-11T09:00:07Z",
|
||||
"org.opencontainers.image.title":"dani/testpol"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
Vendored
BIN
Binary file not shown.
@@ -1 +0,0 @@
|
||||
[1,2,3]
|
||||
@@ -1,9 +0,0 @@
|
||||
package example
|
||||
|
||||
violations contains msg if {
|
||||
msg := "hello"
|
||||
}
|
||||
|
||||
allow if {
|
||||
count(violations) == 0
|
||||
}
|
||||
Vendored
-19
@@ -1,19 +0,0 @@
|
||||
{
|
||||
"schemaVersion":2,
|
||||
"config":{
|
||||
"mediaType":"application/vnd.oci.image.config.v1+json",
|
||||
"digest":"sha256:44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a",
|
||||
"size":2
|
||||
},
|
||||
"layers":[
|
||||
{
|
||||
"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip",
|
||||
"digest":"sha256:7fccf82798e6e627afd04144889570d966583788473db2888f0d0d325904d273",
|
||||
"size":764,
|
||||
"annotations":{
|
||||
"org.opencontainers.image.created":"2022-02-11T09:00:07Z",
|
||||
"org.opencontainers.image.title":"dani/testpol"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
Vendored
BIN
Binary file not shown.
@@ -1 +0,0 @@
|
||||
[1,2,3]
|
||||
@@ -1 +0,0 @@
|
||||
package example
|
||||
@@ -1,557 +0,0 @@
|
||||
//go:build slow
|
||||
// +build slow
|
||||
|
||||
package download
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/open-policy-agent/opa/bundle"
|
||||
"github.com/open-policy-agent/opa/keys"
|
||||
"github.com/open-policy-agent/opa/plugins/rest"
|
||||
)
|
||||
|
||||
var errUnauthorized = errors.New("401 Unauthorized")
|
||||
|
||||
type testFixture struct {
|
||||
d *Downloader
|
||||
client rest.Client
|
||||
server *testServer
|
||||
updates []Update
|
||||
mockBundleActivationError bool
|
||||
etags map[string]string
|
||||
}
|
||||
|
||||
func newTestFixture(t *testing.T, opts ...fixtureOpt) testFixture {
|
||||
t.Helper()
|
||||
|
||||
ts := newTestServer(t)
|
||||
ts.start()
|
||||
|
||||
restConfig := []byte(fmt.Sprintf(`{"url": %q}`, ts.server.URL))
|
||||
|
||||
client, err := rest.New(restConfig, map[string]*keys.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
fixture := testFixture{
|
||||
server: ts,
|
||||
client: client,
|
||||
etags: make(map[string]string),
|
||||
}
|
||||
|
||||
for i, opt := range opts {
|
||||
if err := opt(&fixture); err != nil {
|
||||
t.Fatalf("Failed applying option #%d: %s", i, err)
|
||||
}
|
||||
}
|
||||
|
||||
return fixture
|
||||
}
|
||||
|
||||
type fixtureOpt func(*testFixture) error
|
||||
|
||||
// withPublicRegistryAuth sets up a token auth flow according to
|
||||
// the spec https://docs.docker.com/registry/spec/auth/token/.
|
||||
//
|
||||
// This authentication method is implemented by public
|
||||
// repositories of Github Container Registry, Docker Hub and
|
||||
// AWS ECR (and likely others) and corresponds with the auth
|
||||
// method `token` of the github.com/distribution/distribution
|
||||
// registry project.
|
||||
// See https://docs.docker.com/registry/configuration/#token.
|
||||
//
|
||||
// The token issuing and validation differs between providers
|
||||
// and we only use a minimal version for testing.
|
||||
func withPublicRegistryAuth() fixtureOpt {
|
||||
const token = "some-test-token"
|
||||
tokenServer := httptest.NewServer(tokenHandler(token))
|
||||
|
||||
const wwwAuthenticateFmt = "Bearer realm=%q service=%q scope=%q"
|
||||
tokenServiceURL := tokenServer.URL + "/token"
|
||||
wwwAuthenticate := fmt.Sprintf(wwwAuthenticateFmt,
|
||||
tokenServiceURL,
|
||||
"testRegistry.io",
|
||||
"[pull]")
|
||||
|
||||
return func(tf *testFixture) error {
|
||||
tf.server.customAuth = func(w http.ResponseWriter, r *http.Request) error {
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
if authHeader == "" {
|
||||
w.Header().Set("WWW-Authenticate", wwwAuthenticate)
|
||||
return fmt.Errorf("no authorization header: %w", errUnauthorized)
|
||||
}
|
||||
|
||||
if !strings.HasPrefix(authHeader, "Bearer ") {
|
||||
w.Header().Set("WWW-Authenticate", wwwAuthenticate)
|
||||
return fmt.Errorf("expects bearer scheme: %w", errUnauthorized)
|
||||
}
|
||||
|
||||
bearerToken := strings.TrimPrefix(authHeader, "Bearer ")
|
||||
if bearerToken != token {
|
||||
w.Header().Set("WWW-Authenticate", wwwAuthenticate)
|
||||
return fmt.Errorf("token %q doesn't match %q: %w", bearerToken, token, errUnauthorized)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// withAuthenticatedTokenAuth sets up a token auth flow according to
|
||||
// the spec https://docs.docker.com/registry/spec/auth/token/.
|
||||
//
|
||||
// The flow is the same as for public registries but additionally
|
||||
// the request for fetching the token also has to be authenticated.
|
||||
// Used for example with gitlab registries.
|
||||
//
|
||||
// The token issuing and validation differs between providers,
|
||||
// and we only use a minimal version for testing.
|
||||
func withAuthenticatedTokenAuth() fixtureOpt {
|
||||
const token = "some-test-token"
|
||||
tokenServer := httptest.NewServer(tokenHandlerAuth("c2VjcmV0", token))
|
||||
|
||||
const wwwAuthenticateFmt = "Bearer realm=%q service=%q scope=%q"
|
||||
tokenServiceURL := tokenServer.URL + "/token"
|
||||
wwwAuthenticate := fmt.Sprintf(wwwAuthenticateFmt,
|
||||
tokenServiceURL,
|
||||
"testRegistry.io",
|
||||
"[pull]")
|
||||
|
||||
return func(tf *testFixture) error {
|
||||
tf.server.customAuth = func(w http.ResponseWriter, r *http.Request) error {
|
||||
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
|
||||
if authHeader == "" {
|
||||
w.Header().Set("WWW-Authenticate", wwwAuthenticate)
|
||||
return fmt.Errorf("no authorization header: %w", errUnauthorized)
|
||||
}
|
||||
|
||||
if !strings.HasPrefix(authHeader, "Bearer ") {
|
||||
w.Header().Set("WWW-Authenticate", wwwAuthenticate)
|
||||
return fmt.Errorf("expects bearer scheme: %w", errUnauthorized)
|
||||
}
|
||||
|
||||
bearerToken := strings.TrimPrefix(authHeader, "Bearer ")
|
||||
if bearerToken != token {
|
||||
w.Header().Set("WWW-Authenticate", wwwAuthenticate)
|
||||
return fmt.Errorf("token %q doesn't match %q: %w", bearerToken, token, errUnauthorized)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// tokenHandler returns an http.Handler that responds with the
|
||||
// specified token to GET /token requests.
|
||||
func tokenHandler(issuedToken string) http.HandlerFunc {
|
||||
return tokenHandlerAuth("", issuedToken)
|
||||
}
|
||||
|
||||
// tokenHandlerAuth returns an http.Handler that responds with the
|
||||
// specified token to GET /token requests.
|
||||
//
|
||||
// If expectedToken is not empty, the handler will check that the
|
||||
// Authorization header matches the expected token.
|
||||
func tokenHandlerAuth(expectedToken, issuedToken string) http.HandlerFunc {
|
||||
tokenResponse := struct {
|
||||
Token string `json:"token"`
|
||||
}{
|
||||
Token: issuedToken,
|
||||
}
|
||||
|
||||
responseBody, err := json.Marshal(tokenResponse)
|
||||
if err != nil {
|
||||
panic("failed to marshal token response: " + err.Error())
|
||||
}
|
||||
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != "GET" {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
if r.URL.Path != "/token" {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// If no expected token is set, we don't check the Authorization header.
|
||||
if expectedToken == "" {
|
||||
_, _ = w.Write(responseBody)
|
||||
return
|
||||
}
|
||||
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
if authHeader == "" {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
if !strings.HasPrefix(authHeader, "Bearer ") {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
bearerToken := strings.TrimPrefix(authHeader, "Bearer ")
|
||||
if bearerToken != expectedToken {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
_, _ = w.Write(responseBody)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *testFixture) setClient(client rest.Client) {
|
||||
t.client = client
|
||||
}
|
||||
|
||||
func (t *testFixture) oneShot(ctx context.Context, u Update) {
|
||||
|
||||
t.updates = append(t.updates, u)
|
||||
|
||||
if u.Error != nil {
|
||||
etag := t.etags["test/bundle1"]
|
||||
t.d.SetCache(etag)
|
||||
return
|
||||
}
|
||||
|
||||
if u.Bundle != nil {
|
||||
if t.mockBundleActivationError {
|
||||
etag := t.etags["test/bundle1"]
|
||||
t.d.SetCache(etag)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
t.etags["test/bundle1"] = u.ETag
|
||||
}
|
||||
|
||||
type fileInfo struct {
|
||||
name string
|
||||
length int64
|
||||
}
|
||||
|
||||
type testServer struct {
|
||||
t *testing.T
|
||||
customAuth func(http.ResponseWriter, *http.Request) error
|
||||
expCode int
|
||||
expResp string
|
||||
expEtag string
|
||||
expAuth string
|
||||
bundles map[string]bundle.Bundle
|
||||
server *httptest.Server
|
||||
etagInResponse bool
|
||||
longPoll bool
|
||||
testdataHashes map[string]fileInfo
|
||||
}
|
||||
|
||||
func newTestServer(t *testing.T) *testServer {
|
||||
return &testServer{
|
||||
t: t,
|
||||
bundles: map[string]bundle.Bundle{
|
||||
"test/bundle1": {
|
||||
Manifest: bundle.Manifest{
|
||||
Revision: "quickbrownfaux",
|
||||
},
|
||||
Data: map[string]interface{}{
|
||||
"foo": map[string]interface{}{
|
||||
"bar": json.Number("1"),
|
||||
"baz": "qux",
|
||||
},
|
||||
},
|
||||
Modules: []bundle.ModuleFile{
|
||||
{
|
||||
Path: `/example.rego`,
|
||||
Raw: []byte("package foo\n\ncorge=1"),
|
||||
},
|
||||
},
|
||||
},
|
||||
"test/bundle2": {
|
||||
Manifest: bundle.Manifest{
|
||||
Revision: deltaBundleMode,
|
||||
},
|
||||
Patch: bundle.Patch{Data: []bundle.PatchOperation{
|
||||
{
|
||||
Op: "upsert",
|
||||
Path: "/a/c/d",
|
||||
Value: []string{"foo", "bar"},
|
||||
},
|
||||
}},
|
||||
},
|
||||
"test/v1compat/no_imports": {
|
||||
Manifest: bundle.Manifest{
|
||||
Revision: "quickbrownfaux",
|
||||
},
|
||||
Data: map[string]interface{}{},
|
||||
Modules: []bundle.ModuleFile{
|
||||
{
|
||||
Path: `/example.rego`,
|
||||
Raw: []byte(`package test
|
||||
import data.foo
|
||||
import data.bar as foo
|
||||
p contains 1 if {
|
||||
input.x == 2
|
||||
}`),
|
||||
},
|
||||
},
|
||||
},
|
||||
"test/v1compat/rego_import": {
|
||||
Manifest: bundle.Manifest{
|
||||
Revision: "quickbrownfaux",
|
||||
},
|
||||
Data: map[string]interface{}{},
|
||||
Modules: []bundle.ModuleFile{
|
||||
{
|
||||
Path: `/example.rego`,
|
||||
Raw: []byte(`package test
|
||||
import rego.v1
|
||||
import data.foo
|
||||
import data.bar as foo
|
||||
p contains 1 if {
|
||||
input.x == 2
|
||||
}`),
|
||||
},
|
||||
},
|
||||
},
|
||||
"test/v1compat/keywords_not_used": {
|
||||
Manifest: bundle.Manifest{
|
||||
Revision: "quickbrownfaux",
|
||||
},
|
||||
Data: map[string]interface{}{},
|
||||
Modules: []bundle.ModuleFile{
|
||||
{
|
||||
Path: `/example.rego`,
|
||||
Raw: []byte(`package test
|
||||
p[1] {
|
||||
input.x == 2
|
||||
}`),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
if t.longPoll {
|
||||
|
||||
var timeout time.Duration
|
||||
|
||||
wait := getPreferHeaderField(r, "wait")
|
||||
if wait != "" {
|
||||
waitTime, err := strconv.Atoi(wait)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
timeout = time.Duration(waitTime) * time.Second
|
||||
}
|
||||
|
||||
// simulate long operation
|
||||
time.Sleep(timeout)
|
||||
}
|
||||
|
||||
if t.expCode != 0 {
|
||||
w.WriteHeader(t.expCode)
|
||||
|
||||
if t.expResp != "" {
|
||||
w.Write([]byte(t.expResp))
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if t.customAuth != nil {
|
||||
if err := t.customAuth(w, r); err != nil {
|
||||
t.t.Logf("Failed authorization: %s", err)
|
||||
if errors.Is(err, errUnauthorized) {
|
||||
w.WriteHeader(401)
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(500)
|
||||
return
|
||||
}
|
||||
} else if t.expAuth != "" {
|
||||
if r.Header.Get("Authorization") != t.expAuth {
|
||||
w.WriteHeader(401)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
|
||||
if strings.HasPrefix(r.URL.Path, "/v2/org/repo/") {
|
||||
// build test data to hash map to serve testdata files by hash
|
||||
if t.testdataHashes == nil {
|
||||
t.testdataHashes = make(map[string]fileInfo)
|
||||
files, err := os.ReadDir("testdata")
|
||||
if err != nil {
|
||||
t.t.Fatalf("failed to read testdata directory: %s", err)
|
||||
}
|
||||
for _, file := range files {
|
||||
if file.IsDir() {
|
||||
continue
|
||||
}
|
||||
hash, length, err := getFileSHAandSize("testdata/" + file.Name())
|
||||
if err != nil {
|
||||
t.t.Fatalf("failed to read testdata file: %s", err)
|
||||
}
|
||||
t.testdataHashes[fmt.Sprintf("%x", hash)] = fileInfo{name: file.Name(), length: length}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if strings.HasPrefix(r.URL.Path, "/v2/org/repo/blobs/sha256:") || strings.HasPrefix(r.URL.Path, "/v2/org/repo/manifests/sha256:") {
|
||||
sha := strings.TrimPrefix(strings.TrimPrefix(r.URL.Path, "/v2/org/repo/blobs/sha256:"), "/v2/org/repo/manifests/sha256:")
|
||||
if fileInfo, ok := t.testdataHashes[sha]; ok {
|
||||
w.Header().Add("Content-Length", strconv.Itoa(int(fileInfo.length)))
|
||||
w.Header().Add("Content-Type", "application/gzip")
|
||||
w.Header().Add("Docker-Content-Digest", "sha256:"+sha)
|
||||
w.WriteHeader(200)
|
||||
bs, err := os.ReadFile("testdata/" + fileInfo.name)
|
||||
if err != nil {
|
||||
w.WriteHeader(404)
|
||||
return
|
||||
}
|
||||
buf.WriteString(string(bs))
|
||||
w.Write(buf.Bytes())
|
||||
return
|
||||
}
|
||||
w.WriteHeader(404)
|
||||
return
|
||||
}
|
||||
|
||||
if strings.HasPrefix(r.URL.Path, "/v2/org/repo/manifests/") {
|
||||
sha, size, err := getFileSHAandSize("testdata/" + strings.TrimPrefix(r.URL.Path, "/v2/org/repo/manifests/") + ".manifest")
|
||||
if err != nil {
|
||||
w.WriteHeader(404)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Add("Content-Length", strconv.Itoa(int(size)))
|
||||
w.Header().Add("Content-Type", "application/vnd.oci.image.manifest.v1+json")
|
||||
w.Header().Add("Docker-Content-Digest", "sha256:"+fmt.Sprintf("%x", sha))
|
||||
w.WriteHeader(200)
|
||||
return
|
||||
}
|
||||
name := strings.TrimPrefix(r.URL.Path, "/bundles/")
|
||||
b, ok := t.bundles[name]
|
||||
if !ok {
|
||||
w.WriteHeader(404)
|
||||
return
|
||||
}
|
||||
|
||||
// check to verify if server can send a delta bundle to OPA
|
||||
if b.Manifest.Revision == deltaBundleMode {
|
||||
modes := strings.Split(getPreferHeaderField(r, "modes"), ",")
|
||||
|
||||
found := false
|
||||
for _, m := range modes {
|
||||
if m == deltaBundleMode {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if !found {
|
||||
panic("delta bundle requested but OPA does not support it")
|
||||
}
|
||||
}
|
||||
|
||||
contentTypeShouldBeSend := true
|
||||
if t.expEtag != "" {
|
||||
etag := r.Header.Get("If-None-Match")
|
||||
if etag == t.expEtag {
|
||||
contentTypeShouldBeSend = false
|
||||
if t.etagInResponse {
|
||||
w.Header().Add("Etag", t.expEtag)
|
||||
}
|
||||
w.WriteHeader(304)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if t.longPoll && contentTypeShouldBeSend {
|
||||
// in 304 Content-Type is not send according https://datatracker.ietf.org/doc/html/rfc7232#section-4.1
|
||||
w.Header().Add("Content-Type", "application/vnd.openpolicyagent.bundles")
|
||||
} else {
|
||||
if r.URL.Path == "/bundles/not-a-bundle" {
|
||||
w.Header().Add("Content-Type", "text/html")
|
||||
} else {
|
||||
w.Header().Add("Content-Type", "application/gzip")
|
||||
}
|
||||
}
|
||||
|
||||
if t.expEtag != "" {
|
||||
w.Header().Add("Etag", t.expEtag)
|
||||
}
|
||||
|
||||
w.WriteHeader(200)
|
||||
|
||||
if err := bundle.Write(&buf, b); err != nil {
|
||||
w.WriteHeader(500)
|
||||
}
|
||||
|
||||
if _, err := w.Write(buf.Bytes()); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *testServer) start() {
|
||||
t.server = httptest.NewServer(http.HandlerFunc(t.handle))
|
||||
}
|
||||
|
||||
func (t *testServer) stop() {
|
||||
t.server.Close()
|
||||
}
|
||||
|
||||
func getPreferHeaderField(r *http.Request, field string) string {
|
||||
for _, line := range r.Header.Values("prefer") {
|
||||
for _, part := range strings.Split(line, ";") {
|
||||
preference := strings.Split(strings.TrimSpace(part), "=")
|
||||
if len(preference) == 2 {
|
||||
if strings.ToLower(preference[0]) == field {
|
||||
return preference[1]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func getFileSHAandSize(filePath string) ([]byte, int64, error) {
|
||||
f, err := os.Open(filePath)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer f.Close()
|
||||
hash := sha256.New()
|
||||
w, err := io.Copy(hash, f)
|
||||
if err != nil {
|
||||
return nil, w, err
|
||||
}
|
||||
return hash.Sum(nil), w, nil
|
||||
}
|
||||
Reference in New Issue
Block a user