feat: Add ability to retrieve subjects for an envelope

Implements the GetSubjects RPC. Adds a retrieve subjects command to
archivistctl.

Signed-off-by: Mikhail Swift <[email protected]>

cmd/archivist/main.go

@@ -31,7 +31,7 @@ import (
 	"github.com/testifysec/archivist-api/pkg/api/archivist"
 	"github.com/testifysec/archivist/internal/config"
 	"github.com/testifysec/archivist/internal/metadatastorage/mysqlstore"
-	blob "github.com/testifysec/archivist/internal/objectstorage/blobstore"
+	"github.com/testifysec/archivist/internal/objectstorage/blobstore"
 	"github.com/testifysec/archivist/internal/objectstorage/filestore"
 	"github.com/testifysec/archivist/internal/server"
 
@@ -172,7 +172,7 @@ func initObjectStore(ctx context.Context, cfg *config.Config) (server.ObjectStor
 		return filestore.New(ctx, cfg.FileDir, cfg.FileServeOn)
 
 	case "BLOB":
-		return blob.New(
+		return blobstore.New(
 			ctx,
 			cfg.BlobStoreEndpoint,
 			cfg.BlobStoreAccessKeyId,

cmd/archivistctl/cmd/retrieve.go

@@ -3,8 +3,10 @@ package cmd
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"os"
+	"strings"
 
 	"github.com/spf13/cobra"
 	"github.com/testifysec/archivist-api/pkg/api/archivist"
@@ -15,6 +17,12 @@ var (
 
 	retrieveCmd = &cobra.Command{
 		Use:          "retrieve",
+		Short:        "Retrieve information from an archivist server",
+		SilenceUsage: true,
+	}
+
+	envelopeCmd = &cobra.Command{
+		Use:          "envelope",
 		Short:        "Retrieves a dsse envelope by it's gitoid from archivist",
 		SilenceUsage: true,
 		Args:         cobra.ExactArgs(1),
@@ -38,11 +46,59 @@ var (
 			return retrieveEnvelope(cmd.Context(), archivist.NewCollectorClient(conn), args[0], out)
 		},
 	}
+
+	subjectCmd = &cobra.Command{
+		Use:          "subjects",
+		Short:        "Retrieves all subjects on an in-toto statement by the envelope gitoid",
+		SilenceUsage: true,
+		Args:         cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			conn, err := newConn(archivistUrl)
+			if err != nil {
+				return err
+			}
+
+			return retrieveSubjects(cmd.Context(), archivist.NewArchivistClient(conn), args[0])
+		},
+	}
 )
 
 func init() {
 	rootCmd.AddCommand(retrieveCmd)
-	retrieveCmd.Flags().StringVarP(&outFile, "out", "o", "", "File to write the envelope out to. Defaults to stdout")
+	retrieveCmd.AddCommand(envelopeCmd)
+	retrieveCmd.AddCommand(subjectCmd)
+	envelopeCmd.Flags().StringVarP(&outFile, "out", "o", "", "File to write the envelope out to. Defaults to stdout")
+}
+
+func retrieveSubjects(ctx context.Context, client archivist.ArchivistClient, gitoid string) error {
+	stream, err := client.GetSubjects(ctx, &archivist.GetSubjectsRequest{EnvelopeGitoid: gitoid})
+	if err != nil {
+		return err
+	}
+
+	for {
+		subject, err := stream.Recv()
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			return err
+		}
+
+		fmt.Printf("Name: %s\nDigests:\n%s\n", subject.GetName(), digestString(subject.GetDigest()))
+	}
+
+	return nil
+}
+
+func digestString(digest map[string]string) string {
+	sb := strings.Builder{}
+	for algo, value := range digest {
+		sb.WriteString(fmt.Sprintf("Algo: %s\nValue: %s\n", algo, value))
+	}
+
+	return sb.String()
 }
 
 func retrieveEnvelope(ctx context.Context, client archivist.CollectorClient, gitoid string, out io.Writer) error {

compose.yml

@@ -15,7 +15,7 @@ services:
 
   archivist:
     build: .
-    restart: always
+    restart: unless-stopped
     environment:
       ARCHIVIST_ENABLE_SPIFFE: "false"
       ARCHIVIST_LISTEN_ON: tcp://0.0.0.0:8080

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/cobra v1.4.0
 	github.com/spiffe/go-spiffe/v2 v2.0.0
-	github.com/testifysec/archivist-api v0.0.0-20220719182705-980989009502
+	github.com/testifysec/archivist-api v0.0.0-20220719182744-ef1778aa0dba
 	github.com/testifysec/go-witness v0.1.11
 	google.golang.org/grpc v1.46.0
 )

go.sum

@@ -272,8 +272,8 @@ github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
-github.com/testifysec/archivist-api v0.0.0-20220719182705-980989009502 h1:9y2qBW9yrxDa5R+mHr4uHmPIOIkyjhCD5XzNtynHVMU=
-github.com/testifysec/archivist-api v0.0.0-20220719182705-980989009502/go.mod h1:HWpNFd8qFXCoU8gEF/xiuG10ni9EBFhPcpAFTcWDAmc=
+github.com/testifysec/archivist-api v0.0.0-20220719182744-ef1778aa0dba h1:TPJ+JyjlGYZcyHvkd8LJf+bJaLvDDKNQAfP4EGgfO58=
+github.com/testifysec/archivist-api v0.0.0-20220719182744-ef1778aa0dba/go.mod h1:HWpNFd8qFXCoU8gEF/xiuG10ni9EBFhPcpAFTcWDAmc=
 github.com/testifysec/go-witness v0.1.11 h1:CK5I7g7yu+ObXraYN96KHZu9VmLLs4vKEvfcEi4E35E=
 github.com/testifysec/go-witness v0.1.11/go.mod h1:EGTMK84vV6/7kiCbJYonESTvaeOW2eMJVHh3mW/EWYU=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=

internal/metadatastorage/mysqlstore/mysql.go

@@ -135,6 +135,44 @@ func (s *Store) GetBySubjectDigest(ctx context.Context, request *archivist.GetBy
 	return out, nil
 }
 
+func (s *Store) GetSubjects(ctx context.Context, req *archivist.GetSubjectsRequest) (<-chan *archivist.GetSubjectsResponse, error) {
+	subjects, err := s.client.Subject.Query().Where(
+		subject.HasStatementWith(
+			statement.HasDsseWith(
+				entdsse.GitbomSha256(req.GetEnvelopeGitoid()),
+			),
+		),
+	).WithSubjectDigests().All(ctx)
+
+	if err != nil {
+		return nil, err
+	}
+
+	out := make(chan *archivist.GetSubjectsResponse, 1)
+	go func() {
+		defer close(out)
+
+		for _, subject := range subjects {
+			response := &archivist.GetSubjectsResponse{
+				Name:   subject.Name,
+				Digest: make(map[string]string),
+			}
+
+			for _, digest := range subject.Edges.SubjectDigests {
+				response.Digest[digest.Algorithm] = digest.Value
+			}
+
+			select {
+			case <-ctx.Done():
+				return
+			case out <- response:
+			}
+		}
+	}()
+
+	return out, nil
+}
+
 func (s *Store) withTx(ctx context.Context, fn func(tx *ent.Tx) error) error {
 	tx, err := s.client.Tx(ctx)
 	if err != nil {

internal/objectstorage/blobstore/minio.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package blob
+package blobstore
 
 import (
 	"bytes"

internal/server/server.go

@@ -56,6 +56,23 @@ func (s *archivistServer) GetBySubjectDigest(request *archivist.GetBySubjectDige
 	return nil
 }
 
+func (s *archivistServer) GetSubjects(req *archivist.GetSubjectsRequest, server archivist.Archivist_GetSubjectsServer) error {
+	ctx, cancel := context.WithCancel(server.Context())
+	defer cancel()
+	subjects, err := s.store.GetSubjects(ctx, req)
+	if err != nil {
+		return err
+	}
+
+	for subject := range subjects {
+		if err := server.Send(subject); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 type collectorServer struct {
 	archivist.UnimplementedCollectorServer
 
@@ -70,6 +87,7 @@ type Storer interface {
 type MetadataStorer interface {
 	Storer
 	GetBySubjectDigest(context.Context, *archivist.GetBySubjectDigestRequest) (<-chan *archivist.GetBySubjectDigestResponse, error)
+	GetSubjects(context.Context, *archivist.GetSubjectsRequest) (<-chan *archivist.GetSubjectsResponse, error)
 }
 
 type ObjectStorer interface {