diff --git a/pkg/client/v1/proto/Makefile b/pkg/client/v1/proto/Makefile index ff048fa..be9b795 100644 --- a/pkg/client/v1/proto/Makefile +++ b/pkg/client/v1/proto/Makefile @@ -1,7 +1,7 @@ .PHONY: proto install-deps help proto: - @echo "Generating gRPC code from proto files using protoc-gen-gogo..." + @echo "Generating message bindings with protoc-gen-gogo..." @cd ../../../.. && go mod vendor @protoc \ --proto_path=. \ @@ -9,20 +9,28 @@ proto: --gogo_out=. \ --gogo_opt=paths=source_relative \ storage_service.proto + @echo "Generating gRPC service stubs with protoc-gen-go-grpc..." + @protoc \ + --proto_path=. \ + --proto_path=../../../../vendor \ + --go-grpc_out=. \ + --go-grpc_opt=paths=source_relative \ + storage_service.proto + @cd ../../../.. && rm -rf vendor @echo "Proto generation complete!" @echo "" @echo "IMPORTANT: Commit the generated files (*.pb.go, *_grpc.pb.go) to the repository" - @echo " You can delete the vendor directory after generation" install-deps: @echo "Installing protoc dependencies..." go install github.com/gogo/protobuf/protoc-gen-gogo@latest + go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest @echo "Dependencies installed!" help: @echo "Available targets:" - @echo " make install-deps - Install protoc-gen-gogo" - @echo " make proto - Generate Go code from proto files using gogo (runs go mod vendor first)" + @echo " make install-deps - Install protoc-gen-gogo + protoc-gen-go-grpc" + @echo " make proto - Generate Go code from proto files (messages via gogo, services via go-grpc)" @echo "" @echo "After running 'make proto', commit the generated files to git:" @echo " git add *.pb.go *_grpc.pb.go" diff --git a/pkg/client/v1/proto/storage_service.pb.go b/pkg/client/v1/proto/storage_service.pb.go index d7968c7..f8d4158 100644 --- a/pkg/client/v1/proto/storage_service.pb.go +++ b/pkg/client/v1/proto/storage_service.pb.go @@ -33,6 +33,8 @@ const ( ErrorCode_ERROR_CODE_PROFILE_NOT_FOUND ErrorCode = 5 ErrorCode_ERROR_CODE_INTERNAL_ERROR ErrorCode = 6 ErrorCode_ERROR_CODE_PULSAR_ERROR ErrorCode = 7 + ErrorCode_ERROR_CODE_SBOM_NOT_FOUND ErrorCode = 8 + ErrorCode_ERROR_CODE_SBOM_TOO_LARGE ErrorCode = 9 ) var ErrorCode_name = map[int32]string{ @@ -44,6 +46,8 @@ var ErrorCode_name = map[int32]string{ 5: "ERROR_CODE_PROFILE_NOT_FOUND", 6: "ERROR_CODE_INTERNAL_ERROR", 7: "ERROR_CODE_PULSAR_ERROR", + 8: "ERROR_CODE_SBOM_NOT_FOUND", + 9: "ERROR_CODE_SBOM_TOO_LARGE", } var ErrorCode_value = map[string]int32{ @@ -55,6 +59,8 @@ var ErrorCode_value = map[string]int32{ "ERROR_CODE_PROFILE_NOT_FOUND": 5, "ERROR_CODE_INTERNAL_ERROR": 6, "ERROR_CODE_PULSAR_ERROR": 7, + "ERROR_CODE_SBOM_NOT_FOUND": 8, + "ERROR_CODE_SBOM_TOO_LARGE": 9, } func (x ErrorCode) String() string { @@ -65,6 +71,41 @@ func (ErrorCode) EnumDescriptor() ([]byte, []int) { return fileDescriptor_3d90829bc66d9c54, []int{0} } +// SBOMSource identifies how an SBOM entered the backend. +type SBOMSource int32 + +const ( + SBOMSource_SBOM_SOURCE_UNSPECIFIED SBOMSource = 0 + // SBOM_SOURCE_WORKLOAD: generated by node-agent for an in-cluster workload image. + SBOMSource_SBOM_SOURCE_WORKLOAD SBOMSource = 1 + // SBOM_SOURCE_REGISTRY: generated by kubevuln for a registry image (Registry Epic). + SBOMSource_SBOM_SOURCE_REGISTRY SBOMSource = 2 + // SBOM_SOURCE_HOST: generated by the host scanner for an EC2 host snapshot. + SBOMSource_SBOM_SOURCE_HOST SBOMSource = 3 +) + +var SBOMSource_name = map[int32]string{ + 0: "SBOM_SOURCE_UNSPECIFIED", + 1: "SBOM_SOURCE_WORKLOAD", + 2: "SBOM_SOURCE_REGISTRY", + 3: "SBOM_SOURCE_HOST", +} + +var SBOMSource_value = map[string]int32{ + "SBOM_SOURCE_UNSPECIFIED": 0, + "SBOM_SOURCE_WORKLOAD": 1, + "SBOM_SOURCE_REGISTRY": 2, + "SBOM_SOURCE_HOST": 3, +} + +func (x SBOMSource) String() string { + return proto.EnumName(SBOMSource_name, int32(x)) +} + +func (SBOMSource) EnumDescriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{1} +} + // SendContainerProfileRequest contains the container profile to be stored // customer_guid, cluster, host_type, and host_id are sent via gRPC metadata headers type SendContainerProfileRequest struct { @@ -167,7 +208,7 @@ func (m *SendContainerProfileResponse) GetErrorCode() ErrorCode { // GetProfileRequest requests an aggregated profile // customer_guid, cluster, host_type, and host_id are sent via gRPC metadata headers type GetProfileRequest struct { - // Kind specifies the type of profile: "applicationProfile" or "networkNeighborhood" + // Kind specifies the type of profile: "applicationProfile", "networkNeighborhood", or "containerProfile" Kind string `protobuf:"bytes,1,opt,name=kind,proto3" json:"kind,omitempty"` // Namespace of the workload (k8s scope identifier) Namespace string `protobuf:"bytes,2,opt,name=namespace,proto3" json:"namespace,omitempty"` @@ -632,8 +673,725 @@ func (m *ListNetworkNeighborhoodsResponse) GetCont() string { return "" } +// ContainerProfileChunk is a single chunk of a streamed ContainerProfile +// upload (SendContainerProfileStream). The first chunk in the stream MUST +// set metadata. Subsequent chunks set only blob_chunk. Concatenated +// blob_chunk values form the marshaled ContainerProfile proto payload. +type ContainerProfileChunk struct { + // metadata is set on the first chunk only. + Metadata *ContainerProfileChunkMetadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"` + // blob_chunk is a slice of the marshaled ContainerProfile payload. + BlobChunk []byte `protobuf:"bytes,2,opt,name=blob_chunk,json=blobChunk,proto3" json:"blob_chunk,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *ContainerProfileChunk) Reset() { *m = ContainerProfileChunk{} } +func (m *ContainerProfileChunk) String() string { return proto.CompactTextString(m) } +func (*ContainerProfileChunk) ProtoMessage() {} +func (*ContainerProfileChunk) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{8} +} +func (m *ContainerProfileChunk) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_ContainerProfileChunk.Unmarshal(m, b) +} +func (m *ContainerProfileChunk) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_ContainerProfileChunk.Marshal(b, m, deterministic) +} +func (m *ContainerProfileChunk) XXX_Merge(src proto.Message) { + xxx_messageInfo_ContainerProfileChunk.Merge(m, src) +} +func (m *ContainerProfileChunk) XXX_Size() int { + return xxx_messageInfo_ContainerProfileChunk.Size(m) +} +func (m *ContainerProfileChunk) XXX_DiscardUnknown() { + xxx_messageInfo_ContainerProfileChunk.DiscardUnknown(m) +} + +var xxx_messageInfo_ContainerProfileChunk proto.InternalMessageInfo + +func (m *ContainerProfileChunk) GetMetadata() *ContainerProfileChunkMetadata { + if m != nil { + return m.Metadata + } + return nil +} + +func (m *ContainerProfileChunk) GetBlobChunk() []byte { + if m != nil { + return m.BlobChunk + } + return nil +} + +// ContainerProfileChunkMetadata is the metadata header sent on the first +// chunk of a SendContainerProfileStream call. All identifying fields +// (cluster, host_type, host_id) continue to be carried via gRPC metadata +// headers as in the unary path; this message is reserved for any future +// per-call options that the unary form would have put in the request. +type ContainerProfileChunkMetadata struct { + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *ContainerProfileChunkMetadata) Reset() { *m = ContainerProfileChunkMetadata{} } +func (m *ContainerProfileChunkMetadata) String() string { return proto.CompactTextString(m) } +func (*ContainerProfileChunkMetadata) ProtoMessage() {} +func (*ContainerProfileChunkMetadata) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{9} +} +func (m *ContainerProfileChunkMetadata) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_ContainerProfileChunkMetadata.Unmarshal(m, b) +} +func (m *ContainerProfileChunkMetadata) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_ContainerProfileChunkMetadata.Marshal(b, m, deterministic) +} +func (m *ContainerProfileChunkMetadata) XXX_Merge(src proto.Message) { + xxx_messageInfo_ContainerProfileChunkMetadata.Merge(m, src) +} +func (m *ContainerProfileChunkMetadata) XXX_Size() int { + return xxx_messageInfo_ContainerProfileChunkMetadata.Size(m) +} +func (m *ContainerProfileChunkMetadata) XXX_DiscardUnknown() { + xxx_messageInfo_ContainerProfileChunkMetadata.DiscardUnknown(m) +} + +var xxx_messageInfo_ContainerProfileChunkMetadata proto.InternalMessageInfo + +// GetContainerProfileStreamRequest mirrors the ContainerProfile-shaped +// fields of GetProfileRequest. customer_guid, cluster, host_type, and +// host_id are sent via gRPC metadata headers. +type GetContainerProfileStreamRequest struct { + // Namespace of the workload. + Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` + // Name of the container profile to fetch. + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` + // Region of the resource (non-k8s scope identifier). + Region string `protobuf:"bytes,3,opt,name=region,proto3" json:"region,omitempty"` + // CloudAccountIdentifier of the resource (non-k8s scope identifier). + CloudAccountIdentifier string `protobuf:"bytes,4,opt,name=cloud_account_identifier,json=cloudAccountIdentifier,proto3" json:"cloud_account_identifier,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *GetContainerProfileStreamRequest) Reset() { *m = GetContainerProfileStreamRequest{} } +func (m *GetContainerProfileStreamRequest) String() string { return proto.CompactTextString(m) } +func (*GetContainerProfileStreamRequest) ProtoMessage() {} +func (*GetContainerProfileStreamRequest) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{10} +} +func (m *GetContainerProfileStreamRequest) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_GetContainerProfileStreamRequest.Unmarshal(m, b) +} +func (m *GetContainerProfileStreamRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_GetContainerProfileStreamRequest.Marshal(b, m, deterministic) +} +func (m *GetContainerProfileStreamRequest) XXX_Merge(src proto.Message) { + xxx_messageInfo_GetContainerProfileStreamRequest.Merge(m, src) +} +func (m *GetContainerProfileStreamRequest) XXX_Size() int { + return xxx_messageInfo_GetContainerProfileStreamRequest.Size(m) +} +func (m *GetContainerProfileStreamRequest) XXX_DiscardUnknown() { + xxx_messageInfo_GetContainerProfileStreamRequest.DiscardUnknown(m) +} + +var xxx_messageInfo_GetContainerProfileStreamRequest proto.InternalMessageInfo + +func (m *GetContainerProfileStreamRequest) GetNamespace() string { + if m != nil { + return m.Namespace + } + return "" +} + +func (m *GetContainerProfileStreamRequest) GetName() string { + if m != nil { + return m.Name + } + return "" +} + +func (m *GetContainerProfileStreamRequest) GetRegion() string { + if m != nil { + return m.Region + } + return "" +} + +func (m *GetContainerProfileStreamRequest) GetCloudAccountIdentifier() string { + if m != nil { + return m.CloudAccountIdentifier + } + return "" +} + +// GetContainerProfileStreamChunk is a single chunk of a streamed +// ContainerProfile download. The first chunk sets metadata + status; +// subsequent chunks set only blob_chunk. +type GetContainerProfileStreamChunk struct { + // metadata is set on the first chunk only. + Metadata *GetContainerProfileStreamChunkMetadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"` + // blob_chunk is a slice of the marshaled ContainerProfile payload. + BlobChunk []byte `protobuf:"bytes,2,opt,name=blob_chunk,json=blobChunk,proto3" json:"blob_chunk,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *GetContainerProfileStreamChunk) Reset() { *m = GetContainerProfileStreamChunk{} } +func (m *GetContainerProfileStreamChunk) String() string { return proto.CompactTextString(m) } +func (*GetContainerProfileStreamChunk) ProtoMessage() {} +func (*GetContainerProfileStreamChunk) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{11} +} +func (m *GetContainerProfileStreamChunk) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_GetContainerProfileStreamChunk.Unmarshal(m, b) +} +func (m *GetContainerProfileStreamChunk) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_GetContainerProfileStreamChunk.Marshal(b, m, deterministic) +} +func (m *GetContainerProfileStreamChunk) XXX_Merge(src proto.Message) { + xxx_messageInfo_GetContainerProfileStreamChunk.Merge(m, src) +} +func (m *GetContainerProfileStreamChunk) XXX_Size() int { + return xxx_messageInfo_GetContainerProfileStreamChunk.Size(m) +} +func (m *GetContainerProfileStreamChunk) XXX_DiscardUnknown() { + xxx_messageInfo_GetContainerProfileStreamChunk.DiscardUnknown(m) +} + +var xxx_messageInfo_GetContainerProfileStreamChunk proto.InternalMessageInfo + +func (m *GetContainerProfileStreamChunk) GetMetadata() *GetContainerProfileStreamChunkMetadata { + if m != nil { + return m.Metadata + } + return nil +} + +func (m *GetContainerProfileStreamChunk) GetBlobChunk() []byte { + if m != nil { + return m.BlobChunk + } + return nil +} + +// GetContainerProfileStreamChunkMetadata is the status header sent on the +// first chunk of a GetContainerProfileStream response. If success is +// false or exists is false the server closes the stream without sending +// blob chunks. +type GetContainerProfileStreamChunkMetadata struct { + Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"` + ErrorMessage string `protobuf:"bytes,2,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"` + ErrorCode ErrorCode `protobuf:"varint,3,opt,name=error_code,json=errorCode,proto3,enum=storageserver.v1.ErrorCode" json:"error_code,omitempty"` + // exists indicates whether a ContainerProfile row was found. + Exists bool `protobuf:"varint,4,opt,name=exists,proto3" json:"exists,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *GetContainerProfileStreamChunkMetadata) Reset() { + *m = GetContainerProfileStreamChunkMetadata{} +} +func (m *GetContainerProfileStreamChunkMetadata) String() string { return proto.CompactTextString(m) } +func (*GetContainerProfileStreamChunkMetadata) ProtoMessage() {} +func (*GetContainerProfileStreamChunkMetadata) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{12} +} +func (m *GetContainerProfileStreamChunkMetadata) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_GetContainerProfileStreamChunkMetadata.Unmarshal(m, b) +} +func (m *GetContainerProfileStreamChunkMetadata) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_GetContainerProfileStreamChunkMetadata.Marshal(b, m, deterministic) +} +func (m *GetContainerProfileStreamChunkMetadata) XXX_Merge(src proto.Message) { + xxx_messageInfo_GetContainerProfileStreamChunkMetadata.Merge(m, src) +} +func (m *GetContainerProfileStreamChunkMetadata) XXX_Size() int { + return xxx_messageInfo_GetContainerProfileStreamChunkMetadata.Size(m) +} +func (m *GetContainerProfileStreamChunkMetadata) XXX_DiscardUnknown() { + xxx_messageInfo_GetContainerProfileStreamChunkMetadata.DiscardUnknown(m) +} + +var xxx_messageInfo_GetContainerProfileStreamChunkMetadata proto.InternalMessageInfo + +func (m *GetContainerProfileStreamChunkMetadata) GetSuccess() bool { + if m != nil { + return m.Success + } + return false +} + +func (m *GetContainerProfileStreamChunkMetadata) GetErrorMessage() string { + if m != nil { + return m.ErrorMessage + } + return "" +} + +func (m *GetContainerProfileStreamChunkMetadata) GetErrorCode() ErrorCode { + if m != nil { + return m.ErrorCode + } + return ErrorCode_ERROR_CODE_UNSPECIFIED +} + +func (m *GetContainerProfileStreamChunkMetadata) GetExists() bool { + if m != nil { + return m.Exists + } + return false +} + +// PutSBOMResponse indicates success or failure of the operation. S3 + +// Postgres persistence happens asynchronously via Pulsar; success here +// means the upload was accepted and the Pulsar message was published. +type PutSBOMResponse struct { + Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"` + ErrorMessage string `protobuf:"bytes,2,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"` + ErrorCode ErrorCode `protobuf:"varint,3,opt,name=error_code,json=errorCode,proto3,enum=storageserver.v1.ErrorCode" json:"error_code,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *PutSBOMResponse) Reset() { *m = PutSBOMResponse{} } +func (m *PutSBOMResponse) String() string { return proto.CompactTextString(m) } +func (*PutSBOMResponse) ProtoMessage() {} +func (*PutSBOMResponse) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{13} +} +func (m *PutSBOMResponse) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_PutSBOMResponse.Unmarshal(m, b) +} +func (m *PutSBOMResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_PutSBOMResponse.Marshal(b, m, deterministic) +} +func (m *PutSBOMResponse) XXX_Merge(src proto.Message) { + xxx_messageInfo_PutSBOMResponse.Merge(m, src) +} +func (m *PutSBOMResponse) XXX_Size() int { + return xxx_messageInfo_PutSBOMResponse.Size(m) +} +func (m *PutSBOMResponse) XXX_DiscardUnknown() { + xxx_messageInfo_PutSBOMResponse.DiscardUnknown(m) +} + +var xxx_messageInfo_PutSBOMResponse proto.InternalMessageInfo + +func (m *PutSBOMResponse) GetSuccess() bool { + if m != nil { + return m.Success + } + return false +} + +func (m *PutSBOMResponse) GetErrorMessage() string { + if m != nil { + return m.ErrorMessage + } + return "" +} + +func (m *PutSBOMResponse) GetErrorCode() ErrorCode { + if m != nil { + return m.ErrorCode + } + return ErrorCode_ERROR_CODE_UNSPECIFIED +} + +// PutSBOMChunk is a single chunk of a streamed SBOM upload. The first +// chunk in the stream MUST set metadata. Subsequent chunks set only +// blob_chunk. Concatenated blob_chunk values form the marshaled SBOMSyft +// proto payload. +type PutSBOMChunk struct { + // metadata is set on the first chunk only. + Metadata *PutSBOMChunkMetadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"` + // blob_chunk is a slice of the marshaled SBOMSyft payload. + BlobChunk []byte `protobuf:"bytes,2,opt,name=blob_chunk,json=blobChunk,proto3" json:"blob_chunk,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *PutSBOMChunk) Reset() { *m = PutSBOMChunk{} } +func (m *PutSBOMChunk) String() string { return proto.CompactTextString(m) } +func (*PutSBOMChunk) ProtoMessage() {} +func (*PutSBOMChunk) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{14} +} +func (m *PutSBOMChunk) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_PutSBOMChunk.Unmarshal(m, b) +} +func (m *PutSBOMChunk) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_PutSBOMChunk.Marshal(b, m, deterministic) +} +func (m *PutSBOMChunk) XXX_Merge(src proto.Message) { + xxx_messageInfo_PutSBOMChunk.Merge(m, src) +} +func (m *PutSBOMChunk) XXX_Size() int { + return xxx_messageInfo_PutSBOMChunk.Size(m) +} +func (m *PutSBOMChunk) XXX_DiscardUnknown() { + xxx_messageInfo_PutSBOMChunk.DiscardUnknown(m) +} + +var xxx_messageInfo_PutSBOMChunk proto.InternalMessageInfo + +func (m *PutSBOMChunk) GetMetadata() *PutSBOMChunkMetadata { + if m != nil { + return m.Metadata + } + return nil +} + +func (m *PutSBOMChunk) GetBlobChunk() []byte { + if m != nil { + return m.BlobChunk + } + return nil +} + +// PutSBOMChunkMetadata is the metadata header sent on the first chunk of a +// PutSBOM call. +type PutSBOMChunkMetadata struct { + ImageDigest string `protobuf:"bytes,1,opt,name=image_digest,json=imageDigest,proto3" json:"image_digest,omitempty"` + SyftVersion string `protobuf:"bytes,2,opt,name=syft_version,json=syftVersion,proto3" json:"syft_version,omitempty"` + Source SBOMSource `protobuf:"varint,3,opt,name=source,proto3,enum=storageserver.v1.SBOMSource" json:"source,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *PutSBOMChunkMetadata) Reset() { *m = PutSBOMChunkMetadata{} } +func (m *PutSBOMChunkMetadata) String() string { return proto.CompactTextString(m) } +func (*PutSBOMChunkMetadata) ProtoMessage() {} +func (*PutSBOMChunkMetadata) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{15} +} +func (m *PutSBOMChunkMetadata) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_PutSBOMChunkMetadata.Unmarshal(m, b) +} +func (m *PutSBOMChunkMetadata) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_PutSBOMChunkMetadata.Marshal(b, m, deterministic) +} +func (m *PutSBOMChunkMetadata) XXX_Merge(src proto.Message) { + xxx_messageInfo_PutSBOMChunkMetadata.Merge(m, src) +} +func (m *PutSBOMChunkMetadata) XXX_Size() int { + return xxx_messageInfo_PutSBOMChunkMetadata.Size(m) +} +func (m *PutSBOMChunkMetadata) XXX_DiscardUnknown() { + xxx_messageInfo_PutSBOMChunkMetadata.DiscardUnknown(m) +} + +var xxx_messageInfo_PutSBOMChunkMetadata proto.InternalMessageInfo + +func (m *PutSBOMChunkMetadata) GetImageDigest() string { + if m != nil { + return m.ImageDigest + } + return "" +} + +func (m *PutSBOMChunkMetadata) GetSyftVersion() string { + if m != nil { + return m.SyftVersion + } + return "" +} + +func (m *PutSBOMChunkMetadata) GetSource() SBOMSource { + if m != nil { + return m.Source + } + return SBOMSource_SBOM_SOURCE_UNSPECIFIED +} + +// GetSBOMRequest probes for or fetches an SBOM by primary key. +// customer_guid is sent via gRPC metadata headers. +type GetSBOMRequest struct { + // image_digest is the SHA-256 digest of the requested image (without the + // "sha256:" prefix). + ImageDigest string `protobuf:"bytes,1,opt,name=image_digest,json=imageDigest,proto3" json:"image_digest,omitempty"` + // syft_version is the syft version that produced the requested SBOM. The + // (image_digest, syft_version) pair is the lookup key. Leave empty to + // request "latest available version" semantics; the server returns the + // most recently created SBOM for the image. + SyftVersion string `protobuf:"bytes,2,opt,name=syft_version,json=syftVersion,proto3" json:"syft_version,omitempty"` + // metadata_only skips the blob and returns only SBOMMetadata. Used by + // agents to decide whether to regenerate. + MetadataOnly bool `protobuf:"varint,3,opt,name=metadata_only,json=metadataOnly,proto3" json:"metadata_only,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *GetSBOMRequest) Reset() { *m = GetSBOMRequest{} } +func (m *GetSBOMRequest) String() string { return proto.CompactTextString(m) } +func (*GetSBOMRequest) ProtoMessage() {} +func (*GetSBOMRequest) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{16} +} +func (m *GetSBOMRequest) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_GetSBOMRequest.Unmarshal(m, b) +} +func (m *GetSBOMRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_GetSBOMRequest.Marshal(b, m, deterministic) +} +func (m *GetSBOMRequest) XXX_Merge(src proto.Message) { + xxx_messageInfo_GetSBOMRequest.Merge(m, src) +} +func (m *GetSBOMRequest) XXX_Size() int { + return xxx_messageInfo_GetSBOMRequest.Size(m) +} +func (m *GetSBOMRequest) XXX_DiscardUnknown() { + xxx_messageInfo_GetSBOMRequest.DiscardUnknown(m) +} + +var xxx_messageInfo_GetSBOMRequest proto.InternalMessageInfo + +func (m *GetSBOMRequest) GetImageDigest() string { + if m != nil { + return m.ImageDigest + } + return "" +} + +func (m *GetSBOMRequest) GetSyftVersion() string { + if m != nil { + return m.SyftVersion + } + return "" +} + +func (m *GetSBOMRequest) GetMetadataOnly() bool { + if m != nil { + return m.MetadataOnly + } + return false +} + +// GetSBOMChunk is a single chunk of a streamed SBOM response. The first +// chunk sets metadata + status; subsequent chunks set only blob_chunk. +type GetSBOMChunk struct { + // metadata is set on the first chunk only. + Metadata *GetSBOMChunkMetadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"` + // blob_chunk is a slice of the marshaled SBOMSyft payload. + BlobChunk []byte `protobuf:"bytes,2,opt,name=blob_chunk,json=blobChunk,proto3" json:"blob_chunk,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *GetSBOMChunk) Reset() { *m = GetSBOMChunk{} } +func (m *GetSBOMChunk) String() string { return proto.CompactTextString(m) } +func (*GetSBOMChunk) ProtoMessage() {} +func (*GetSBOMChunk) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{17} +} +func (m *GetSBOMChunk) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_GetSBOMChunk.Unmarshal(m, b) +} +func (m *GetSBOMChunk) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_GetSBOMChunk.Marshal(b, m, deterministic) +} +func (m *GetSBOMChunk) XXX_Merge(src proto.Message) { + xxx_messageInfo_GetSBOMChunk.Merge(m, src) +} +func (m *GetSBOMChunk) XXX_Size() int { + return xxx_messageInfo_GetSBOMChunk.Size(m) +} +func (m *GetSBOMChunk) XXX_DiscardUnknown() { + xxx_messageInfo_GetSBOMChunk.DiscardUnknown(m) +} + +var xxx_messageInfo_GetSBOMChunk proto.InternalMessageInfo + +func (m *GetSBOMChunk) GetMetadata() *GetSBOMChunkMetadata { + if m != nil { + return m.Metadata + } + return nil +} + +func (m *GetSBOMChunk) GetBlobChunk() []byte { + if m != nil { + return m.BlobChunk + } + return nil +} + +// GetSBOMChunkMetadata is the status header sent on the first chunk of a +// GetSBOM response. If exists=false, success=false, or the request set +// metadata_only=true, the server closes the stream without sending blob +// chunks. +type GetSBOMChunkMetadata struct { + Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"` + ErrorMessage string `protobuf:"bytes,2,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"` + ErrorCode ErrorCode `protobuf:"varint,3,opt,name=error_code,json=errorCode,proto3,enum=storageserver.v1.ErrorCode" json:"error_code,omitempty"` + Exists bool `protobuf:"varint,4,opt,name=exists,proto3" json:"exists,omitempty"` + SbomMetadata *SBOMMetadata `protobuf:"bytes,5,opt,name=sbom_metadata,json=sbomMetadata,proto3" json:"sbom_metadata,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *GetSBOMChunkMetadata) Reset() { *m = GetSBOMChunkMetadata{} } +func (m *GetSBOMChunkMetadata) String() string { return proto.CompactTextString(m) } +func (*GetSBOMChunkMetadata) ProtoMessage() {} +func (*GetSBOMChunkMetadata) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{18} +} +func (m *GetSBOMChunkMetadata) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_GetSBOMChunkMetadata.Unmarshal(m, b) +} +func (m *GetSBOMChunkMetadata) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_GetSBOMChunkMetadata.Marshal(b, m, deterministic) +} +func (m *GetSBOMChunkMetadata) XXX_Merge(src proto.Message) { + xxx_messageInfo_GetSBOMChunkMetadata.Merge(m, src) +} +func (m *GetSBOMChunkMetadata) XXX_Size() int { + return xxx_messageInfo_GetSBOMChunkMetadata.Size(m) +} +func (m *GetSBOMChunkMetadata) XXX_DiscardUnknown() { + xxx_messageInfo_GetSBOMChunkMetadata.DiscardUnknown(m) +} + +var xxx_messageInfo_GetSBOMChunkMetadata proto.InternalMessageInfo + +func (m *GetSBOMChunkMetadata) GetSuccess() bool { + if m != nil { + return m.Success + } + return false +} + +func (m *GetSBOMChunkMetadata) GetErrorMessage() string { + if m != nil { + return m.ErrorMessage + } + return "" +} + +func (m *GetSBOMChunkMetadata) GetErrorCode() ErrorCode { + if m != nil { + return m.ErrorCode + } + return ErrorCode_ERROR_CODE_UNSPECIFIED +} + +func (m *GetSBOMChunkMetadata) GetExists() bool { + if m != nil { + return m.Exists + } + return false +} + +func (m *GetSBOMChunkMetadata) GetSbomMetadata() *SBOMMetadata { + if m != nil { + return m.SbomMetadata + } + return nil +} + +// SBOMMetadata is the indexed view of an SBOM row, returned by GetSBOM +// with or without metadata_only=true. +type SBOMMetadata struct { + // image_digest of the scanned image (without the "sha256:" prefix). + ImageDigest string `protobuf:"bytes,1,opt,name=image_digest,json=imageDigest,proto3" json:"image_digest,omitempty"` + // syft_version that produced the SBOM blob. + SyftVersion string `protobuf:"bytes,2,opt,name=syft_version,json=syftVersion,proto3" json:"syft_version,omitempty"` + // source identifies how this SBOM was produced. + Source SBOMSource `protobuf:"varint,3,opt,name=source,proto3,enum=storageserver.v1.SBOMSource" json:"source,omitempty"` + // blob_size_bytes is the size of the SBOM blob in S3. + BlobSizeBytes int64 `protobuf:"varint,4,opt,name=blob_size_bytes,json=blobSizeBytes,proto3" json:"blob_size_bytes,omitempty"` + // created_at is the RFC3339 timestamp when the row was first inserted. + CreatedAt string `protobuf:"bytes,5,opt,name=created_at,json=createdAt,proto3" json:"created_at,omitempty"` + // last_referenced_at is the RFC3339 timestamp when the orchestrator most + // recently resolved an inventory entry to this SBOM. Drives eviction. + LastReferencedAt string `protobuf:"bytes,6,opt,name=last_referenced_at,json=lastReferencedAt,proto3" json:"last_referenced_at,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *SBOMMetadata) Reset() { *m = SBOMMetadata{} } +func (m *SBOMMetadata) String() string { return proto.CompactTextString(m) } +func (*SBOMMetadata) ProtoMessage() {} +func (*SBOMMetadata) Descriptor() ([]byte, []int) { + return fileDescriptor_3d90829bc66d9c54, []int{19} +} +func (m *SBOMMetadata) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_SBOMMetadata.Unmarshal(m, b) +} +func (m *SBOMMetadata) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_SBOMMetadata.Marshal(b, m, deterministic) +} +func (m *SBOMMetadata) XXX_Merge(src proto.Message) { + xxx_messageInfo_SBOMMetadata.Merge(m, src) +} +func (m *SBOMMetadata) XXX_Size() int { + return xxx_messageInfo_SBOMMetadata.Size(m) +} +func (m *SBOMMetadata) XXX_DiscardUnknown() { + xxx_messageInfo_SBOMMetadata.DiscardUnknown(m) +} + +var xxx_messageInfo_SBOMMetadata proto.InternalMessageInfo + +func (m *SBOMMetadata) GetImageDigest() string { + if m != nil { + return m.ImageDigest + } + return "" +} + +func (m *SBOMMetadata) GetSyftVersion() string { + if m != nil { + return m.SyftVersion + } + return "" +} + +func (m *SBOMMetadata) GetSource() SBOMSource { + if m != nil { + return m.Source + } + return SBOMSource_SBOM_SOURCE_UNSPECIFIED +} + +func (m *SBOMMetadata) GetBlobSizeBytes() int64 { + if m != nil { + return m.BlobSizeBytes + } + return 0 +} + +func (m *SBOMMetadata) GetCreatedAt() string { + if m != nil { + return m.CreatedAt + } + return "" +} + +func (m *SBOMMetadata) GetLastReferencedAt() string { + if m != nil { + return m.LastReferencedAt + } + return "" +} + func init() { proto.RegisterEnum("storageserver.v1.ErrorCode", ErrorCode_name, ErrorCode_value) + proto.RegisterEnum("storageserver.v1.SBOMSource", SBOMSource_name, SBOMSource_value) proto.RegisterType((*SendContainerProfileRequest)(nil), "storageserver.v1.SendContainerProfileRequest") proto.RegisterType((*SendContainerProfileResponse)(nil), "storageserver.v1.SendContainerProfileResponse") proto.RegisterType((*GetProfileRequest)(nil), "storageserver.v1.GetProfileRequest") @@ -642,64 +1400,108 @@ func init() { proto.RegisterType((*ListApplicationProfilesResponse)(nil), "storageserver.v1.ListApplicationProfilesResponse") proto.RegisterType((*ListNetworkNeighborhoodsRequest)(nil), "storageserver.v1.ListNetworkNeighborhoodsRequest") proto.RegisterType((*ListNetworkNeighborhoodsResponse)(nil), "storageserver.v1.ListNetworkNeighborhoodsResponse") + proto.RegisterType((*ContainerProfileChunk)(nil), "storageserver.v1.ContainerProfileChunk") + proto.RegisterType((*ContainerProfileChunkMetadata)(nil), "storageserver.v1.ContainerProfileChunkMetadata") + proto.RegisterType((*GetContainerProfileStreamRequest)(nil), "storageserver.v1.GetContainerProfileStreamRequest") + proto.RegisterType((*GetContainerProfileStreamChunk)(nil), "storageserver.v1.GetContainerProfileStreamChunk") + proto.RegisterType((*GetContainerProfileStreamChunkMetadata)(nil), "storageserver.v1.GetContainerProfileStreamChunkMetadata") + proto.RegisterType((*PutSBOMResponse)(nil), "storageserver.v1.PutSBOMResponse") + proto.RegisterType((*PutSBOMChunk)(nil), "storageserver.v1.PutSBOMChunk") + proto.RegisterType((*PutSBOMChunkMetadata)(nil), "storageserver.v1.PutSBOMChunkMetadata") + proto.RegisterType((*GetSBOMRequest)(nil), "storageserver.v1.GetSBOMRequest") + proto.RegisterType((*GetSBOMChunk)(nil), "storageserver.v1.GetSBOMChunk") + proto.RegisterType((*GetSBOMChunkMetadata)(nil), "storageserver.v1.GetSBOMChunkMetadata") + proto.RegisterType((*SBOMMetadata)(nil), "storageserver.v1.SBOMMetadata") } func init() { proto.RegisterFile("storage_service.proto", fileDescriptor_3d90829bc66d9c54) } var fileDescriptor_3d90829bc66d9c54 = []byte{ - // 856 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x56, 0x41, 0x8f, 0xda, 0x46, - 0x14, 0xae, 0x81, 0x65, 0xcb, 0x4b, 0x1b, 0x39, 0xb3, 0x64, 0xe3, 0xb2, 0xdb, 0x14, 0x91, 0x1e, - 0x56, 0x95, 0x6a, 0x07, 0x7a, 0xa9, 0x7a, 0xa3, 0xe0, 0x4d, 0x91, 0x08, 0x26, 0x03, 0xa4, 0x52, - 0x2e, 0x96, 0xb1, 0xdf, 0xb2, 0x23, 0xc0, 0xe3, 0x7a, 0xcc, 0xe6, 0x56, 0xb5, 0x97, 0xa8, 0x55, - 0x7f, 0x41, 0x6f, 0xbd, 0xf5, 0xd0, 0x7f, 0xd0, 0x1c, 0xfa, 0x67, 0xfa, 0x43, 0x2a, 0xc6, 0x66, - 0x97, 0xc5, 0x06, 0x25, 0x52, 0xb6, 0xab, 0x9c, 0x98, 0xf9, 0xde, 0x9b, 0x37, 0x1f, 0xef, 0x7d, - 0xf3, 0xfc, 0xe0, 0xbe, 0x88, 0x78, 0xe8, 0x4c, 0xd0, 0x16, 0x18, 0x5e, 0x30, 0x17, 0xf5, 0x20, - 0xe4, 0x11, 0x27, 0x6a, 0x02, 0x2f, 0x51, 0x0c, 0xf5, 0x8b, 0x7a, 0xe5, 0xd9, 0x84, 0x45, 0xe7, - 0x8b, 0xb1, 0xee, 0xf2, 0xb9, 0x31, 0x5d, 0x8c, 0x51, 0xb8, 0x4e, 0x80, 0x46, 0xe2, 0x66, 0x04, - 0xd3, 0x89, 0xe1, 0x04, 0x4c, 0x18, 0x82, 0x9f, 0x45, 0x2f, 0x9d, 0x10, 0x5d, 0x3e, 0x0f, 0xb8, - 0x60, 0x11, 0xe3, 0xbe, 0x71, 0x51, 0x1f, 0x63, 0xe4, 0xd4, 0x8d, 0x09, 0xfa, 0x18, 0x3a, 0x11, - 0x7a, 0xf1, 0x25, 0xb5, 0x3f, 0x14, 0x38, 0x1a, 0xa0, 0xef, 0xb5, 0xb8, 0x1f, 0x39, 0xcc, 0xc7, - 0xb0, 0x1f, 0xf2, 0x33, 0x36, 0x43, 0x8a, 0x3f, 0x2c, 0x50, 0x44, 0xe4, 0x27, 0x05, 0xee, 0xb9, - 0x2b, 0x9b, 0x1d, 0xc4, 0x46, 0x4d, 0xa9, 0x2a, 0x27, 0x77, 0x1a, 0x03, 0xfd, 0x8a, 0x8f, 0x7e, - 0xc9, 0x47, 0x4f, 0xf8, 0xe8, 0xc1, 0x74, 0xa2, 0x2f, 0xf9, 0xe8, 0x19, 0x7c, 0xf4, 0x84, 0x8f, - 0x9e, 0xba, 0x57, 0x75, 0x37, 0x90, 0xda, 0xef, 0x0a, 0x1c, 0x67, 0x53, 0x14, 0x01, 0xf7, 0x05, - 0x12, 0x0d, 0xf6, 0xc5, 0xc2, 0x75, 0x51, 0x08, 0x49, 0xec, 0x43, 0xba, 0xda, 0x92, 0x47, 0xf0, - 0x31, 0x86, 0x21, 0x0f, 0xed, 0x39, 0x0a, 0xe1, 0x4c, 0x50, 0xcb, 0x55, 0x95, 0x93, 0x12, 0xfd, - 0x48, 0x82, 0x4f, 0x63, 0x8c, 0x7c, 0x03, 0x10, 0x3b, 0xb9, 0xdc, 0x43, 0x2d, 0x5f, 0x55, 0x4e, - 0xee, 0x36, 0x8e, 0xf4, 0xcd, 0xe4, 0xeb, 0xe6, 0xd2, 0xa7, 0xc5, 0x3d, 0xa4, 0x25, 0x5c, 0x2d, - 0x6b, 0x7f, 0x29, 0x70, 0xef, 0x09, 0x46, 0x1b, 0x49, 0x23, 0x50, 0x98, 0x32, 0xdf, 0x93, 0x6c, - 0x4a, 0x54, 0xae, 0xc9, 0x31, 0x94, 0x7c, 0x67, 0x8e, 0x22, 0x70, 0xdc, 0x15, 0x8d, 0x2b, 0x60, - 0x79, 0x62, 0xb9, 0x91, 0xb7, 0x97, 0xa8, 0x5c, 0x93, 0x43, 0x28, 0x86, 0x38, 0x61, 0xdc, 0xd7, - 0x0a, 0x12, 0x4d, 0x76, 0xe4, 0x6b, 0xd0, 0xdc, 0x19, 0x5f, 0x78, 0xb6, 0xe3, 0xba, 0x7c, 0xe1, - 0x47, 0x36, 0xf3, 0xd0, 0x8f, 0xd8, 0x19, 0xc3, 0x50, 0xdb, 0x93, 0x9e, 0x87, 0xd2, 0xde, 0x8c, - 0xcd, 0x9d, 0x4b, 0x6b, 0xed, 0xcf, 0x02, 0x90, 0x75, 0xb6, 0xb7, 0x9e, 0x3f, 0xf2, 0x4a, 0x81, - 0x03, 0x27, 0x08, 0x66, 0xcc, 0x75, 0x96, 0xb2, 0xb8, 0x14, 0x58, 0x41, 0x0a, 0x6c, 0xf4, 0x0e, - 0x04, 0xd6, 0xbc, 0x8a, 0xbe, 0xfa, 0xdf, 0xc4, 0x49, 0x61, 0xe4, 0x57, 0x05, 0xca, 0x3e, 0x46, - 0x2f, 0x79, 0x38, 0xb5, 0x7d, 0x64, 0x93, 0xf3, 0x31, 0x0f, 0xcf, 0x39, 0xf7, 0x64, 0x46, 0xef, - 0x34, 0x9e, 0xbf, 0x03, 0x26, 0xbd, 0x38, 0x7c, 0x6f, 0x2d, 0x3a, 0x3d, 0xf0, 0xd3, 0xe0, 0x96, - 0x37, 0x57, 0xfc, 0x3f, 0xdf, 0xdc, 0xdf, 0x0a, 0x3c, 0xec, 0x32, 0x11, 0xa5, 0xb3, 0x27, 0x56, - 0x22, 0xbf, 0x26, 0x68, 0x65, 0x53, 0xd0, 0x65, 0xd8, 0x9b, 0xb1, 0x39, 0x8b, 0x64, 0x25, 0xf3, - 0x34, 0xde, 0x2c, 0x65, 0xbe, 0xbc, 0x2a, 0x91, 0xa9, 0x5c, 0xaf, 0xc9, 0xbc, 0xf8, 0xc6, 0x32, - 0xdf, 0xdf, 0x29, 0xf3, 0xd7, 0x39, 0xf8, 0x6c, 0x2b, 0xf9, 0xdb, 0xd7, 0xfc, 0x2f, 0x0a, 0x94, - 0x33, 0x34, 0x2f, 0xb4, 0x42, 0x35, 0x7f, 0x73, 0xa2, 0x3f, 0x48, 0x8b, 0x5e, 0x64, 0xd5, 0xa3, - 0xf6, 0x5a, 0x89, 0xb3, 0x97, 0x21, 0xd7, 0xf7, 0xa0, 0xf6, 0xff, 0xe4, 0xa0, 0xba, 0x9d, 0xfd, - 0xed, 0x17, 0xff, 0x37, 0x05, 0xee, 0x67, 0xf5, 0x99, 0x55, 0xf5, 0x6f, 0xaa, 0xd1, 0x94, 0x33, - 0x1a, 0x4d, 0x66, 0xfd, 0xbf, 0x78, 0x95, 0x83, 0xd2, 0x25, 0x75, 0x52, 0x81, 0x43, 0x93, 0x52, - 0x8b, 0xda, 0x2d, 0xab, 0x6d, 0xda, 0xa3, 0xde, 0xa0, 0x6f, 0xb6, 0x3a, 0xa7, 0x1d, 0xb3, 0xad, - 0x7e, 0x40, 0x1e, 0x42, 0x65, 0xcd, 0xd6, 0xe9, 0x3d, 0x6f, 0x76, 0x3b, 0x6d, 0x9b, 0x9a, 0xcf, - 0x46, 0xe6, 0x60, 0xa8, 0x2a, 0xe4, 0x08, 0x1e, 0x5c, 0x3b, 0xdb, 0x1c, 0x0d, 0xbf, 0xb3, 0x68, - 0xe7, 0x85, 0xd9, 0x56, 0x73, 0xa4, 0x0a, 0xc7, 0x6b, 0xc6, 0x3e, 0xb5, 0x4e, 0x3b, 0x5d, 0xd3, - 0x1e, 0x5a, 0x96, 0xdd, 0x6d, 0xd2, 0x27, 0xa6, 0x9a, 0xdf, 0xe2, 0xd1, 0xb2, 0x9e, 0xf6, 0xbb, - 0xe6, 0xd0, 0x6c, 0xab, 0x85, 0x2d, 0x1e, 0x3d, 0x6b, 0x68, 0x9f, 0x5a, 0xa3, 0x5e, 0x5b, 0xdd, - 0x23, 0x9f, 0xc2, 0x27, 0xd7, 0x28, 0x0e, 0x4d, 0xda, 0x6b, 0x76, 0x6d, 0x89, 0xa9, 0xc5, 0x0d, - 0x86, 0xfd, 0x51, 0x77, 0xd0, 0xa4, 0x89, 0x71, 0xbf, 0xf1, 0x6f, 0x1e, 0xee, 0x0e, 0xe2, 0xbc, - 0x0f, 0xe2, 0xc1, 0x8c, 0x2c, 0xa0, 0x9c, 0x35, 0x89, 0x90, 0x2f, 0xd3, 0xd5, 0xdf, 0x31, 0x54, - 0x55, 0xf4, 0x37, 0x75, 0x4f, 0xf4, 0xfa, 0x3d, 0xc0, 0xd5, 0x67, 0x9b, 0x3c, 0x4a, 0x9f, 0x4e, - 0x8d, 0x20, 0x95, 0xcf, 0x77, 0x3b, 0x25, 0x81, 0x7f, 0x84, 0x07, 0x5b, 0x1a, 0x25, 0x79, 0x9c, - 0x0e, 0xb0, 0xfb, 0x83, 0x50, 0xa9, 0xbf, 0xc5, 0x89, 0xe4, 0xfe, 0x9f, 0x15, 0xd0, 0xb6, 0xbd, - 0x56, 0xb2, 0x25, 0xde, 0x8e, 0xbe, 0x54, 0x69, 0xbc, 0xcd, 0x91, 0x98, 0xc3, 0xb7, 0x8d, 0x17, - 0x8f, 0x33, 0xc7, 0xea, 0xb1, 0xe3, 0x4e, 0xd1, 0xf7, 0xe4, 0x58, 0xed, 0xce, 0x18, 0xfa, 0x91, - 0x71, 0x51, 0x37, 0xe4, 0xd4, 0x3c, 0x2e, 0xca, 0x9f, 0xaf, 0xfe, 0x0b, 0x00, 0x00, 0xff, 0xff, - 0x8b, 0xc1, 0x4a, 0xbf, 0xba, 0x0b, 0x00, 0x00, + // 1372 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x58, 0x4f, 0x6f, 0x1a, 0x47, + 0x14, 0xef, 0x00, 0xc1, 0xe6, 0x19, 0x27, 0x9b, 0x09, 0x71, 0x08, 0x71, 0x1c, 0x42, 0xaa, 0xd4, + 0x8a, 0xda, 0xc5, 0xa6, 0x3d, 0x44, 0xbd, 0x61, 0xd8, 0x38, 0x28, 0x98, 0x75, 0x66, 0x21, 0x51, + 0x73, 0x59, 0x2d, 0xcb, 0x18, 0xaf, 0x80, 0x5d, 0xb2, 0x33, 0x38, 0x75, 0x2a, 0x55, 0xad, 0x22, + 0x55, 0xa9, 0xda, 0x53, 0xab, 0x1e, 0x7a, 0xeb, 0xa9, 0x3d, 0xf4, 0x1b, 0x34, 0x87, 0x5e, 0xfa, + 0x79, 0xfa, 0x15, 0x5a, 0xed, 0x3f, 0x4c, 0x60, 0x17, 0x19, 0x29, 0xa9, 0xd3, 0x13, 0xec, 0xfb, + 0xfb, 0xdb, 0xf7, 0x7e, 0x3b, 0xf3, 0x66, 0xe0, 0x32, 0xe3, 0x96, 0xad, 0x75, 0xa9, 0xca, 0xa8, + 0x7d, 0x64, 0xe8, 0x54, 0x1c, 0xda, 0x16, 0xb7, 0xb0, 0xe0, 0x8b, 0x1d, 0x29, 0xb5, 0xc5, 0xa3, + 0xed, 0xdc, 0xc3, 0xae, 0xc1, 0x0f, 0x47, 0x6d, 0x51, 0xb7, 0x06, 0xc5, 0xde, 0xa8, 0x4d, 0x99, + 0xae, 0x0d, 0x69, 0xd1, 0x37, 0x2b, 0x0e, 0x7b, 0xdd, 0xa2, 0x36, 0x34, 0x58, 0x91, 0x59, 0x07, + 0xfc, 0x99, 0x66, 0x53, 0xdd, 0x1a, 0x0c, 0x2d, 0x66, 0x70, 0xc3, 0x32, 0x8b, 0x47, 0xdb, 0x6d, + 0xca, 0xb5, 0xed, 0x62, 0x97, 0x9a, 0xd4, 0xd6, 0x38, 0xed, 0x78, 0x49, 0x0a, 0xbf, 0x20, 0xb8, + 0xa6, 0x50, 0xb3, 0x53, 0xb1, 0x4c, 0xae, 0x19, 0x26, 0xb5, 0xf7, 0x6d, 0xeb, 0xc0, 0xe8, 0x53, + 0x42, 0x9f, 0x8e, 0x28, 0xe3, 0xf8, 0x2b, 0x04, 0x17, 0xf5, 0x40, 0xa7, 0x0e, 0x3d, 0x65, 0x16, + 0xe5, 0xd1, 0xe6, 0x4a, 0x49, 0x11, 0x4f, 0xf0, 0x88, 0x63, 0x3c, 0xa2, 0x8f, 0x47, 0x1c, 0xf6, + 0xba, 0xa2, 0x83, 0x47, 0x0c, 0xc1, 0x23, 0xfa, 0x78, 0xc4, 0x99, 0xbc, 0x82, 0x3e, 0x25, 0x29, + 0xfc, 0x8c, 0x60, 0x3d, 0x1c, 0x22, 0x1b, 0x5a, 0x26, 0xa3, 0x38, 0x0b, 0x4b, 0x6c, 0xa4, 0xeb, + 0x94, 0x31, 0x17, 0xd8, 0x32, 0x09, 0x1e, 0xf1, 0x2d, 0x58, 0xa5, 0xb6, 0x6d, 0xd9, 0xea, 0x80, + 0x32, 0xa6, 0x75, 0x69, 0x36, 0x96, 0x47, 0x9b, 0x29, 0x92, 0x76, 0x85, 0x7b, 0x9e, 0x0c, 0x7f, + 0x0a, 0xe0, 0x19, 0xe9, 0x56, 0x87, 0x66, 0xe3, 0x79, 0xb4, 0x79, 0xbe, 0x74, 0x4d, 0x9c, 0x2e, + 0xbe, 0x28, 0x39, 0x36, 0x15, 0xab, 0x43, 0x49, 0x8a, 0x06, 0x7f, 0x0b, 0xbf, 0x23, 0xb8, 0xb8, + 0x4b, 0xf9, 0x54, 0xd1, 0x30, 0x24, 0x7a, 0x86, 0xd9, 0x71, 0xd1, 0xa4, 0x88, 0xfb, 0x1f, 0xaf, + 0x43, 0xca, 0xd4, 0x06, 0x94, 0x0d, 0x35, 0x3d, 0x80, 0x71, 0x22, 0x70, 0x3c, 0x9c, 0x07, 0x37, + 0x7b, 0x8a, 0xb8, 0xff, 0xf1, 0x1a, 0x24, 0x6d, 0xda, 0x35, 0x2c, 0x33, 0x9b, 0x70, 0xa5, 0xfe, + 0x13, 0xbe, 0x0b, 0x59, 0xbd, 0x6f, 0x8d, 0x3a, 0xaa, 0xa6, 0xeb, 0xd6, 0xc8, 0xe4, 0xaa, 0xd1, + 0xa1, 0x26, 0x37, 0x0e, 0x0c, 0x6a, 0x67, 0xcf, 0xb9, 0x96, 0x6b, 0xae, 0xbe, 0xec, 0xa9, 0x6b, + 0x63, 0x6d, 0xe1, 0xb7, 0x04, 0xe0, 0x49, 0xb4, 0x67, 0x5e, 0x3f, 0xfc, 0x0d, 0x82, 0x4b, 0xda, + 0x70, 0xd8, 0x37, 0x74, 0xcd, 0xa1, 0xc5, 0x98, 0x60, 0x09, 0x97, 0x60, 0xad, 0x37, 0x40, 0xb0, + 0xf2, 0x49, 0xf4, 0xe0, 0xbd, 0xb1, 0x36, 0x23, 0xc3, 0xdf, 0x22, 0xc8, 0x98, 0x94, 0x3f, 0xb3, + 0xec, 0x9e, 0x6a, 0x52, 0xa3, 0x7b, 0xd8, 0xb6, 0xec, 0x43, 0xcb, 0xea, 0xb8, 0x15, 0x5d, 0x29, + 0x3d, 0x7a, 0x03, 0x48, 0x1a, 0x5e, 0xf8, 0xc6, 0x44, 0x74, 0x72, 0xc9, 0x9c, 0x15, 0x46, 0x7c, + 0x73, 0xc9, 0xff, 0xf2, 0x9b, 0xfb, 0x03, 0xc1, 0x46, 0xdd, 0x60, 0x7c, 0xb6, 0x7a, 0x2c, 0x20, + 0xf9, 0x6b, 0x84, 0x46, 0xd3, 0x84, 0xce, 0xc0, 0xb9, 0xbe, 0x31, 0x30, 0xb8, 0xdb, 0xc9, 0x38, + 0xf1, 0x1e, 0x1c, 0x9a, 0x3b, 0xa9, 0x7c, 0x9a, 0xba, 0xff, 0x27, 0x68, 0x9e, 0x3c, 0x35, 0xcd, + 0x97, 0xe6, 0xd2, 0xfc, 0x55, 0x0c, 0x6e, 0x44, 0x82, 0x3f, 0x7b, 0xce, 0xbf, 0x44, 0x90, 0x09, + 0xe1, 0x3c, 0xcb, 0x26, 0xf2, 0xf1, 0xb7, 0x47, 0xfa, 0x4b, 0xb3, 0xa4, 0x67, 0x61, 0xfd, 0x28, + 0xbc, 0x42, 0x5e, 0xf5, 0x42, 0xe8, 0xfa, 0x3f, 0xe8, 0xfd, 0x9f, 0x31, 0xc8, 0x47, 0xa3, 0x3f, + 0xfb, 0xe6, 0x7f, 0x87, 0xe0, 0x72, 0xd8, 0x3a, 0x13, 0x74, 0xff, 0x6d, 0x2d, 0x34, 0x99, 0x90, + 0x85, 0x26, 0xbc, 0xff, 0x2f, 0x10, 0x5c, 0x9e, 0x5e, 0x21, 0x2a, 0x87, 0x23, 0xb3, 0x87, 0x1f, + 0xc0, 0xf2, 0x80, 0x72, 0xad, 0xa3, 0x71, 0xcd, 0x9f, 0x00, 0x8a, 0xb3, 0x6f, 0x1d, 0xea, 0xba, + 0xe7, 0xbb, 0x91, 0x71, 0x00, 0x7c, 0x1d, 0xa0, 0xdd, 0xb7, 0xda, 0xaa, 0xee, 0xe8, 0xdd, 0x32, + 0xa7, 0x49, 0xca, 0x91, 0xb8, 0x0e, 0x85, 0x1b, 0x70, 0x7d, 0x6e, 0xa4, 0xc2, 0xaf, 0x08, 0xf2, + 0xbb, 0x94, 0x4f, 0x1b, 0x29, 0xdc, 0xa6, 0xda, 0xe0, 0x74, 0x3c, 0x0d, 0x36, 0xdd, 0x58, 0xe8, + 0xa6, 0x1b, 0x3f, 0x35, 0x23, 0x13, 0x73, 0x19, 0xf9, 0x13, 0x82, 0x8d, 0x48, 0xa0, 0x5e, 0x61, + 0x9b, 0x33, 0x85, 0xbd, 0x3b, 0x5b, 0xd8, 0xf9, 0x31, 0x16, 0xaf, 0xf0, 0x2b, 0x04, 0xb7, 0x4f, + 0x17, 0xf3, 0x2c, 0xbf, 0x97, 0x35, 0x48, 0xd2, 0xcf, 0x0d, 0xc6, 0x99, 0x5b, 0xe5, 0x65, 0xe2, + 0x3f, 0x15, 0xbe, 0x47, 0x70, 0x61, 0x7f, 0xc4, 0x95, 0x1d, 0x79, 0xef, 0x5d, 0x98, 0x03, 0x9f, + 0x42, 0xda, 0x47, 0xe3, 0x75, 0x74, 0x67, 0xa6, 0xa3, 0xb7, 0x67, 0x23, 0x4d, 0x7a, 0x2c, 0xde, + 0xbf, 0x1f, 0x10, 0x64, 0xc2, 0x22, 0xe0, 0x9b, 0x90, 0x36, 0x06, 0xce, 0x71, 0xa2, 0x63, 0x74, + 0x29, 0xe3, 0x3e, 0xef, 0x57, 0x5c, 0x59, 0xd5, 0x15, 0x39, 0x26, 0xec, 0xf8, 0x80, 0xab, 0x47, + 0xd4, 0x66, 0x0e, 0xd7, 0xbd, 0x72, 0xac, 0x38, 0xb2, 0x47, 0x9e, 0x08, 0x7f, 0x02, 0x49, 0x66, + 0x8d, 0x6c, 0x3d, 0xa8, 0xc4, 0xfa, 0x2c, 0x7e, 0x27, 0xb5, 0xe2, 0xda, 0x10, 0xdf, 0xb6, 0xf0, + 0x05, 0x9c, 0xdf, 0xa5, 0x7e, 0x57, 0xbc, 0x4f, 0xf0, 0xcd, 0xa0, 0xb9, 0x05, 0xab, 0x41, 0x5d, + 0x54, 0xcb, 0xec, 0x1f, 0xbb, 0xa0, 0x96, 0x49, 0x3a, 0x10, 0xca, 0x66, 0xff, 0xd8, 0x69, 0x82, + 0x9f, 0x7c, 0x81, 0x26, 0x4c, 0x7a, 0x2c, 0xde, 0x84, 0xbf, 0x11, 0x64, 0xc2, 0x22, 0xbc, 0x83, + 0x9f, 0x0c, 0xae, 0xc0, 0x2a, 0x6b, 0x5b, 0x03, 0x75, 0x5c, 0x13, 0x6f, 0xb4, 0xdd, 0x08, 0x6f, + 0xec, 0xb8, 0x16, 0x69, 0xc7, 0x69, 0xbc, 0xec, 0xfe, 0x83, 0x20, 0x3d, 0xa9, 0x3e, 0x4b, 0xb6, + 0xe1, 0xdb, 0x70, 0xc1, 0x6d, 0x0e, 0x33, 0x9e, 0x53, 0xb5, 0x7d, 0xcc, 0x29, 0xf3, 0x47, 0x8e, + 0x55, 0x47, 0xac, 0x18, 0xcf, 0xe9, 0x8e, 0x23, 0x74, 0x9a, 0xa8, 0xdb, 0xd4, 0x39, 0xf5, 0xaa, + 0x5a, 0xb0, 0xd9, 0xa5, 0x7c, 0x49, 0x99, 0xe3, 0x0f, 0x01, 0xf7, 0x35, 0xc6, 0x55, 0x9b, 0x1e, + 0x50, 0x9b, 0x9a, 0xba, 0x67, 0xe6, 0x4d, 0x24, 0x82, 0xa3, 0x21, 0x63, 0x45, 0x99, 0xdf, 0xf9, + 0x2b, 0x06, 0xa9, 0x71, 0xdd, 0x71, 0x0e, 0xd6, 0x24, 0x42, 0x64, 0xa2, 0x56, 0xe4, 0xaa, 0xa4, + 0xb6, 0x1a, 0xca, 0xbe, 0x54, 0xa9, 0xdd, 0xab, 0x49, 0x55, 0xe1, 0x3d, 0xbc, 0x01, 0xb9, 0x09, + 0x5d, 0xad, 0xf1, 0xa8, 0x5c, 0xaf, 0x55, 0x55, 0x22, 0x3d, 0x6c, 0x49, 0x4a, 0x53, 0x40, 0xf8, + 0x1a, 0x5c, 0x79, 0xcd, 0xb7, 0xdc, 0x6a, 0xde, 0x97, 0x49, 0xed, 0x89, 0x54, 0x15, 0x62, 0x38, + 0x0f, 0xeb, 0x13, 0xca, 0x7d, 0x22, 0xdf, 0xab, 0xd5, 0x25, 0xb5, 0x29, 0xcb, 0x6a, 0xbd, 0x4c, + 0x76, 0x25, 0x21, 0x1e, 0x61, 0x51, 0x91, 0xf7, 0xf6, 0xeb, 0x52, 0x53, 0xaa, 0x0a, 0x89, 0x08, + 0x8b, 0x86, 0xdc, 0x54, 0xef, 0xc9, 0xad, 0x46, 0x55, 0x38, 0x87, 0xaf, 0xc3, 0xd5, 0xd7, 0x20, + 0x36, 0x25, 0xd2, 0x28, 0xd7, 0x55, 0x57, 0x26, 0x24, 0xa7, 0x10, 0xee, 0xb7, 0xea, 0x4a, 0x99, + 0xf8, 0xca, 0xa5, 0x29, 0x5f, 0xa7, 0x3d, 0x13, 0xa1, 0x97, 0xc3, 0xd4, 0x27, 0xe8, 0x53, 0x77, + 0x18, 0xc0, 0x49, 0x47, 0x9d, 0x44, 0xae, 0x85, 0x22, 0xb7, 0x48, 0x65, 0xba, 0x8e, 0x59, 0xc8, + 0x4c, 0x2a, 0x1f, 0xcb, 0xe4, 0x41, 0x5d, 0x2e, 0x57, 0x05, 0x34, 0xad, 0x21, 0xd2, 0x6e, 0x4d, + 0x69, 0x92, 0xcf, 0x84, 0x18, 0xce, 0x80, 0x30, 0xa9, 0xb9, 0x2f, 0x2b, 0x4d, 0x21, 0x5e, 0xfa, + 0x71, 0x09, 0xce, 0x2b, 0x1e, 0xb1, 0x14, 0xef, 0xae, 0x05, 0x3f, 0x87, 0x4c, 0xd8, 0xe5, 0x02, + 0xfe, 0x28, 0x84, 0x81, 0xd1, 0xf7, 0x24, 0x39, 0xf1, 0xb4, 0xe6, 0xde, 0x5e, 0x55, 0x88, 0xbf, + 0x8c, 0x21, 0xfc, 0x18, 0xe0, 0xe4, 0x38, 0x8e, 0x6f, 0x85, 0x2e, 0x4e, 0x53, 0x79, 0xde, 0x9f, + 0x6f, 0xe4, 0xef, 0x84, 0x5f, 0xc2, 0x95, 0x88, 0x03, 0x10, 0xde, 0x9a, 0x0d, 0x30, 0xff, 0xa0, + 0x97, 0xdb, 0x5e, 0xc0, 0xc3, 0xcf, 0xff, 0x35, 0x82, 0x6c, 0xd4, 0x14, 0x8e, 0x23, 0xe2, 0xcd, + 0x39, 0x6f, 0xe4, 0x4a, 0x8b, 0xb8, 0xf8, 0x18, 0x9e, 0x41, 0x2e, 0xac, 0x03, 0xde, 0x7c, 0x83, + 0x3f, 0x38, 0xe5, 0xe4, 0xba, 0x68, 0x63, 0x37, 0x11, 0x7e, 0x81, 0xe0, 0x6a, 0xe4, 0x60, 0x85, + 0x4b, 0x0b, 0x4c, 0x76, 0xc1, 0xeb, 0x6f, 0x2d, 0x3a, 0x0d, 0x6e, 0x21, 0xdc, 0x84, 0x55, 0x7f, + 0x3a, 0xf0, 0x13, 0x6f, 0xcc, 0x1f, 0x40, 0x72, 0x37, 0x23, 0xf5, 0x13, 0xef, 0xa6, 0xc0, 0xaa, + 0xbf, 0xdd, 0xf9, 0x51, 0xf3, 0x91, 0x3b, 0x6a, 0x00, 0x7e, 0x63, 0xfe, 0x9e, 0xbb, 0x85, 0x76, + 0x4a, 0x4f, 0xb6, 0x42, 0x2f, 0x36, 0xdb, 0x9a, 0xde, 0xa3, 0x66, 0xc7, 0xbd, 0xd8, 0xd4, 0xfb, + 0x06, 0x35, 0x79, 0xf1, 0x68, 0xbb, 0xe8, 0xde, 0x5b, 0xb6, 0x93, 0xee, 0xcf, 0xc7, 0xff, 0x06, + 0x00, 0x00, 0xff, 0xff, 0x28, 0x46, 0xac, 0x2a, 0x3c, 0x15, 0x00, 0x00, } diff --git a/pkg/client/v1/proto/storage_service.proto b/pkg/client/v1/proto/storage_service.proto index e500d10..df78491 100644 --- a/pkg/client/v1/proto/storage_service.proto +++ b/pkg/client/v1/proto/storage_service.proto @@ -9,11 +9,22 @@ import "github.com/kubescape/storage/pkg/apis/softwarecomposition/v1beta1/genera // StorageService provides gRPC endpoints for node agent to communicate with backend storage service StorageService { // SendContainerProfile receives a container profile (time-series snapshot) from node agent - // and sends it to Pulsar for processing by the ingester - rpc SendContainerProfile(SendContainerProfileRequest) returns (SendContainerProfileResponse); + // and sends it to Pulsar for processing by the ingester. + // + // Deprecated: use SendContainerProfileStream. The unary form is silently + // capped by gRPC's default 4 MiB message limit; container profiles with + // many or large entries (or with MaxContainerProfileSize raised) can + // exceed this and fail on the wire. The streaming variant has no such + // bound. + rpc SendContainerProfile(SendContainerProfileRequest) returns (SendContainerProfileResponse) { + option deprecated = true; + } // GetProfile retrieves an aggregated profile (ApplicationProfile, NetworkNeighborhood, - // or ContainerProfile) by fetching them from S3 + // or ContainerProfile) by fetching them from S3. + // + // Note: for ContainerProfile specifically, prefer GetContainerProfileStream, + // which is not subject to the default 4 MiB unary gRPC message limit. rpc GetProfile(GetProfileRequest) returns (GetProfileResponse); // ListApplicationProfiles lists all ApplicationProfiles in a namespace (returns metadata only, nil Spec) @@ -21,6 +32,49 @@ service StorageService { // ListNetworkNeighborhoods lists all NetworkNeighborhoods in a namespace (returns metadata only, nil Spec) rpc ListNetworkNeighborhoods(ListNetworkNeighborhoodsRequest) returns (ListNetworkNeighborhoodsResponse); + + // SendContainerProfileStream is the streaming replacement for + // SendContainerProfile. Client-streaming so the caller need not know the + // payload size in advance — required because container profiles with + // many or large entries can exceed the default 4 MiB unary gRPC limit. + // + // The first chunk MUST set metadata (which carries identifying labels + // and may be empty otherwise) and MAY include blob_chunk for small + // payloads. Subsequent chunks set only blob_chunk. Concatenated + // blob_chunk values form the marshaled ContainerProfile proto payload. + rpc SendContainerProfileStream(stream ContainerProfileChunk) returns (SendContainerProfileResponse); + + // GetContainerProfileStream is the streaming replacement for + // GetProfile(kind=ContainerProfile). Server-streaming so the caller need + // not know the response size in advance. + // + // The first chunk in the response carries metadata + status. When the + // profile does not exist or success is false, the server closes the + // stream after the metadata chunk. Otherwise subsequent chunks carry + // the marshaled ContainerProfile bytes. + rpc GetContainerProfileStream(GetContainerProfileStreamRequest) returns (stream GetContainerProfileStreamChunk); + + // PutSBOMStream uploads an SBOM produced for an image. Client-streaming + // so the caller need not know the payload size in advance and can stream + // bytes from any io.Reader. Idempotent on (customer_guid, image_digest, + // syft_version) — duplicate calls upsert the metadata row and overwrite + // the S3 blob; end-state is consistent. + // + // The first chunk MUST set metadata (image_digest, syft_version, source) + // and MAY include blob_chunk for small payloads. Subsequent chunks set + // only blob_chunk. Concatenated blob_chunk values form the marshaled + // SBOMSyft proto payload. + rpc PutSBOMStream(stream PutSBOMChunk) returns (PutSBOMResponse); + + // GetSBOMStream probes for or fetches an SBOM by (customer_guid, + // image_digest, syft_version). Server-streaming so the caller need not + // know the response size in advance. + // + // The first chunk in the response carries metadata + status. When the + // request sets metadata_only=true OR the row does not exist OR success + // is false, the server closes the stream after the metadata chunk. + // Otherwise subsequent chunks carry the marshaled SBOMSyft bytes. + rpc GetSBOMStream(GetSBOMRequest) returns (stream GetSBOMChunk); } // SendContainerProfileRequest contains the container profile to be stored @@ -92,6 +146,19 @@ enum ErrorCode { ERROR_CODE_PROFILE_NOT_FOUND = 5; ERROR_CODE_INTERNAL_ERROR = 6; ERROR_CODE_PULSAR_ERROR = 7; + ERROR_CODE_SBOM_NOT_FOUND = 8; + ERROR_CODE_SBOM_TOO_LARGE = 9; +} + +// SBOMSource identifies how an SBOM entered the backend. +enum SBOMSource { + SBOM_SOURCE_UNSPECIFIED = 0; + // SBOM_SOURCE_WORKLOAD: generated by node-agent for an in-cluster workload image. + SBOM_SOURCE_WORKLOAD = 1; + // SBOM_SOURCE_REGISTRY: generated by kubevuln for a registry image (Registry Epic). + SBOM_SOURCE_REGISTRY = 2; + // SBOM_SOURCE_HOST: generated by the host scanner for an EC2 host snapshot. + SBOM_SOURCE_HOST = 3; } // ListApplicationProfilesRequest requests a list of ApplicationProfiles in a namespace @@ -167,3 +234,159 @@ message ListNetworkNeighborhoodsResponse { // Continue token for next page (empty if no more results) string cont = 5; } + +// ContainerProfileChunk is a single chunk of a streamed ContainerProfile +// upload (SendContainerProfileStream). The first chunk in the stream MUST +// set metadata. Subsequent chunks set only blob_chunk. Concatenated +// blob_chunk values form the marshaled ContainerProfile proto payload. +message ContainerProfileChunk { + // metadata is set on the first chunk only. + ContainerProfileChunkMetadata metadata = 1; + + // blob_chunk is a slice of the marshaled ContainerProfile payload. + bytes blob_chunk = 2; +} + +// ContainerProfileChunkMetadata is the metadata header sent on the first +// chunk of a SendContainerProfileStream call. All identifying fields +// (cluster, host_type, host_id) continue to be carried via gRPC metadata +// headers as in the unary path; this message is reserved for any future +// per-call options that the unary form would have put in the request. +message ContainerProfileChunkMetadata { + // reserved for future use; the message exists so the chunked-protocol + // shape matches PutSBOMChunk and clients can begin streaming today + // without an additional API revision when options are added. +} + +// GetContainerProfileStreamRequest mirrors the ContainerProfile-shaped +// fields of GetProfileRequest. customer_guid, cluster, host_type, and +// host_id are sent via gRPC metadata headers. +message GetContainerProfileStreamRequest { + // Namespace of the workload. + string namespace = 1; + + // Name of the container profile to fetch. + string name = 2; + + // Region of the resource (non-k8s scope identifier). + string region = 3; + + // CloudAccountIdentifier of the resource (non-k8s scope identifier). + string cloud_account_identifier = 4; +} + +// GetContainerProfileStreamChunk is a single chunk of a streamed +// ContainerProfile download. The first chunk sets metadata + status; +// subsequent chunks set only blob_chunk. +message GetContainerProfileStreamChunk { + // metadata is set on the first chunk only. + GetContainerProfileStreamChunkMetadata metadata = 1; + + // blob_chunk is a slice of the marshaled ContainerProfile payload. + bytes blob_chunk = 2; +} + +// GetContainerProfileStreamChunkMetadata is the status header sent on the +// first chunk of a GetContainerProfileStream response. If success is +// false or exists is false the server closes the stream without sending +// blob chunks. +message GetContainerProfileStreamChunkMetadata { + bool success = 1; + string error_message = 2; + ErrorCode error_code = 3; + + // exists indicates whether a ContainerProfile row was found. + bool exists = 4; +} + +// PutSBOMResponse indicates success or failure of the operation. S3 + +// Postgres persistence happens asynchronously via Pulsar; success here +// means the upload was accepted and the Pulsar message was published. +message PutSBOMResponse { + bool success = 1; + string error_message = 2; + ErrorCode error_code = 3; +} + +// PutSBOMChunk is a single chunk of a streamed SBOM upload. The first +// chunk in the stream MUST set metadata. Subsequent chunks set only +// blob_chunk. Concatenated blob_chunk values form the marshaled SBOMSyft +// proto payload. +message PutSBOMChunk { + // metadata is set on the first chunk only. + PutSBOMChunkMetadata metadata = 1; + + // blob_chunk is a slice of the marshaled SBOMSyft payload. + bytes blob_chunk = 2; +} + +// PutSBOMChunkMetadata is the metadata header sent on the first chunk of a +// PutSBOM call. +message PutSBOMChunkMetadata { + string image_digest = 1; + string syft_version = 2; + SBOMSource source = 3; +} + +// GetSBOMRequest probes for or fetches an SBOM by primary key. +// customer_guid is sent via gRPC metadata headers. +message GetSBOMRequest { + // image_digest is the SHA-256 digest of the requested image (without the + // "sha256:" prefix). + string image_digest = 1; + + // syft_version is the syft version that produced the requested SBOM. The + // (image_digest, syft_version) pair is the lookup key. Leave empty to + // request "latest available version" semantics; the server returns the + // most recently created SBOM for the image. + string syft_version = 2; + + // metadata_only skips the blob and returns only SBOMMetadata. Used by + // agents to decide whether to regenerate. + bool metadata_only = 3; +} + +// GetSBOMChunk is a single chunk of a streamed SBOM response. The first +// chunk sets metadata + status; subsequent chunks set only blob_chunk. +message GetSBOMChunk { + // metadata is set on the first chunk only. + GetSBOMChunkMetadata metadata = 1; + + // blob_chunk is a slice of the marshaled SBOMSyft payload. + bytes blob_chunk = 2; +} + +// GetSBOMChunkMetadata is the status header sent on the first chunk of a +// GetSBOM response. If exists=false, success=false, or the request set +// metadata_only=true, the server closes the stream without sending blob +// chunks. +message GetSBOMChunkMetadata { + bool success = 1; + string error_message = 2; + ErrorCode error_code = 3; + bool exists = 4; + SBOMMetadata sbom_metadata = 5; +} + +// SBOMMetadata is the indexed view of an SBOM row, returned by GetSBOM +// with or without metadata_only=true. +message SBOMMetadata { + // image_digest of the scanned image (without the "sha256:" prefix). + string image_digest = 1; + + // syft_version that produced the SBOM blob. + string syft_version = 2; + + // source identifies how this SBOM was produced. + SBOMSource source = 3; + + // blob_size_bytes is the size of the SBOM blob in S3. + int64 blob_size_bytes = 4; + + // created_at is the RFC3339 timestamp when the row was first inserted. + string created_at = 5; + + // last_referenced_at is the RFC3339 timestamp when the orchestrator most + // recently resolved an inventory entry to this SBOM. Drives eviction. + string last_referenced_at = 6; +} diff --git a/pkg/client/v1/proto/storage_service_grpc.pb.go b/pkg/client/v1/proto/storage_service_grpc.pb.go index 03c98e4..cc6a00b 100644 --- a/pkg/client/v1/proto/storage_service_grpc.pb.go +++ b/pkg/client/v1/proto/storage_service_grpc.pb.go @@ -19,10 +19,14 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - StorageService_SendContainerProfile_FullMethodName = "/storageserver.v1.StorageService/SendContainerProfile" - StorageService_GetProfile_FullMethodName = "/storageserver.v1.StorageService/GetProfile" - StorageService_ListApplicationProfiles_FullMethodName = "/storageserver.v1.StorageService/ListApplicationProfiles" - StorageService_ListNetworkNeighborhoods_FullMethodName = "/storageserver.v1.StorageService/ListNetworkNeighborhoods" + StorageService_SendContainerProfile_FullMethodName = "/storageserver.v1.StorageService/SendContainerProfile" + StorageService_GetProfile_FullMethodName = "/storageserver.v1.StorageService/GetProfile" + StorageService_ListApplicationProfiles_FullMethodName = "/storageserver.v1.StorageService/ListApplicationProfiles" + StorageService_ListNetworkNeighborhoods_FullMethodName = "/storageserver.v1.StorageService/ListNetworkNeighborhoods" + StorageService_SendContainerProfileStream_FullMethodName = "/storageserver.v1.StorageService/SendContainerProfileStream" + StorageService_GetContainerProfileStream_FullMethodName = "/storageserver.v1.StorageService/GetContainerProfileStream" + StorageService_PutSBOMStream_FullMethodName = "/storageserver.v1.StorageService/PutSBOMStream" + StorageService_GetSBOMStream_FullMethodName = "/storageserver.v1.StorageService/GetSBOMStream" ) // StorageServiceClient is the client API for StorageService service. @@ -31,16 +35,65 @@ const ( // // StorageService provides gRPC endpoints for node agent to communicate with backend storage type StorageServiceClient interface { + // Deprecated: Do not use. // SendContainerProfile receives a container profile (time-series snapshot) from node agent - // and sends it to Pulsar for processing by the ingester + // and sends it to Pulsar for processing by the ingester. + // + // Deprecated: use SendContainerProfileStream. The unary form is silently + // capped by gRPC's default 4 MiB message limit; container profiles with + // many or large entries (or with MaxContainerProfileSize raised) can + // exceed this and fail on the wire. The streaming variant has no such + // bound. SendContainerProfile(ctx context.Context, in *SendContainerProfileRequest, opts ...grpc.CallOption) (*SendContainerProfileResponse, error) - // GetProfile retrieves an aggregated profile (ApplicationProfile or NetworkNeighborhood) - // by fetching container profiles from S3 and aggregating them + // GetProfile retrieves an aggregated profile (ApplicationProfile, NetworkNeighborhood, + // or ContainerProfile) by fetching them from S3. + // + // Note: for ContainerProfile specifically, prefer GetContainerProfileStream, + // which is not subject to the default 4 MiB unary gRPC message limit. GetProfile(ctx context.Context, in *GetProfileRequest, opts ...grpc.CallOption) (*GetProfileResponse, error) // ListApplicationProfiles lists all ApplicationProfiles in a namespace (returns metadata only, nil Spec) ListApplicationProfiles(ctx context.Context, in *ListApplicationProfilesRequest, opts ...grpc.CallOption) (*ListApplicationProfilesResponse, error) // ListNetworkNeighborhoods lists all NetworkNeighborhoods in a namespace (returns metadata only, nil Spec) ListNetworkNeighborhoods(ctx context.Context, in *ListNetworkNeighborhoodsRequest, opts ...grpc.CallOption) (*ListNetworkNeighborhoodsResponse, error) + // SendContainerProfileStream is the streaming replacement for + // SendContainerProfile. Client-streaming so the caller need not know the + // payload size in advance — required because container profiles with + // many or large entries can exceed the default 4 MiB unary gRPC limit. + // + // The first chunk MUST set metadata (which carries identifying labels + // and may be empty otherwise) and MAY include blob_chunk for small + // payloads. Subsequent chunks set only blob_chunk. Concatenated + // blob_chunk values form the marshaled ContainerProfile proto payload. + SendContainerProfileStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[ContainerProfileChunk, SendContainerProfileResponse], error) + // GetContainerProfileStream is the streaming replacement for + // GetProfile(kind=ContainerProfile). Server-streaming so the caller need + // not know the response size in advance. + // + // The first chunk in the response carries metadata + status. When the + // profile does not exist or success is false, the server closes the + // stream after the metadata chunk. Otherwise subsequent chunks carry + // the marshaled ContainerProfile bytes. + GetContainerProfileStream(ctx context.Context, in *GetContainerProfileStreamRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetContainerProfileStreamChunk], error) + // PutSBOMStream uploads an SBOM produced for an image. Client-streaming + // so the caller need not know the payload size in advance and can stream + // bytes from any io.Reader. Idempotent on (customer_guid, image_digest, + // syft_version) — duplicate calls upsert the metadata row and overwrite + // the S3 blob; end-state is consistent. + // + // The first chunk MUST set metadata (image_digest, syft_version, source) + // and MAY include blob_chunk for small payloads. Subsequent chunks set + // only blob_chunk. Concatenated blob_chunk values form the marshaled + // SBOMSyft proto payload. + PutSBOMStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PutSBOMChunk, PutSBOMResponse], error) + // GetSBOMStream probes for or fetches an SBOM by (customer_guid, + // image_digest, syft_version). Server-streaming so the caller need not + // know the response size in advance. + // + // The first chunk in the response carries metadata + status. When the + // request sets metadata_only=true OR the row does not exist OR success + // is false, the server closes the stream after the metadata chunk. + // Otherwise subsequent chunks carry the marshaled SBOMSyft bytes. + GetSBOMStream(ctx context.Context, in *GetSBOMRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetSBOMChunk], error) } type storageServiceClient struct { @@ -51,6 +104,7 @@ func NewStorageServiceClient(cc grpc.ClientConnInterface) StorageServiceClient { return &storageServiceClient{cc} } +// Deprecated: Do not use. func (c *storageServiceClient) SendContainerProfile(ctx context.Context, in *SendContainerProfileRequest, opts ...grpc.CallOption) (*SendContainerProfileResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(SendContainerProfileResponse) @@ -91,22 +145,135 @@ func (c *storageServiceClient) ListNetworkNeighborhoods(ctx context.Context, in return out, nil } +func (c *storageServiceClient) SendContainerProfileStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[ContainerProfileChunk, SendContainerProfileResponse], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &StorageService_ServiceDesc.Streams[0], StorageService_SendContainerProfileStream_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[ContainerProfileChunk, SendContainerProfileResponse]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StorageService_SendContainerProfileStreamClient = grpc.ClientStreamingClient[ContainerProfileChunk, SendContainerProfileResponse] + +func (c *storageServiceClient) GetContainerProfileStream(ctx context.Context, in *GetContainerProfileStreamRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetContainerProfileStreamChunk], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &StorageService_ServiceDesc.Streams[1], StorageService_GetContainerProfileStream_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[GetContainerProfileStreamRequest, GetContainerProfileStreamChunk]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StorageService_GetContainerProfileStreamClient = grpc.ServerStreamingClient[GetContainerProfileStreamChunk] + +func (c *storageServiceClient) PutSBOMStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PutSBOMChunk, PutSBOMResponse], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &StorageService_ServiceDesc.Streams[2], StorageService_PutSBOMStream_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[PutSBOMChunk, PutSBOMResponse]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StorageService_PutSBOMStreamClient = grpc.ClientStreamingClient[PutSBOMChunk, PutSBOMResponse] + +func (c *storageServiceClient) GetSBOMStream(ctx context.Context, in *GetSBOMRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetSBOMChunk], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &StorageService_ServiceDesc.Streams[3], StorageService_GetSBOMStream_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[GetSBOMRequest, GetSBOMChunk]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StorageService_GetSBOMStreamClient = grpc.ServerStreamingClient[GetSBOMChunk] + // StorageServiceServer is the server API for StorageService service. // All implementations must embed UnimplementedStorageServiceServer // for forward compatibility. // // StorageService provides gRPC endpoints for node agent to communicate with backend storage type StorageServiceServer interface { + // Deprecated: Do not use. // SendContainerProfile receives a container profile (time-series snapshot) from node agent - // and sends it to Pulsar for processing by the ingester + // and sends it to Pulsar for processing by the ingester. + // + // Deprecated: use SendContainerProfileStream. The unary form is silently + // capped by gRPC's default 4 MiB message limit; container profiles with + // many or large entries (or with MaxContainerProfileSize raised) can + // exceed this and fail on the wire. The streaming variant has no such + // bound. SendContainerProfile(context.Context, *SendContainerProfileRequest) (*SendContainerProfileResponse, error) - // GetProfile retrieves an aggregated profile (ApplicationProfile or NetworkNeighborhood) - // by fetching container profiles from S3 and aggregating them + // GetProfile retrieves an aggregated profile (ApplicationProfile, NetworkNeighborhood, + // or ContainerProfile) by fetching them from S3. + // + // Note: for ContainerProfile specifically, prefer GetContainerProfileStream, + // which is not subject to the default 4 MiB unary gRPC message limit. GetProfile(context.Context, *GetProfileRequest) (*GetProfileResponse, error) // ListApplicationProfiles lists all ApplicationProfiles in a namespace (returns metadata only, nil Spec) ListApplicationProfiles(context.Context, *ListApplicationProfilesRequest) (*ListApplicationProfilesResponse, error) // ListNetworkNeighborhoods lists all NetworkNeighborhoods in a namespace (returns metadata only, nil Spec) ListNetworkNeighborhoods(context.Context, *ListNetworkNeighborhoodsRequest) (*ListNetworkNeighborhoodsResponse, error) + // SendContainerProfileStream is the streaming replacement for + // SendContainerProfile. Client-streaming so the caller need not know the + // payload size in advance — required because container profiles with + // many or large entries can exceed the default 4 MiB unary gRPC limit. + // + // The first chunk MUST set metadata (which carries identifying labels + // and may be empty otherwise) and MAY include blob_chunk for small + // payloads. Subsequent chunks set only blob_chunk. Concatenated + // blob_chunk values form the marshaled ContainerProfile proto payload. + SendContainerProfileStream(grpc.ClientStreamingServer[ContainerProfileChunk, SendContainerProfileResponse]) error + // GetContainerProfileStream is the streaming replacement for + // GetProfile(kind=ContainerProfile). Server-streaming so the caller need + // not know the response size in advance. + // + // The first chunk in the response carries metadata + status. When the + // profile does not exist or success is false, the server closes the + // stream after the metadata chunk. Otherwise subsequent chunks carry + // the marshaled ContainerProfile bytes. + GetContainerProfileStream(*GetContainerProfileStreamRequest, grpc.ServerStreamingServer[GetContainerProfileStreamChunk]) error + // PutSBOMStream uploads an SBOM produced for an image. Client-streaming + // so the caller need not know the payload size in advance and can stream + // bytes from any io.Reader. Idempotent on (customer_guid, image_digest, + // syft_version) — duplicate calls upsert the metadata row and overwrite + // the S3 blob; end-state is consistent. + // + // The first chunk MUST set metadata (image_digest, syft_version, source) + // and MAY include blob_chunk for small payloads. Subsequent chunks set + // only blob_chunk. Concatenated blob_chunk values form the marshaled + // SBOMSyft proto payload. + PutSBOMStream(grpc.ClientStreamingServer[PutSBOMChunk, PutSBOMResponse]) error + // GetSBOMStream probes for or fetches an SBOM by (customer_guid, + // image_digest, syft_version). Server-streaming so the caller need not + // know the response size in advance. + // + // The first chunk in the response carries metadata + status. When the + // request sets metadata_only=true OR the row does not exist OR success + // is false, the server closes the stream after the metadata chunk. + // Otherwise subsequent chunks carry the marshaled SBOMSyft bytes. + GetSBOMStream(*GetSBOMRequest, grpc.ServerStreamingServer[GetSBOMChunk]) error mustEmbedUnimplementedStorageServiceServer() } @@ -129,6 +296,18 @@ func (UnimplementedStorageServiceServer) ListApplicationProfiles(context.Context func (UnimplementedStorageServiceServer) ListNetworkNeighborhoods(context.Context, *ListNetworkNeighborhoodsRequest) (*ListNetworkNeighborhoodsResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method ListNetworkNeighborhoods not implemented") } +func (UnimplementedStorageServiceServer) SendContainerProfileStream(grpc.ClientStreamingServer[ContainerProfileChunk, SendContainerProfileResponse]) error { + return status.Errorf(codes.Unimplemented, "method SendContainerProfileStream not implemented") +} +func (UnimplementedStorageServiceServer) GetContainerProfileStream(*GetContainerProfileStreamRequest, grpc.ServerStreamingServer[GetContainerProfileStreamChunk]) error { + return status.Errorf(codes.Unimplemented, "method GetContainerProfileStream not implemented") +} +func (UnimplementedStorageServiceServer) PutSBOMStream(grpc.ClientStreamingServer[PutSBOMChunk, PutSBOMResponse]) error { + return status.Errorf(codes.Unimplemented, "method PutSBOMStream not implemented") +} +func (UnimplementedStorageServiceServer) GetSBOMStream(*GetSBOMRequest, grpc.ServerStreamingServer[GetSBOMChunk]) error { + return status.Errorf(codes.Unimplemented, "method GetSBOMStream not implemented") +} func (UnimplementedStorageServiceServer) mustEmbedUnimplementedStorageServiceServer() {} func (UnimplementedStorageServiceServer) testEmbeddedByValue() {} @@ -222,6 +401,42 @@ func _StorageService_ListNetworkNeighborhoods_Handler(srv interface{}, ctx conte return interceptor(ctx, in, info, handler) } +func _StorageService_SendContainerProfileStream_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(StorageServiceServer).SendContainerProfileStream(&grpc.GenericServerStream[ContainerProfileChunk, SendContainerProfileResponse]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StorageService_SendContainerProfileStreamServer = grpc.ClientStreamingServer[ContainerProfileChunk, SendContainerProfileResponse] + +func _StorageService_GetContainerProfileStream_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(GetContainerProfileStreamRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(StorageServiceServer).GetContainerProfileStream(m, &grpc.GenericServerStream[GetContainerProfileStreamRequest, GetContainerProfileStreamChunk]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StorageService_GetContainerProfileStreamServer = grpc.ServerStreamingServer[GetContainerProfileStreamChunk] + +func _StorageService_PutSBOMStream_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(StorageServiceServer).PutSBOMStream(&grpc.GenericServerStream[PutSBOMChunk, PutSBOMResponse]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StorageService_PutSBOMStreamServer = grpc.ClientStreamingServer[PutSBOMChunk, PutSBOMResponse] + +func _StorageService_GetSBOMStream_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(GetSBOMRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(StorageServiceServer).GetSBOMStream(m, &grpc.GenericServerStream[GetSBOMRequest, GetSBOMChunk]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StorageService_GetSBOMStreamServer = grpc.ServerStreamingServer[GetSBOMChunk] + // StorageService_ServiceDesc is the grpc.ServiceDesc for StorageService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -246,6 +461,27 @@ var StorageService_ServiceDesc = grpc.ServiceDesc{ Handler: _StorageService_ListNetworkNeighborhoods_Handler, }, }, - Streams: []grpc.StreamDesc{}, + Streams: []grpc.StreamDesc{ + { + StreamName: "SendContainerProfileStream", + Handler: _StorageService_SendContainerProfileStream_Handler, + ClientStreams: true, + }, + { + StreamName: "GetContainerProfileStream", + Handler: _StorageService_GetContainerProfileStream_Handler, + ServerStreams: true, + }, + { + StreamName: "PutSBOMStream", + Handler: _StorageService_PutSBOMStream_Handler, + ClientStreams: true, + }, + { + StreamName: "GetSBOMStream", + Handler: _StorageService_GetSBOMStream_Handler, + ServerStreams: true, + }, + }, Metadata: "storage_service.proto", } diff --git a/pkg/client/v1/storageclient.go b/pkg/client/v1/storageclient.go index ef3c38f..5bea461 100644 --- a/pkg/client/v1/storageclient.go +++ b/pkg/client/v1/storageclient.go @@ -1,9 +1,11 @@ package v1 import ( + "bytes" "context" "crypto/tls" "fmt" + "io" "net/url" "strconv" "strings" @@ -17,6 +19,11 @@ import ( "google.golang.org/grpc/metadata" ) +// sbomStreamChunkSize is the per-chunk byte budget used by PutSBOMStream +// and GetSBOMStream. Set well below the default 4 MiB gRPC message limit +// to leave headroom for framing overhead. +const sbomStreamChunkSize = 1 << 20 // 1 MiB + // Default gRPC ports const ( DefaultGRPCPort = 50051 // Non-secure gRPC @@ -241,7 +248,12 @@ func (c *StorageClient) withMetadata(ctx context.Context) context.Context { return metadata.NewOutgoingContext(ctx, c.metadata) } -// SendContainerProfile sends a container profile to the storage server +// SendContainerProfile sends a container profile to the storage server. +// +// Deprecated: use SendContainerProfileStream. The unary form is silently +// capped at gRPC's default 4 MiB message size; profiles with many or +// large entries can exceed this and fail on the wire. The streaming +// variant has no such bound. func (c *StorageClient) SendContainerProfile(ctx context.Context, profile *v1beta1.ContainerProfile) (*proto.SendContainerProfileResponse, error) { if c.protoClient == nil { return nil, fmt.Errorf("client is not connected") @@ -262,6 +274,131 @@ func (c *StorageClient) SendContainerProfile(ctx context.Context, profile *v1bet return c.protoClient.SendContainerProfile(ctx, req) } +// SendContainerProfileStream is the streaming replacement for +// SendContainerProfile. The profile is marshaled and sent to the server +// in chunks of sbomStreamChunkSize. Use this whenever you might write a +// large container profile (many entries, long stack traces, long paths). +func (c *StorageClient) SendContainerProfileStream(ctx context.Context, profile *v1beta1.ContainerProfile) (*proto.SendContainerProfileResponse, error) { + if c.protoClient == nil { + return nil, fmt.Errorf("client is not connected") + } + if profile == nil { + return nil, fmt.Errorf("profile is nil") + } + + payload, err := profile.Marshal() + if err != nil { + return nil, fmt.Errorf("failed to marshal ContainerProfile: %w", err) + } + + // Note: we deliberately do NOT apply callTimeout here. Stream duration + // depends on payload size / network speed; the per-call timeout was + // chosen for short unary RPCs. Callers wanting a deadline should pass + // a ctx with their own. + ctx = c.withMetadata(ctx) + + stream, err := c.protoClient.SendContainerProfileStream(ctx) + if err != nil { + return nil, fmt.Errorf("failed to open SendContainerProfileStream: %w", err) + } + + // First chunk MUST set metadata. The metadata message is currently empty + // (reserved for future per-call options); identifying fields continue to + // be carried via gRPC metadata headers. + cut := sbomStreamChunkSize + if cut > len(payload) { + cut = len(payload) + } + first := &proto.ContainerProfileChunk{ + Metadata: &proto.ContainerProfileChunkMetadata{}, + BlobChunk: payload[:cut], + } + if err := stream.Send(first); err != nil { + return nil, fmt.Errorf("failed to send first chunk: %w", err) + } + + for offset := cut; offset < len(payload); offset += sbomStreamChunkSize { + end := offset + sbomStreamChunkSize + if end > len(payload) { + end = len(payload) + } + if err := stream.Send(&proto.ContainerProfileChunk{BlobChunk: payload[offset:end]}); err != nil { + return nil, fmt.Errorf("failed to send chunk: %w", err) + } + } + + return stream.CloseAndRecv() +} + +// GetContainerProfileStream is the streaming replacement for the +// ContainerProfile branch of GetProfile. Use whenever the profile may +// exceed the default 4 MiB unary gRPC message limit. The chunks are +// reassembled and the marshaled bytes are unmarshaled internally — the +// caller receives a typed *v1beta1.ContainerProfile just as with the +// existing GetContainerProfile wrapper. +func (c *StorageClient) GetContainerProfileStream(ctx context.Context, namespace, name string, opts ...ProfileOption) (*v1beta1.ContainerProfile, error) { + if c.protoClient == nil { + return nil, fmt.Errorf("client is not connected") + } + + profileOpts := profileOptionsWithDefaults(opts) + + req := &proto.GetContainerProfileStreamRequest{ + Namespace: namespace, + Name: name, + Region: profileOpts.Region, + CloudAccountIdentifier: profileOpts.CloudAccountIdentifier, + } + + // Note: we deliberately do NOT apply callTimeout here. Stream duration + // depends on payload size / network speed; the per-call timeout was + // chosen for short unary RPCs. Callers wanting a deadline should pass + // a ctx with their own. + ctx = c.withMetadata(ctx) + + stream, err := c.protoClient.GetContainerProfileStream(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to open GetContainerProfileStream: %w", err) + } + + // First chunk MUST carry metadata. + firstChunk, err := stream.Recv() + if err != nil { + return nil, fmt.Errorf("failed to receive first chunk: %w", err) + } + md := firstChunk.GetMetadata() + if md == nil { + return nil, fmt.Errorf("first GetContainerProfileStream chunk missing metadata") + } + if !md.Success { + return nil, fmt.Errorf("failed to get container profile: %s (code: %v)", md.ErrorMessage, md.ErrorCode) + } + if !md.Exists { + return nil, fmt.Errorf("container profile %s/%s not found", namespace, name) + } + + var buf []byte + if len(firstChunk.BlobChunk) > 0 { + buf = append(buf, firstChunk.BlobChunk...) + } + for { + chunk, err := stream.Recv() + if err == io.EOF { + break + } + if err != nil { + return nil, fmt.Errorf("failed to receive chunk: %w", err) + } + buf = append(buf, chunk.BlobChunk...) + } + + profile := &v1beta1.ContainerProfile{} + if err := profile.Unmarshal(buf); err != nil { + return nil, fmt.Errorf("failed to unmarshal ContainerProfile: %w", err) + } + return profile, nil +} + // GetApplicationProfile retrieves an aggregated ApplicationProfile from the storage server // For backward compatibility, region and cloudAccountIdentifier can be provided via ProfileOption // Old way: GetApplicationProfile(ctx, "ns", "name") @@ -340,7 +477,12 @@ func (c *StorageClient) GetNetworkNeighborhood(ctx context.Context, namespace, n return resp.NetworkNeighborhood, nil } -// GetContainerProfile retrieves a ContainerProfile from the storage server +// GetContainerProfile retrieves a ContainerProfile from the storage server. +// +// Deprecated: use GetContainerProfileStream. The unary form goes through +// GetProfile, which is capped at gRPC's default 4 MiB message size; +// profiles with many or large entries can exceed this and fail on the +// wire. The streaming variant has no such bound. func (c *StorageClient) GetContainerProfile(ctx context.Context, namespace, name string, opts ...ProfileOption) (*v1beta1.ContainerProfile, error) { if c.protoClient == nil { return nil, fmt.Errorf("client is not connected") @@ -486,3 +628,210 @@ func (c *StorageClient) ListNetworkNeighborhoods(ctx context.Context, namespace return list, nil } + +// PutSBOMStream uploads an SBOM identified by (image_digest, syft_version, +// source). The payload is the marshaled SBOMSyft proto, read from r and +// sent to the server in chunks of sbomStreamChunkSize. Callers with a +// typed *v1beta1.SBOMSyft should use MarshalSBOM to obtain r. +// +// The underlying RPC is client-streaming; the caller never needs to know +// the payload size in advance. +func (c *StorageClient) PutSBOMStream(ctx context.Context, imageDigest, syftVersion string, source proto.SBOMSource, r io.Reader) (*proto.PutSBOMResponse, error) { + if c.protoClient == nil { + return nil, fmt.Errorf("client is not connected") + } + if r == nil { + return nil, fmt.Errorf("payload reader is nil") + } + + // Note: we deliberately do NOT apply callTimeout here. Stream duration + // depends on payload size / network speed; the per-call timeout was + // chosen for short unary RPCs. Callers wanting a deadline should pass + // a ctx with their own. + ctx = c.withMetadata(ctx) + + stream, err := c.protoClient.PutSBOMStream(ctx) + if err != nil { + return nil, fmt.Errorf("failed to open PutSBOMStream: %w", err) + } + + // First chunk MUST set metadata, and MAY carry the first slice of bytes. + buf := make([]byte, sbomStreamChunkSize) + n, readErr := io.ReadFull(r, buf) + if readErr != nil && readErr != io.EOF && readErr != io.ErrUnexpectedEOF { + return nil, fmt.Errorf("failed to read payload: %w", readErr) + } + first := &proto.PutSBOMChunk{ + Metadata: &proto.PutSBOMChunkMetadata{ + ImageDigest: imageDigest, + SyftVersion: syftVersion, + Source: source, + }, + BlobChunk: buf[:n], + } + if err := stream.Send(first); err != nil { + return nil, fmt.Errorf("failed to send first chunk: %w", err) + } + + // Subsequent chunks carry blob bytes only. ReadFull returns + // io.ErrUnexpectedEOF on a short final read, which is normal — we send + // whatever bytes we got and then stop on the next iteration. + for readErr == nil { + n, readErr = io.ReadFull(r, buf) + if n > 0 { + if err := stream.Send(&proto.PutSBOMChunk{BlobChunk: buf[:n]}); err != nil { + return nil, fmt.Errorf("failed to send chunk: %w", err) + } + } + } + if readErr != io.EOF && readErr != io.ErrUnexpectedEOF { + return nil, fmt.Errorf("failed to read payload: %w", readErr) + } + + return stream.CloseAndRecv() +} + +// GetSBOMStream probes for or fetches an SBOM by (image_digest, syft_version). +// +// Contract: +// - Server-reported failure → returns a non-nil error along with the +// metadata. Caller's `if err != nil { ... }` is enough; no need to +// inspect metadata.Success separately. +// - Row does not exist OR metadataOnly is true → returns (md, nil, nil). +// Caller consults `md.Exists` to distinguish probe-hit from miss. +// - Row exists and metadataOnly is false → returns (md, reader, nil). +// The reader streams the marshaled SBOMSyft proto bytes; use +// UnmarshalSBOM (or read into your own buffer) to reconstruct the +// typed object. The caller MUST Close the reader to release the +// underlying gRPC stream. +// +// The underlying RPC is server-streaming; the caller never needs to know +// the response size in advance. +func (c *StorageClient) GetSBOMStream(ctx context.Context, imageDigest, syftVersion string, metadataOnly bool) (*proto.GetSBOMChunkMetadata, io.ReadCloser, error) { + if c.protoClient == nil { + return nil, nil, fmt.Errorf("client is not connected") + } + + req := &proto.GetSBOMRequest{ + ImageDigest: imageDigest, + SyftVersion: syftVersion, + MetadataOnly: metadataOnly, + } + + ctx = c.withMetadata(ctx) + + // Note: we deliberately do NOT apply callTimeout here, because the + // caller drives the read cadence via the returned io.ReadCloser. A + // per-call timeout would clip large downloads. Callers wanting a + // timeout should pass a ctx with their own deadline. + streamCtx, cancel := context.WithCancel(ctx) + stream, err := c.protoClient.GetSBOMStream(streamCtx, req) + if err != nil { + cancel() + return nil, nil, fmt.Errorf("failed to open GetSBOMStream: %w", err) + } + + // First chunk MUST carry metadata. + firstChunk, err := stream.Recv() + if err != nil { + cancel() + return nil, nil, fmt.Errorf("failed to receive first chunk: %w", err) + } + md := firstChunk.GetMetadata() + if md == nil { + cancel() + return nil, nil, fmt.Errorf("first GetSBOMStream chunk missing metadata") + } + + // A server-side failure is a real error — surface it instead of letting + // callers misread (md, nil, nil) as a clean miss. + if !md.Success { + cancel() + return md, nil, fmt.Errorf("server reported failure: %s (code: %v)", md.ErrorMessage, md.ErrorCode) + } + + // On miss or metadata-only request, the server closes the stream after + // the metadata chunk. Drain and return nil reader. + if !md.Exists || metadataOnly { + cancel() + return md, nil, nil + } + + reader := &sbomStreamReader{ + stream: stream, + pending: firstChunk.BlobChunk, + cancel: cancel, + } + return md, reader, nil +} + +// sbomStreamReader adapts a server-streaming GetSBOM RPC to an +// io.ReadCloser, lazily fetching the next chunk when the previous one is +// drained. +type sbomStreamReader struct { + stream grpc.ServerStreamingClient[proto.GetSBOMChunk] + pending []byte // buffer of bytes from the most-recent chunk not yet read + cancel context.CancelFunc + err error // sticky error; once set, Read returns it on every call +} + +func (r *sbomStreamReader) Read(p []byte) (int, error) { + if r.err != nil { + return 0, r.err + } + for len(r.pending) == 0 { + chunk, err := r.stream.Recv() + if err == io.EOF { + r.err = io.EOF + return 0, io.EOF + } + if err != nil { + r.err = err + return 0, err + } + r.pending = chunk.BlobChunk + } + n := copy(p, r.pending) + r.pending = r.pending[n:] + return n, nil +} + +func (r *sbomStreamReader) Close() error { + if r.cancel != nil { + r.cancel() + r.cancel = nil + } + return nil +} + +// MarshalSBOM marshals a typed SBOMSyft to its proto wire bytes and +// returns a reader over them, ready for PutSBOM. The full marshaled blob +// is held in memory; for very large SBOMs callers may prefer to write +// pre-marshaled bytes to a temp file and pass an *os.File to PutSBOM. +func MarshalSBOM(sbom *v1beta1.SBOMSyft) (io.Reader, error) { + if sbom == nil { + return nil, fmt.Errorf("sbom is nil") + } + b, err := sbom.Marshal() + if err != nil { + return nil, fmt.Errorf("failed to marshal SBOMSyft: %w", err) + } + return bytes.NewReader(b), nil +} + +// UnmarshalSBOM reads marshaled SBOMSyft proto bytes from r and returns +// the decoded object. It reads r until EOF; it does not Close r. +func UnmarshalSBOM(r io.Reader) (*v1beta1.SBOMSyft, error) { + if r == nil { + return nil, fmt.Errorf("reader is nil") + } + b, err := io.ReadAll(r) + if err != nil { + return nil, fmt.Errorf("failed to read SBOM bytes: %w", err) + } + sbom := &v1beta1.SBOMSyft{} + if err := sbom.Unmarshal(b); err != nil { + return nil, fmt.Errorf("failed to unmarshal SBOMSyft: %w", err) + } + return sbom, nil +} diff --git a/pkg/client/v1/storageclient_test.go b/pkg/client/v1/storageclient_test.go index 0952909..34f684e 100644 --- a/pkg/client/v1/storageclient_test.go +++ b/pkg/client/v1/storageclient_test.go @@ -2,6 +2,10 @@ package v1 import ( "context" + "fmt" + "io" + "net" + "sync" "testing" "time" @@ -11,6 +15,9 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/test/bufconn" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) // Mock StorageServiceClient for testing @@ -19,6 +26,10 @@ type mockStorageServiceClient struct { getProfileFunc func(ctx context.Context, in *proto.GetProfileRequest, opts ...grpc.CallOption) (*proto.GetProfileResponse, error) listApplicationProfilesFunc func(ctx context.Context, in *proto.ListApplicationProfilesRequest, opts ...grpc.CallOption) (*proto.ListApplicationProfilesResponse, error) listNetworkNeighborhoodsFunc func(ctx context.Context, in *proto.ListNetworkNeighborhoodsRequest, opts ...grpc.CallOption) (*proto.ListNetworkNeighborhoodsResponse, error) + putSBOMStreamFunc func(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[proto.PutSBOMChunk, proto.PutSBOMResponse], error) + getSBOMStreamFunc func(ctx context.Context, in *proto.GetSBOMRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[proto.GetSBOMChunk], error) + sendContainerProfileStreamFunc func(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[proto.ContainerProfileChunk, proto.SendContainerProfileResponse], error) + getContainerProfileStreamFunc func(ctx context.Context, in *proto.GetContainerProfileStreamRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[proto.GetContainerProfileStreamChunk], error) } func (m *mockStorageServiceClient) SendContainerProfile(ctx context.Context, in *proto.SendContainerProfileRequest, opts ...grpc.CallOption) (*proto.SendContainerProfileResponse, error) { @@ -49,6 +60,34 @@ func (m *mockStorageServiceClient) ListNetworkNeighborhoods(ctx context.Context, return &proto.ListNetworkNeighborhoodsResponse{Success: true}, nil } +func (m *mockStorageServiceClient) PutSBOMStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[proto.PutSBOMChunk, proto.PutSBOMResponse], error) { + if m.putSBOMStreamFunc != nil { + return m.putSBOMStreamFunc(ctx, opts...) + } + return nil, fmt.Errorf("PutSBOMStream not implemented in mock") +} + +func (m *mockStorageServiceClient) GetSBOMStream(ctx context.Context, in *proto.GetSBOMRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[proto.GetSBOMChunk], error) { + if m.getSBOMStreamFunc != nil { + return m.getSBOMStreamFunc(ctx, in, opts...) + } + return nil, fmt.Errorf("GetSBOMStream not implemented in mock") +} + +func (m *mockStorageServiceClient) SendContainerProfileStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[proto.ContainerProfileChunk, proto.SendContainerProfileResponse], error) { + if m.sendContainerProfileStreamFunc != nil { + return m.sendContainerProfileStreamFunc(ctx, opts...) + } + return nil, fmt.Errorf("SendContainerProfileStream not implemented in mock") +} + +func (m *mockStorageServiceClient) GetContainerProfileStream(ctx context.Context, in *proto.GetContainerProfileStreamRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[proto.GetContainerProfileStreamChunk], error) { + if m.getContainerProfileStreamFunc != nil { + return m.getContainerProfileStreamFunc(ctx, in, opts...) + } + return nil, fmt.Errorf("GetContainerProfileStream not implemented in mock") +} + func TestNewStorageClient(t *testing.T) { tests := []struct { name string @@ -615,3 +654,507 @@ func TestParseGRPCURL(t *testing.T) { }) } } + +// storageRoundTripServer is a minimal proto.StorageServiceServer impl that +// records what PutSBOMStream receives and serves what GetSBOMStream should return. +// It exercises the full marshal/unmarshal path of the streaming RPCs +// end-to-end through bufconn — catching wire-shape bugs the prior +// mock-based tests would miss. +type storageRoundTripServer struct { + proto.UnimplementedStorageServiceServer + mu sync.Mutex + + // SBOM upload state: captured from the most recent PutSBOMStream call. + receivedMetadata *proto.PutSBOMChunkMetadata + receivedBytes []byte + + // SBOM download config: returned from the next GetSBOMStream call. + // If serveExists is false the server closes the stream after the + // metadata chunk. + serveExists bool + serveMetadata *proto.SBOMMetadata + serveBytes []byte + serveChunkSize int // 0 → send the whole payload in one chunk + serveChunkDelay time.Duration // sleep before each chunk; for timeout tests + + // ContainerProfile upload state: captured from the most recent + // SendContainerProfileStream call. + cpReceivedBytes []byte + + // ContainerProfile download config: returned from the next + // GetContainerProfileStream call. + cpServeExists bool + cpServeBytes []byte + cpServeChunkSize int +} + +func (s *storageRoundTripServer) PutSBOMStream(stream grpc.ClientStreamingServer[proto.PutSBOMChunk, proto.PutSBOMResponse]) error { + var buf []byte + for { + chunk, err := stream.Recv() + if err == io.EOF { + break + } + if err != nil { + return err + } + s.mu.Lock() + if chunk.Metadata != nil && s.receivedMetadata == nil { + s.receivedMetadata = chunk.Metadata + } + s.mu.Unlock() + buf = append(buf, chunk.BlobChunk...) + } + s.mu.Lock() + s.receivedBytes = buf + s.mu.Unlock() + return stream.SendAndClose(&proto.PutSBOMResponse{Success: true}) +} + +func (s *storageRoundTripServer) GetSBOMStream(req *proto.GetSBOMRequest, stream grpc.ServerStreamingServer[proto.GetSBOMChunk]) error { + s.mu.Lock() + exists := s.serveExists + metadata := s.serveMetadata + payload := s.serveBytes + chunkSize := s.serveChunkSize + chunkDelay := s.serveChunkDelay + s.mu.Unlock() + + // First chunk MUST carry the metadata header. + first := &proto.GetSBOMChunk{ + Metadata: &proto.GetSBOMChunkMetadata{ + Success: true, + Exists: exists, + SbomMetadata: metadata, + }, + } + if err := stream.Send(first); err != nil { + return err + } + + if !exists || req.MetadataOnly { + return nil + } + + if chunkSize <= 0 || chunkSize >= len(payload) { + if chunkDelay > 0 { + time.Sleep(chunkDelay) + } + return stream.Send(&proto.GetSBOMChunk{BlobChunk: payload}) + } + for offset := 0; offset < len(payload); offset += chunkSize { + end := offset + chunkSize + if end > len(payload) { + end = len(payload) + } + if chunkDelay > 0 { + time.Sleep(chunkDelay) + } + if err := stream.Send(&proto.GetSBOMChunk{BlobChunk: payload[offset:end]}); err != nil { + return err + } + } + return nil +} + +func (s *storageRoundTripServer) SendContainerProfileStream(stream grpc.ClientStreamingServer[proto.ContainerProfileChunk, proto.SendContainerProfileResponse]) error { + var buf []byte + for { + chunk, err := stream.Recv() + if err == io.EOF { + break + } + if err != nil { + return err + } + buf = append(buf, chunk.BlobChunk...) + } + s.mu.Lock() + s.cpReceivedBytes = buf + s.mu.Unlock() + return stream.SendAndClose(&proto.SendContainerProfileResponse{Success: true}) +} + +func (s *storageRoundTripServer) GetContainerProfileStream(req *proto.GetContainerProfileStreamRequest, stream grpc.ServerStreamingServer[proto.GetContainerProfileStreamChunk]) error { + s.mu.Lock() + exists := s.cpServeExists + payload := s.cpServeBytes + chunkSize := s.cpServeChunkSize + s.mu.Unlock() + + first := &proto.GetContainerProfileStreamChunk{ + Metadata: &proto.GetContainerProfileStreamChunkMetadata{ + Success: true, + Exists: exists, + }, + } + if err := stream.Send(first); err != nil { + return err + } + if !exists { + return nil + } + if chunkSize <= 0 || chunkSize >= len(payload) { + return stream.Send(&proto.GetContainerProfileStreamChunk{BlobChunk: payload}) + } + for offset := 0; offset < len(payload); offset += chunkSize { + end := offset + chunkSize + if end > len(payload) { + end = len(payload) + } + if err := stream.Send(&proto.GetContainerProfileStreamChunk{BlobChunk: payload[offset:end]}); err != nil { + return err + } + } + return nil +} + +// startBufconnStorageServer starts the round-trip server on an in-memory +// bufconn listener and returns a client connected to it. Optional client +// options (e.g. WithCallTimeout) are forwarded to NewStorageClient. +func startBufconnStorageServer(t *testing.T, opts ...StorageClientOption) (*storageRoundTripServer, *StorageClient, func()) { + t.Helper() + const bufsize = 1024 * 1024 + lis := bufconn.Listen(bufsize) + srv := grpc.NewServer() + rtSrv := &storageRoundTripServer{} + proto.RegisterStorageServiceServer(srv, rtSrv) + go func() { _ = srv.Serve(lis) }() + + conn, err := grpc.NewClient( + "passthrough:///bufnet", + grpc.WithContextDialer(func(_ context.Context, _ string) (net.Conn, error) { return lis.Dial() }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + require.NoError(t, err) + + client, err := NewStorageClient("grpc://example.com:50051", "test-account", "test-key", "test-cluster", opts...) + require.NoError(t, err) + client.conn = conn + client.protoClient = proto.NewStorageServiceClient(conn) + + cleanup := func() { + _ = conn.Close() + srv.Stop() + } + return rtSrv, client, cleanup +} + +// sampleSBOMSyft constructs a non-trivial SBOMSyft for round-trip tests +// so the proto Marshal / Unmarshal path actually has fields to round-trip. +func sampleSBOMSyft() *v1beta1.SBOMSyft { + return &v1beta1.SBOMSyft{ + TypeMeta: metav1.TypeMeta{ + Kind: "SBOMSyft", + APIVersion: "spdx.softwarecomposition.kubescape.io/v1beta1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: "sha256-abc123", + Namespace: "kubescape", + Annotations: map[string]string{ + "image.name": "library/nginx:latest", + "syft.version": "1.0.0", + }, + }, + Spec: v1beta1.SBOMSyftSpec{ + Metadata: v1beta1.SPDXMeta{ + Tool: v1beta1.ToolMeta{ + Name: "syft", + Version: "1.0.0", + }, + }, + Syft: v1beta1.SyftDocument{ + SyftSource: v1beta1.SyftSource{Type: "image"}, + }, + }, + } +} + +// TestStorageClient_SBOMRoundTrip is the end-to-end marshal/unmarshal test +// matthyx asked for: it stands up a real gRPC server over bufconn, ships +// a non-trivial SBOMSyft via the client's PutSBOM, then pulls it back via +// GetSBOM, and verifies semantic equality at every hop. Covers four +// shapes: +// - Probe (metadata_only=true), exists path +// - Probe (metadata_only=true), miss path +// - Full fetch, payload fits in one chunk +// - Full fetch, payload is split across many small chunks (forces the +// reader to drain across multiple Recv calls) +func TestStorageClient_SBOMRoundTrip(t *testing.T) { + const ( + imageDigest = "abc123def456" + syftVersion = "1.0.0" + ) + source := proto.SBOMSource_SBOM_SOURCE_WORKLOAD + + t.Run("PutSBOM marshals through MarshalSBOM and arrives intact", func(t *testing.T) { + rtSrv, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + original := sampleSBOMSyft() + reader, err := MarshalSBOM(original) + require.NoError(t, err) + + resp, err := client.PutSBOMStream(context.Background(), imageDigest, syftVersion, source, reader) + require.NoError(t, err) + assert.True(t, resp.Success) + + // Server received the right metadata header. + require.NotNil(t, rtSrv.receivedMetadata) + assert.Equal(t, imageDigest, rtSrv.receivedMetadata.ImageDigest) + assert.Equal(t, syftVersion, rtSrv.receivedMetadata.SyftVersion) + assert.Equal(t, source, rtSrv.receivedMetadata.Source) + + // Server-received bytes unmarshal to a SBOMSyft equal to the original. + require.NotEmpty(t, rtSrv.receivedBytes) + got := &v1beta1.SBOMSyft{} + require.NoError(t, got.Unmarshal(rtSrv.receivedBytes)) + assert.Equal(t, original.Name, got.Name) + assert.Equal(t, original.Namespace, got.Namespace) + assert.Equal(t, original.Spec.Metadata.Tool.Name, got.Spec.Metadata.Tool.Name) + assert.Equal(t, original.Spec.Metadata.Tool.Version, got.Spec.Metadata.Tool.Version) + assert.Equal(t, original.Annotations["image.name"], got.Annotations["image.name"]) + }) + + t.Run("GetSBOM metadata-only probe returns only metadata", func(t *testing.T) { + rtSrv, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + rtSrv.serveExists = true + rtSrv.serveMetadata = &proto.SBOMMetadata{ + ImageDigest: imageDigest, + SyftVersion: syftVersion, + } + + md, reader, err := client.GetSBOMStream(context.Background(), imageDigest, syftVersion, true) + require.NoError(t, err) + assert.True(t, md.Success) + assert.True(t, md.Exists) + require.NotNil(t, md.SbomMetadata) + assert.Equal(t, imageDigest, md.SbomMetadata.ImageDigest) + assert.Nil(t, reader, "metadata-only probe must not return a reader") + }) + + t.Run("GetSBOM probe miss returns exists=false and no reader", func(t *testing.T) { + rtSrv, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + rtSrv.serveExists = false + + md, reader, err := client.GetSBOMStream(context.Background(), imageDigest, syftVersion, true) + require.NoError(t, err) + assert.True(t, md.Success) + assert.False(t, md.Exists) + assert.Nil(t, reader) + }) + + t.Run("GetSBOM full fetch round-trips through UnmarshalSBOM (single chunk)", func(t *testing.T) { + rtSrv, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + original := sampleSBOMSyft() + payload, err := original.Marshal() + require.NoError(t, err) + rtSrv.serveExists = true + rtSrv.serveMetadata = &proto.SBOMMetadata{ImageDigest: imageDigest, SyftVersion: syftVersion} + rtSrv.serveBytes = payload + rtSrv.serveChunkSize = 0 // single chunk + + md, reader, err := client.GetSBOMStream(context.Background(), imageDigest, syftVersion, false) + require.NoError(t, err) + require.NotNil(t, reader) + defer reader.Close() + assert.True(t, md.Exists) + + got, err := UnmarshalSBOM(reader) + require.NoError(t, err) + assert.Equal(t, original.Name, got.Name) + assert.Equal(t, original.Spec.Metadata.Tool.Name, got.Spec.Metadata.Tool.Name) + }) + + t.Run("GetSBOM full fetch round-trips with payload split across many chunks", func(t *testing.T) { + rtSrv, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + original := sampleSBOMSyft() + payload, err := original.Marshal() + require.NoError(t, err) + rtSrv.serveExists = true + rtSrv.serveMetadata = &proto.SBOMMetadata{ImageDigest: imageDigest, SyftVersion: syftVersion} + rtSrv.serveBytes = payload + // 7-byte chunks force the reader to drain over many Recv calls and + // across reads smaller than a chunk — exercises the buffering logic. + rtSrv.serveChunkSize = 7 + + md, reader, err := client.GetSBOMStream(context.Background(), imageDigest, syftVersion, false) + require.NoError(t, err) + require.NotNil(t, reader) + defer reader.Close() + assert.True(t, md.Exists) + + // Read in 13-byte sips to also stress the partial-Read consumer path. + var collected []byte + sip := make([]byte, 13) + for { + n, err := reader.Read(sip) + collected = append(collected, sip[:n]...) + if err == io.EOF { + break + } + require.NoError(t, err) + } + assert.Equal(t, payload, collected, "bytes received over many chunks must match the originally marshaled payload") + + got := &v1beta1.SBOMSyft{} + require.NoError(t, got.Unmarshal(collected)) + assert.Equal(t, original.Name, got.Name) + }) +} + +// TestStorageClient_PutSBOM_NilReader and the next test cover the trivial +// argument-validation paths that the bufconn round-trip doesn't exercise. +func TestStorageClient_PutSBOM_NilReader(t *testing.T) { + _, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + _, err := client.PutSBOMStream(context.Background(), "abc", "1.0.0", proto.SBOMSource_SBOM_SOURCE_WORKLOAD, nil) + require.Error(t, err) + assert.Contains(t, err.Error(), "nil") +} + +func TestMarshalSBOM_NilInput(t *testing.T) { + _, err := MarshalSBOM(nil) + require.Error(t, err) +} + +func TestUnmarshalSBOM_NilReader(t *testing.T) { + _, err := UnmarshalSBOM(nil) + require.Error(t, err) +} + +// sampleContainerProfile is the CP counterpart of sampleSBOMSyft: a +// non-trivial ContainerProfile used to verify that the streaming RPCs +// marshal and unmarshal fields correctly end-to-end. +func sampleContainerProfile() *v1beta1.ContainerProfile { + return &v1beta1.ContainerProfile{ + TypeMeta: metav1.TypeMeta{ + Kind: "ContainerProfile", + APIVersion: "spdx.softwarecomposition.kubescape.io/v1beta1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: "nginx-7d4f8b9c-abc12", + Namespace: "default", + Labels: map[string]string{ + "app.kubernetes.io/name": "nginx", + }, + }, + } +} + +// TestStorageClient_ContainerProfileStreamRoundTrip mirrors the SBOM +// round-trip test for the new CP streaming RPCs. Stands up a real gRPC +// server over bufconn and verifies the upload + download paths. +func TestStorageClient_ContainerProfileStreamRoundTrip(t *testing.T) { + t.Run("SendContainerProfileStream marshals and arrives intact", func(t *testing.T) { + rtSrv, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + original := sampleContainerProfile() + resp, err := client.SendContainerProfileStream(context.Background(), original) + require.NoError(t, err) + assert.True(t, resp.Success) + + require.NotEmpty(t, rtSrv.cpReceivedBytes) + got := &v1beta1.ContainerProfile{} + require.NoError(t, got.Unmarshal(rtSrv.cpReceivedBytes)) + assert.Equal(t, original.Name, got.Name) + assert.Equal(t, original.Namespace, got.Namespace) + assert.Equal(t, original.Labels["app.kubernetes.io/name"], got.Labels["app.kubernetes.io/name"]) + }) + + t.Run("GetContainerProfileStream round-trips with payload split across chunks", func(t *testing.T) { + rtSrv, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + original := sampleContainerProfile() + payload, err := original.Marshal() + require.NoError(t, err) + rtSrv.cpServeExists = true + rtSrv.cpServeBytes = payload + rtSrv.cpServeChunkSize = 7 // small chunks stress the reassembly loop + + got, err := client.GetContainerProfileStream(context.Background(), original.Namespace, original.Name) + require.NoError(t, err) + assert.Equal(t, original.Name, got.Name) + assert.Equal(t, original.Namespace, got.Namespace) + assert.Equal(t, original.Labels["app.kubernetes.io/name"], got.Labels["app.kubernetes.io/name"]) + }) + + t.Run("GetContainerProfileStream returns not-found error when row is absent", func(t *testing.T) { + _, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + _, err := client.GetContainerProfileStream(context.Background(), "default", "missing") + require.Error(t, err) + assert.Contains(t, err.Error(), "not found") + }) +} + +func TestStorageClient_SendContainerProfileStream_NilProfile(t *testing.T) { + _, client, cleanup := startBufconnStorageServer(t) + defer cleanup() + + _, err := client.SendContainerProfileStream(context.Background(), nil) + require.Error(t, err) + assert.Contains(t, err.Error(), "nil") +} + +// TestStorageClient_GetSBOMStream_NotBoundedByCallTimeout proves that the +// streaming RPC is NOT clipped by WithCallTimeout — a regression matthyx +// flagged. We configure a very short callTimeout (well under any +// production stream latency) and have the server delay between chunks +// by an order of magnitude longer. If the stream were wrapped in +// context.WithTimeout(callTimeout) the call would fail with +// "context deadline exceeded"; success here demonstrates the timeout is +// only applied to unary RPCs. +// +// Test stays under 500ms wall time so it's safe in CI. +func TestStorageClient_GetSBOMStream_NotBoundedByCallTimeout(t *testing.T) { + const ( + shortCallTimeout = 20 * time.Millisecond + serverChunkDelay = 50 * time.Millisecond + chunks = 5 + ) + // Total stream time on the server ≈ chunks * serverChunkDelay = 250ms, + // which is 12× the configured callTimeout. A unary timeout would have + // fired within the first 20ms. + + rtSrv, client, cleanup := startBufconnStorageServer(t, WithCallTimeout(shortCallTimeout)) + defer cleanup() + + original := sampleSBOMSyft() + payload, err := original.Marshal() + require.NoError(t, err) + require.GreaterOrEqual(t, len(payload), chunks, "payload must have enough bytes to split") + + rtSrv.serveExists = true + rtSrv.serveMetadata = &proto.SBOMMetadata{ImageDigest: "abc", SyftVersion: "1.0.0"} + rtSrv.serveBytes = payload + rtSrv.serveChunkSize = len(payload) / chunks + rtSrv.serveChunkDelay = serverChunkDelay + + start := time.Now() + md, reader, err := client.GetSBOMStream(context.Background(), "abc", "1.0.0", false) + require.NoError(t, err, "stream must not be clipped by callTimeout") + require.NotNil(t, reader) + defer reader.Close() + assert.True(t, md.Exists) + + got, err := UnmarshalSBOM(reader) + require.NoError(t, err) + assert.Equal(t, original.Name, got.Name) + + elapsed := time.Since(start) + assert.Greater(t, elapsed, shortCallTimeout, + "sanity: the stream genuinely ran longer than callTimeout (otherwise the test proves nothing)") +}