OCI: skip reloading bundle if tarball SHA did not change (#4658)

Fixes #4637.

Signed-off-by: carabasdaniel <dani@aserto.com>

download/oci_dowload.go

@@ -43,6 +43,7 @@ type OCIDownloader struct {
 	stopped        bool
 	persist        bool
 	store          *content.OCI
+	etag           string
 }
 
 // New returns a new Downloader that can be started.
@@ -98,8 +99,9 @@ func (d *OCIDownloader) WithBundlePersistence(persist bool) *OCIDownloader {
 func (d *OCIDownloader) ClearCache() {
 }
 
-// TODO: this will need implementation for the OCI downloader.
+// 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
@@ -213,6 +215,7 @@ func (d *OCIDownloader) oneShot(ctx context.Context) error {
 		}
 		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})
@@ -248,14 +251,26 @@ func (d *OCIDownloader) download(ctx context.Context, m metrics.Metrics) (*downl
 	if tarballDescriptor.MediaType == "" {
 		return nil, fmt.Errorf("no tarball descriptor found in the layers")
 	}
-
-	bundleFilePath := filepath.Join(d.localStorePath, "blobs", "sha256", string(tarballDescriptor.Digest.Hex()))
+	etag := string(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)
 	if err != nil {
 		return nil, err
 	}
 	loader := bundle.NewTarballLoaderWithBaseURL(fileReader, d.localStorePath)
-	reader := bundle.NewCustomReader(loader).WithBaseDir(d.localStorePath)
+	reader := bundle.NewCustomReader(loader).WithBaseDir(d.localStorePath).
+		WithMetrics(m).
+		WithBundleVerificationConfig(d.bvc).
+		WithBundleEtag(etag)
 	bundleInfo, err := reader.Read()
 	if err != nil {
 		return &downloaderResponse{}, fmt.Errorf("unexpected error %w", err)
@@ -266,7 +281,7 @@ func (d *OCIDownloader) download(ctx context.Context, m metrics.Metrics) (*downl
 	return &downloaderResponse{
 		b:        &bundleInfo,
 		raw:      fileReader,
-		etag:     "",
+		etag:     etag,
 		longPoll: false,
 	}, nil
 }

download/oci_download_test.go

@@ -84,3 +84,39 @@ func TestOCIFailureAuthn(t *testing.T) {
 		t.Fatal("expected 401 Unauthorized message")
 	}
 }
+
+func TestOCIEtag(t *testing.T) {
+	ctx := context.Background()
+	fixture := newTestFixture(t)
+	fixture.server.expAuth = "" // test on public registry
+	fixture.server.expEtag = "sha256:c5834dbce332cabe6ae68a364de171a50bf5b08024c27d7c08cc72878b4df7ff"
+	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
+	err := d.oneShot(ctx)
+	if err != nil {
+		t.Fatal("unexpected error")
+	}
+	// 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(ctx)
+	if err != nil {
+		t.Fatal("unexpected error")
+	}
+}

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/gorilla/mux v1.8.0
 	github.com/miekg/dns v1.1.43 // indirect
 	github.com/olekukonko/tablewriter v0.0.5
+	github.com/opencontainers/go-digest v1.0.0
 	github.com/opencontainers/image-spec v1.0.3-0.20211202183452-c5a74bcca799
 	github.com/peterh/liner v0.0.0-20170211195444-bf27d3ba8e1d
 	github.com/pkg/errors v0.9.1

sdk/test/test.go

@@ -2,16 +2,26 @@ package test
 
 import (
 	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/json"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"os"
+	"path/filepath"
 	"sort"
 	"strings"
+	"time"
 
 	"github.com/open-policy-agent/opa/ast"
 	"github.com/open-policy-agent/opa/bundle"
 	"github.com/open-policy-agent/opa/compile"
+
+	"github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 )
 
 // MockBundle sets a bundle named file on the test server containing the given
@@ -26,6 +36,17 @@ func MockBundle(file string, policies map[string]string) func(*Server) error {
 	}
 }
 
+// MockOCIBundle prepares the server to allow serving "/v2" OCI responses from the supplied policies
+// Ref parameter must be in the form of <registry>/<org>/<repo>:<tag> that will be used in detecting future calls
+func MockOCIBundle(ref string, policies map[string]string) func(*Server) error {
+	return func(s *Server) error {
+		if !strings.Contains(ref, "/") {
+			return fmt.Errorf("mock oci bundle ref must contain 'org/repo' but got %q", ref)
+		}
+		return s.buildBundles(ref, policies)
+	}
+}
+
 // Ready provides a channel that the server will use to gate readiness. The
 // caller can provide this channel to prevent the server from becoming ready.
 // The server will response with HTTP 500 responses until ready. The caller
@@ -87,6 +108,112 @@ func (s *Server) URL() string {
 	return s.server.URL
 }
 
+// Builds the tarball from the supplied policies and prepares the layers in a temporary directory
+func (s *Server) buildBundles(ref string, policies map[string]string) error {
+	// Prepare the modules to include in the bundle. Sort them so bundles are deterministic.
+	var modules []bundle.ModuleFile
+	for url, str := range policies {
+		module, err := ast.ParseModule(url, str)
+		if err != nil {
+			return fmt.Errorf("failed to parse module: %v", err)
+		}
+		modules = append(modules, bundle.ModuleFile{
+			URL:    url,
+			Parsed: module,
+		})
+	}
+	sort.Slice(modules, func(i, j int) bool {
+		return modules[i].URL < modules[j].URL
+	})
+
+	// Compile the bundle out into a buffer
+	buf := bytes.NewBuffer(nil)
+	err := compile.New().WithOutput(buf).WithBundle(&bundle.Bundle{
+		Data:    map[string]interface{}{},
+		Modules: modules,
+	}).Build(context.Background())
+	if err != nil {
+		return err
+	}
+	directoryName, err := os.MkdirTemp("", "oci-test-temp")
+	fmt.Println("Testing OCI temporary directory:", directoryName)
+	if err != nil {
+		return err
+	}
+	// Write buf tarball to layer
+	tarLayer := filepath.Join(directoryName, "tar.layer")
+	err = ioutil.WriteFile(tarLayer, buf.Bytes(), 0655)
+	if err != nil {
+		return err
+	}
+	// Write empty config layer
+	configLayer := filepath.Join(directoryName, "config.layer")
+	err = ioutil.WriteFile(configLayer, []byte("{}"), 0655)
+	if err != nil {
+		return err
+	}
+	// Calculate SHA and size and prepare manifest layer
+	tarSHA, err := getFileSHA(tarLayer)
+	if err != nil {
+		return err
+	}
+
+	configSHA, err := getFileSHA(configLayer)
+	if err != nil {
+		return err
+	}
+
+	var manifest ocispec.Manifest
+	manifest.SchemaVersion = 2
+	manifest.Config = ocispec.Descriptor{
+		MediaType: ocispec.MediaTypeImageConfig,
+		Digest:    digest.Digest(fmt.Sprintf("sha256:%x", configSHA)),
+		Size:      int64(2), // config size is set to 2 as an empty config is used
+	}
+	manifest.Layers = []ocispec.Descriptor{
+		{
+			MediaType: ocispec.MediaTypeImageLayerGzip,
+			Digest:    digest.Digest(fmt.Sprintf("sha256:%x", tarSHA)),
+			Size:      int64(buf.Len()),
+			Annotations: map[string]string{
+				ocispec.AnnotationTitle:   ref,
+				ocispec.AnnotationCreated: time.Now().Format(time.RFC3339),
+			},
+		},
+	}
+
+	manifestData, err := json.Marshal(manifest)
+	if err != nil {
+		return err
+	}
+	manifestLayer := filepath.Join(directoryName, "manifest.layer")
+	err = ioutil.WriteFile(manifestLayer, manifestData, 0655)
+	if err != nil {
+		return err
+	}
+
+	// Set ref layer paths to server bundles
+	s.bundles[ref] = map[string]string{
+		"manifest": manifestLayer,
+		"config":   configLayer,
+		"tar":      tarLayer,
+	}
+	return nil
+}
+
+func getFileSHA(filePath string) ([]byte, error) {
+	f, err := os.Open(filePath)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	hash := sha256.New()
+	if _, err := io.Copy(hash, f); err != nil {
+		return nil, err
+	}
+	return hash.Sum(nil), nil
+}
+
 func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
 
 	select {
@@ -96,6 +223,11 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if strings.HasPrefix(r.URL.Path, "/v2") {
+		s.handleOCIBundles(w, r)
+		return
+	}
+
 	if strings.HasPrefix(r.URL.Path, "/bundles") {
 		s.handleBundles(w, r)
 		return
@@ -104,6 +236,124 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusInternalServerError)
 }
 
+func (s *Server) handleOCIBundles(w http.ResponseWriter, r *http.Request) {
+	ref := ""  // key used to detect layers from s.bundles
+	tag := ""  // image tag used in request path verification
+	repo := "" // image repo used in request path verification
+	var buf bytes.Buffer
+	//get first key that matches request url pattern
+	for key := range s.bundles {
+		// extract tag
+		parsedRef := strings.Split(key, ":")
+		checkRef := strings.Split(parsedRef[0], "/")
+		// check if request path contains org and repository
+		if strings.Contains(r.URL.Path, checkRef[1]) && strings.Contains(r.URL.Path, checkRef[2]) {
+			ref = key
+			tag = parsedRef[1]
+			repo = checkRef[1] + "/" + checkRef[2]
+			break
+		}
+	}
+	if ref == "" || tag == "" || repo == "" {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	layers := s.bundles[ref]
+	fi, err := os.Stat(layers["manifest"])
+	if err != nil {
+		w.WriteHeader(http.StatusFailedDependency)
+		return
+	}
+	manifestSize := fi.Size()
+	manifestSHA, err := getFileSHA(layers["manifest"])
+	if err != nil {
+		w.WriteHeader(http.StatusFailedDependency)
+		return
+	}
+	fi, err = os.Stat(layers["config"])
+	if err != nil {
+		w.WriteHeader(http.StatusFailedDependency)
+		return
+	}
+	configSize := fi.Size()
+	configSHA, err := getFileSHA(layers["config"])
+	if err != nil {
+		w.WriteHeader(http.StatusFailedDependency)
+		return
+	}
+	fi, err = os.Stat(layers["tar"])
+	if err != nil {
+		w.WriteHeader(http.StatusFailedDependency)
+		return
+	}
+	// get the size
+	tarSize := fi.Size()
+	tarSHA, err := getFileSHA(layers["tar"])
+	if err != nil {
+		w.WriteHeader(http.StatusFailedDependency)
+		return
+	}
+
+	if r.URL.Path == fmt.Sprintf("/v2/%s/manifests/%s", repo, tag) {
+		w.Header().Add("Content-Length", fmt.Sprintf("%d", manifestSize))
+		w.Header().Add("Content-Type", "application/vnd.oci.image.manifest.v1+json")
+		w.Header().Add("Docker-Content-Digest", fmt.Sprintf("sha256:%x", manifestSHA))
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+	if r.URL.Path == fmt.Sprintf("/v2/%s/manifests/sha256:%x", repo, manifestSHA) {
+		w.Header().Add("Content-Length", fmt.Sprintf("%d", manifestSize))
+		w.Header().Add("Content-Type", "application/vnd.oci.image.manifest.v1+json")
+		w.Header().Add("Docker-Content-Digest", fmt.Sprintf("sha256:%x", manifestSHA))
+		w.WriteHeader(200)
+		bs, err := ioutil.ReadFile(layers["manifest"])
+		if err != nil {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		buf.WriteString(string(bs))
+		_, err = w.Write(buf.Bytes())
+		if err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+		}
+		return
+	}
+	if r.URL.Path == fmt.Sprintf("/v2/%s/blobs/sha256:%x", repo, configSHA) {
+		w.Header().Add("Content-Length", fmt.Sprintf("%d", configSize))
+		w.Header().Add("Content-Type", "application/vnd.oci.image.manifest.v1+json")
+		w.Header().Add("Docker-Content-Digest", fmt.Sprintf("sha256:%x", configSHA))
+		w.WriteHeader(200)
+		bs, err := ioutil.ReadFile(layers["config"])
+		if err != nil {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		buf.WriteString(string(bs))
+		_, err = w.Write(buf.Bytes())
+		if err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+		}
+		return
+	}
+	if r.URL.Path == fmt.Sprintf("/v2/%s/blobs/sha256:%x", repo, tarSHA) {
+		w.Header().Add("Content-Length", fmt.Sprintf("%d", tarSize))
+		w.Header().Add("Content-Type", "application/vnd.oci.image.manifest.v1+json")
+		w.Header().Add("Docker-Content-Digest", fmt.Sprintf("sha256:%x", tarSHA))
+		w.WriteHeader(200)
+		bs, err := ioutil.ReadFile(layers["tar"])
+		if err != nil {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		buf.WriteString(string(bs))
+		_, err = w.Write(buf.Bytes())
+		if err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+		}
+		return
+	}
+}
+
 func (s *Server) handleBundles(w http.ResponseWriter, r *http.Request) {
 
 	// Return 404 if bundle path does not exist.

test/e2e/oci/oci_test.go

@@ -0,0 +1,116 @@
+package test
+
+import (
+	"bytes"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/open-policy-agent/opa/logging"
+	test_sdk "github.com/open-policy-agent/opa/sdk/test"
+	"github.com/open-policy-agent/opa/test/e2e"
+)
+
+type SafeBuffer struct {
+	b bytes.Buffer
+	m sync.Mutex
+}
+
+func (b *SafeBuffer) Read(p []byte) (n int, err error) {
+	b.m.Lock()
+	defer b.m.Unlock()
+	return b.b.Read(p)
+}
+func (b *SafeBuffer) Write(p []byte) (n int, err error) {
+	b.m.Lock()
+	defer b.m.Unlock()
+	return b.b.Write(p)
+}
+func (b *SafeBuffer) String() string {
+	b.m.Lock()
+	defer b.m.Unlock()
+	return b.b.String()
+}
+
+func TestEnablePrintStatementsForBundles(t *testing.T) {
+	ref := "registry.io/someorg/somerepo:tag"
+	server := test_sdk.MustNewServer(test_sdk.MockOCIBundle(ref, map[string]string{
+		"post.rego": `
+		package peoplefinder.POST.api.users
+
+		import input.user.properties as user_props
+		
+		default allowed = false
+		
+		allowed {
+			user_props.department == "Operations"
+			user_props.title == "IT Manager"
+		}	
+		`,
+	}))
+	params := e2e.NewAPIServerTestParams()
+
+	buf := SafeBuffer{}
+
+	logger := logging.New()
+	logger.SetLevel(logging.Debug) // set to debug to see the the bundle download skip message
+	logger.SetOutput(&buf)
+	params.Logger = logger
+
+	params.ConfigOverrides = []string{
+		"services.test.url=" + server.URL(),
+		"services.test.type=oci",
+		"bundles.test.resource=" + ref,
+		"bundles.test.polling.min_delay_seconds=1",
+		"bundles.test.polling.max_delay_seconds=3",
+	}
+
+	// Test runtime uses the local OCI image layers stored in download testdata that contain the
+	// rego policies based on the https://github.com/aserto-dev/policy-peoplefinder-abac template
+	e2e.WithRuntime(t, e2e.TestRuntimeOpts{WaitForBundles: true}, params, func(rt *e2e.TestRuntime) {
+		var readBuf []byte
+		type Props struct {
+			Department string `json:"department"`
+			Title      string `json:"title"`
+		}
+		type Attributes struct {
+			Properties Props `json:"properties"`
+		}
+		type Input struct {
+			User Attributes `json:"user"`
+		}
+
+		inputAllowed := Input{User: Attributes{Properties: Props{Department: "Operations", Title: "IT Manager"}}}
+
+		inputNotAllowed := Input{User: Attributes{Properties: Props{Department: "IT", Title: "Engineer"}}}
+
+		readBuf, err := rt.GetDataWithInput("peoplefinder/POST/api/users/allowed", inputAllowed)
+		if err != nil {
+			t.Fatal("failed to get data from runtime")
+		}
+
+		response := string(readBuf)
+		if !strings.Contains(response, "true") {
+			t.Fatalf("expected true but got: %s", response)
+		}
+		readBuf, err = rt.GetDataWithInput("peoplefinder/POST/api/users/allowed", inputNotAllowed)
+		if err != nil {
+			t.Fatal("failed to get data from runtime")
+		}
+		if !strings.Contains(string(readBuf), "false") {
+			t.Fatalf("expected true but got: %s", response)
+		}
+
+		time.Sleep(3 * time.Second) // wait for the downloader pooling mechanism to kick in
+		expContains := "Bundle loaded and activated successfully"
+		skipContains := "Bundle load skipped, server replied with not modified."
+
+		if !strings.Contains(buf.String(), expContains) {
+			t.Fatalf("expected logs to contain %q but got: %v", expContains, buf.String())
+		}
+		if !strings.Contains(buf.String(), skipContains) {
+			t.Fatalf("expected logs to contain %q but got: %v", skipContains, buf.String())
+		}
+	})
+}

vendor/modules.txt

@@ -196,6 +196,7 @@ github.com/morikuni/aec
 ## explicit
 github.com/olekukonko/tablewriter
 # github.com/opencontainers/go-digest v1.0.0
+## explicit
 github.com/opencontainers/go-digest
 # github.com/opencontainers/image-spec v1.0.3-0.20211202183452-c5a74bcca799
 ## explicit