Merge branch 'main' into chunked

.github/workflows/integration-test.yaml

@@ -16,6 +16,7 @@ permissions:
 
 jobs:
   test:
+    if: github.event_name == 'push'
     name: Run tests
     uses: open-component-model/ocm-integrationtest/.github/workflows/integrationtest.yaml@main
     permissions:
@@ -24,4 +25,17 @@ jobs:
       packages: write
     secrets: inherit
     with:
-      ref: ${{ github.ref }}
+      ref: ${{ github.ref }}
+      repo: ${{ github.repository }}
+  test-pr:
+    if: github.event_name == 'pull_request_target'
+    name: Run tests
+    uses: open-component-model/ocm-integrationtest/.github/workflows/integrationtest.yaml@main
+    permissions:
+      contents: write
+      id-token: write
+      packages: write
+    secrets: inherit
+    with:
+      ref: ${{ github.event.pull_request.head.ref }}
+      repo: ${{ github.event.pull_request.head.repo.full_name }}

api/oci/extensions/repositories/ocireg/repository.go

@@ -11,6 +11,7 @@ import (
 	"github.com/containerd/errdefs"
 	"github.com/mandelsoft/goutils/errors"
 	"github.com/mandelsoft/logging"
+	"github.com/moby/locker"
 	"oras.land/oras-go/v2/registry/remote/auth"
 	"oras.land/oras-go/v2/registry/remote/retry"
 
@@ -165,14 +166,22 @@ func (r *RepositoryImpl) getResolver(comp string) (oras.Resolver, error) {
 	}
 
 	authClient := &auth.Client{
-		Client:     client,
-		Cache:      auth.NewCache(),
-		Credential: auth.StaticCredential(r.info.HostPort(), authCreds),
+		Client: client,
+		Cache:  auth.NewCache(),
+		Credential: auth.CredentialFunc(func(ctx context.Context, hostport string) (auth.Credential, error) {
+			if strings.Contains(hostport, r.info.HostPort()) {
+				return authCreds, nil
+			}
+			logger.Warn("no credentials for host", "host", hostport)
+			return auth.EmptyCredential, nil
+		}),
 	}
 
 	return oras.New(oras.ClientOptions{
 		Client:    authClient,
 		PlainHTTP: r.info.Scheme == "http",
+		Logger:    logger,
+		Lock:      locker.New(),
 	}), nil
 }
 

api/tech/oras/client.go

@@ -4,11 +4,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
-	"strings"
-	"sync"
 
 	"github.com/containerd/containerd/errdefs"
+	"github.com/mandelsoft/logging"
+	"github.com/moby/locker"
 	ociv1 "github.com/opencontainers/image-spec/specs-go/v1"
 	oraserr "oras.land/oras-go/v2/errdef"
 	"oras.land/oras-go/v2/registry/remote"
@@ -18,55 +17,37 @@ import (
 type ClientOptions struct {
 	Client    *auth.Client
 	PlainHTTP bool
+	Logger    logging.Logger
+	Lock      *locker.Locker
 }
 
 type Client struct {
 	client    *auth.Client
 	plainHTTP bool
-	ref       string
-	mu        sync.RWMutex
+	logger    logging.Logger
+	lock      *locker.Locker
 }
 
-var (
-	_ Resolver = &Client{}
-	_ Fetcher  = &Client{}
-	_ Pusher   = &Client{}
-	_ Lister   = &Client{}
-)
+var _ Resolver = &Client{}
 
 func New(opts ClientOptions) *Client {
-	return &Client{client: opts.Client, plainHTTP: opts.PlainHTTP}
+	return &Client{client: opts.Client, plainHTTP: opts.PlainHTTP, logger: opts.Logger, lock: opts.Lock}
 }
 
 func (c *Client) Fetcher(ctx context.Context, ref string) (Fetcher, error) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.ref = ref
-	return c, nil
+	return &OrasFetcher{client: c.client, ref: ref, plainHTTP: c.plainHTTP}, nil
 }
 
 func (c *Client) Pusher(ctx context.Context, ref string) (Pusher, error) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.ref = ref
-	return c, nil
+	return &OrasPusher{client: c.client, ref: ref, plainHTTP: c.plainHTTP, lock: c.lock}, nil
 }
 
 func (c *Client) Lister(ctx context.Context, ref string) (Lister, error) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.ref = ref
-	return c, nil
+	return &OrasLister{client: c.client, ref: ref, plainHTTP: c.plainHTTP}, nil
 }
 
 func (c *Client) Resolve(ctx context.Context, ref string) (string, ociv1.Descriptor, error) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	src, err := c.createRepository(ref)
+	src, err := createRepository(ref, c.client, c.plainHTTP)
 	if err != nil {
 		return "", ociv1.Descriptor{}, err
 	}
@@ -88,114 +69,16 @@ func (c *Client) Resolve(ctx context.Context, ref string) (string, ociv1.Descrip
 	return "", desc, nil
 }
 
-func (c *Client) Push(ctx context.Context, d ociv1.Descriptor, src Source) error {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	reader, err := src.Reader()
-	if err != nil {
-		return err
-	}
-
-	repository, err := c.createRepository(c.ref)
-	if err != nil {
-		return err
-	}
-
-	if split := strings.Split(c.ref, ":"); len(split) == 2 {
-		// Once we get a reference that contains a tag, we need to re-push that
-		// layer with the reference included. PushReference then will tag
-		// that layer resulting in the created tag pointing to the right
-		// blob data.
-		if err := repository.PushReference(ctx, d, reader, c.ref); err != nil {
-			return fmt.Errorf("failed to push tag: %w", err)
-		}
-
-		return nil
-	}
-
-	// We have a digest, so we use plain push for the digest.
-	// Push here decides if it's a Manifest or a Blob.
-	if err := repository.Push(ctx, d, reader); err != nil {
-		return fmt.Errorf("failed to push: %w, %s", err, c.ref)
-	}
-
-	return nil
-}
-
-func (c *Client) Fetch(ctx context.Context, desc ociv1.Descriptor) (io.ReadCloser, error) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	src, err := c.createRepository(c.ref)
-	if err != nil {
-		return nil, fmt.Errorf("failed to resolve ref %q: %w", c.ref, err)
-	}
-
-	// oras requires a Resolve to happen before a fetch because
-	// -1 is an invalid size and results in a content-length mismatch error by design.
-	// This is a security consideration on ORAS' side.
-	// manifest is not set in the descriptor
-	// We explicitly call resolve on manifest first because it might be
-	// that the mediatype is not set at this point so we don't want ORAS to try to
-	// select the wrong layer to fetch from.
-	rdesc, err := src.Manifests().Resolve(ctx, desc.Digest.String())
-	if errors.Is(err, oraserr.ErrNotFound) {
-		rdesc, err = src.Blobs().Resolve(ctx, desc.Digest.String())
-		if err != nil {
-			return nil, fmt.Errorf("failed to resolve fetch blob %q: %w", desc.Digest.String(), err)
-		}
-
-		delayer := func() (io.ReadCloser, error) {
-			return src.Blobs().Fetch(ctx, rdesc)
-		}
-
-		return newDelayedReader(delayer)
-	}
-
-	if err != nil {
-		return nil, fmt.Errorf("failed to resolve fetch manifest %q: %w", desc.Digest.String(), err)
-	}
-
-	// lastly, try a manifest fetch.
-	fetch, err := src.Fetch(ctx, rdesc)
-	if err != nil {
-		return nil, fmt.Errorf("failed to fetch manifest: %w", err)
-	}
-
-	return fetch, err
-}
-
-func (c *Client) List(ctx context.Context) ([]string, error) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	src, err := c.createRepository(c.ref)
-	if err != nil {
-		return nil, fmt.Errorf("failed to resolve ref %q: %w", c.ref, err)
-	}
-
-	var result []string
-	if err := src.Tags(ctx, "", func(tags []string) error {
-		result = append(result, tags...)
-		return nil
-	}); err != nil {
-		return nil, fmt.Errorf("failed to list tags: %w", err)
-	}
-
-	return result, nil
-}
-
 // createRepository creates a new repository representation using the passed in ref.
 // This is a cheap operation.
-func (c *Client) createRepository(ref string) (*remote.Repository, error) {
+func createRepository(ref string, client *auth.Client, plain bool) (*remote.Repository, error) {
 	src, err := remote.NewRepository(ref)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create new repository: %w", err)
 	}
 
-	src.Client = c.client // set up authenticated client.
-	src.PlainHTTP = c.plainHTTP
+	src.Client = client // set up authenticated client.
+	src.PlainHTTP = plain
 
 	return src, nil
 }