feat: Support publishing new log entries to Pub/Sub topics

Dockerfile.pubsub-emulator

@@ -0,0 +1,3 @@
+# gcloud sdk for pubsub emulator with netcat added for the startup health check
+FROM google/cloud-sdk:443.0.0@sha256:a59335d227a98b41ecdf6ff3d69923b8aef0703694e58992ecb75c7147140d37
+RUN apt-get install -y netcat

cmd/rekor-server/app/root.go

@@ -93,6 +93,10 @@ func init() {
 Memory and file-based signers should only be used for testing.`)
  rootCmd.PersistentFlags().String("rekor_server.signer-passwd", "", "Password to decrypt signer private key")
 
+ rootCmd.PersistentFlags().String("rekor_server.new_entry_publisher", "", "URL for pub/sub queue to send messages to when new entries are added to the log. Ignored if not set. Supported providers: [gcppubsub]")
+ rootCmd.PersistentFlags().Bool("rekor_server.publish_events_protobuf", false, "Whether to publish events in Protobuf wire format. Applies to all enabled event types.")
+ rootCmd.PersistentFlags().Bool("rekor_server.publish_events_json", false, "Whether to publish events in CloudEvents JSON format. Applies to all enabled event types.")
+
  rootCmd.PersistentFlags().Uint16("port", 3000, "Port to bind to")
 
  rootCmd.PersistentFlags().Bool("enable_retrieve_api", true, "enables Redis-based index API endpoint")

cmd/rekor-server/app/serve.go

@@ -61,7 +61,6 @@ var serveCmd = &cobra.Command{
  Short: "start http server with configured api",
  Long: `Starts a http server and serves the configured api`,
  Run: func(cmd *cobra.Command, args []string) {
-
  // Setup the logger to dev/prod
  log.ConfigureLogger(viper.GetString("log_type"))
 
@@ -83,7 +82,6 @@ var serveCmd = &cobra.Command{
  log.Logger.Error(err)
  }
  }()
-
  //TODO: make this a config option for server to load via viper field
  //TODO: add command line option to print versions supported in binary
 
@@ -101,7 +99,6 @@ var serveCmd = &cobra.Command{
  hashedrekord.KIND: {hashedrekord_v001.APIVERSION},
  dsse.KIND: {dsse_v001.APIVERSION},
  }
-
  for k, v := range pluggableTypeMap {
  log.Logger.Infof("Loading support for pluggable type '%v'", k)
  log.Logger.Infof("Loading version '%v' for pluggable type '%v'", v, k)

docker-compose.test.yml

@@ -20,7 +20,8 @@ services:
  context: .
  target: "test"
  environment:
- - TMPDIR=/var/run/attestations # workaround for https://github.com/google/go-cloud/issues/3294
+ TMPDIR: /var/run/attestations # workaround for https://github.com/google/go-cloud/issues/3294
+ PUBSUB_EMULATOR_HOST: gcp-pubsub-emulator:8085
  command: [
  "rekor-server",
  "-test.coverprofile=rekor-server.cov",
@@ -34,7 +35,29 @@ services:
  "--enable_attestation_storage",
  "--attestation_storage_bucket=file:///var/run/attestations",
  "--max_request_body_size=32792576",
+ "--rekor_server.new_entry_publisher=gcppubsub://projects/test-project/topics/new-entry",
+ "--rekor_server.publish_events_json=true",
  ]
  ports:
  - "3000:3000"
  - "2112:2112"
+ depends_on:
+ - gcp-pubsub-emulator
+ gcp-pubsub-emulator:
+ image: gcp-pubsub-emulator
+ ports:
+ - "8085:8085"
+ command:
+ - gcloud
+ - beta
+ - emulators
+ - pubsub
+ - start
+ - --host-port=0.0.0.0:8085
+ - --project=test-project
+ healthcheck:
+ test: ["CMD", "nc", "-zv", "localhost", "8085"]
+ interval: 10s
+ timeout: 3s
+ retries: 3
+ start_period: 10s

docker-compose.yml

@@ -114,4 +114,3 @@ services:
  timeout: 3s
  retries: 3
  start_period: 5s
-

go.mod

@@ -51,6 +51,7 @@ require (
 )
 
 require (
+ cloud.google.com/go/pubsub v1.33.0
  github.com/AdamKorcz/go-fuzz-headers-1 v0.0.0-20230618160516-e936619f9f18
  github.com/cyberphone/json-canonicalization v0.0.0-20220623050100-57a0ce2678a7
  github.com/go-redis/redismock/v9 v9.0.3
@@ -183,7 +184,7 @@ require (
  golang.org/x/term v0.11.0 // indirect
  golang.org/x/text v0.12.0 // indirect
  golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
- google.golang.org/api v0.135.0 // indirect
+ google.golang.org/api v0.135.0
  google.golang.org/appengine v1.6.7 // indirect
  gopkg.in/square/go-jose.v2 v2.6.0 // indirect
  gopkg.in/yaml.v2 v2.4.0

go.sum

@@ -39,6 +39,8 @@ cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2k
 cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
 cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
 cloud.google.com/go/pubsub v1.3.1/go.mod h1:i+ucay31+CNRpDW4Lu78I4xXG+O1r/MAHgjpRVR+TSU=
+cloud.google.com/go/pubsub v1.33.0 h1:6SPCPvWav64tj0sVX/+npCBKhUi/UjJehy9op/V3p2g=
+cloud.google.com/go/pubsub v1.33.0/go.mod h1:f+w71I33OMyxf9VpMVcZbnG5KSUkCOUHYpFd5U1GdRc=
 cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
 cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0ZeosJ0Rtdos=
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=

pkg/api/api.go

@@ -31,6 +31,7 @@ import (
  "google.golang.org/grpc/credentials/insecure"
 
  "github.com/sigstore/rekor/pkg/log"
+ "github.com/sigstore/rekor/pkg/pubsub"
  "github.com/sigstore/rekor/pkg/sharding"
  "github.com/sigstore/rekor/pkg/signer"
  "github.com/sigstore/rekor/pkg/storage"
@@ -39,6 +40,8 @@ import (
  "github.com/sigstore/sigstore/pkg/cryptoutils"
  "github.com/sigstore/sigstore/pkg/signature"
  "github.com/sigstore/sigstore/pkg/signature/options"
+
+ _ "github.com/sigstore/rekor/pkg/pubsub/gcp" // Load GCP pubsub implementation
 )
 
 func dial(ctx context.Context, rpcServer string) (*grpc.ClientConn, error) {
@@ -63,6 +66,9 @@ type API struct {
  signer signature.Signer
  // stops checkpoint publishing
  checkpointPublishCancel context.CancelFunc
+ // Publishes notifications when new entries are added to the log. May be
+ // nil if no publisher is configured.
+ newEntryPublisher pubsub.Publisher
 }
 
 func NewAPI(treeID uint) (*API, error) {
@@ -112,6 +118,18 @@ func NewAPI(treeID uint) (*API, error) {
 
  pubkey := cryptoutils.PEMEncode(cryptoutils.PublicKeyPEMType, b)
 
+ var newEntryPublisher pubsub.Publisher
+ if p := viper.GetString("rekor_server.new_entry_publisher"); p != "" {
+ if !viper.GetBool("rekor_server.publish_events_protobuf") && !viper.GetBool("rekor_server.publish_events_json") {
+ return nil, fmt.Errorf("%q is configured but neither %q or %q are enabled", "new_entry_publisher", "publish_events_protobuf", "publish_events_json")
+ }
+ newEntryPublisher, err = pubsub.Get(ctx, p)
+ if err != nil {
+ return nil, fmt.Errorf("init event publisher: %w", err)
+ }
+ log.ContextLogger(ctx).Infof("Initialized new entry event publisher: %s", p)
+ }
+
  return &API{
  // Transparency Log Stuff
  logClient: logClient,
@@ -121,6 +139,8 @@ func NewAPI(treeID uint) (*API, error) {
  pubkey: string(pubkey),
  pubkeyHash: hex.EncodeToString(pubkeyHashBytes[:]),
  signer: rekorSigner,
+ // Utility functionality not required for operation of the core service
+ newEntryPublisher: newEntryPublisher,
  }, nil
 }
 
@@ -166,4 +186,8 @@ func ConfigureAPI(treeID uint) {
 
 func StopAPI() {
  api.checkpointPublishCancel()
+
+ if api.newEntryPublisher != nil {
+ api.newEntryPublisher.Close()
+ }
 }

pkg/api/entries.go

@@ -36,10 +36,14 @@ import (
  "google.golang.org/genproto/googleapis/rpc/code"
  "google.golang.org/grpc/codes"
 
+ "github.com/sigstore/rekor/pkg/events"
+ "github.com/sigstore/rekor/pkg/events/newentry"
  "github.com/sigstore/rekor/pkg/generated/models"
  "github.com/sigstore/rekor/pkg/generated/restapi/operations/entries"
  "github.com/sigstore/rekor/pkg/log"
+ "github.com/sigstore/rekor/pkg/pubsub"
  "github.com/sigstore/rekor/pkg/sharding"
+ "github.com/sigstore/rekor/pkg/tle"
  "github.com/sigstore/rekor/pkg/trillianclient"
  "github.com/sigstore/rekor/pkg/types"
  "github.com/sigstore/rekor/pkg/util"
@@ -290,7 +294,7 @@ func createLogEntry(params entries.CreateLogEntryParams) (models.LogEntry, middl
  RootHash: swag.String(hex.EncodeToString(root.RootHash)),
  LogIndex: swag.Int64(queuedLeaf.LeafIndex),
  Hashes: hashes,
- Checkpoint: stringPointer(string(scBytes)),
+ Checkpoint: swag.String(string(scBytes)),
  }
 
  logEntryAnon.Verification = &models.LogEntryAnonVerification{
@@ -301,9 +305,66 @@ func createLogEntry(params entries.CreateLogEntryParams) (models.LogEntry, middl
  logEntry := models.LogEntry{
  entryID: logEntryAnon,
  }
+
+ if api.newEntryPublisher != nil {
+ // Publishing notifications should not block the API response.
+ go func() {
+ verifiers, err := entry.Verifiers()
+ if err != nil {
+ incPublishEvent(newentry.Name, "", false)
+ log.ContextLogger(ctx).Errorf("Could not get verifiers for log entry %s: %v", entryID, err)
+ return
+ }
+ var subjects []string
+ for _, v := range verifiers {
+ subjects = append(subjects, v.Subjects()...)
+ }
+
+ pbEntry, err := tle.GenerateTransparencyLogEntry(logEntryAnon)
+ if err != nil {
+ incPublishEvent(newentry.Name, "", false)
+ log.ContextLogger(ctx).Error(err)
+ return
+ }
+ event, err := newentry.New(entryID, pbEntry, subjects)
+ if err != nil {
+ incPublishEvent(newentry.Name, "", false)
+ log.ContextLogger(ctx).Error(err)
+ return
+ }
+ if viper.GetBool("rekor_server.publish_events_protobuf") {
+ go publishEvent(ctx, api.newEntryPublisher, event, events.ContentTypeProtobuf)
+ }
+ if viper.GetBool("rekor_server.publish_events_json") {
+ go publishEvent(ctx, api.newEntryPublisher, event, events.ContentTypeJSON)
+ }
+ }()
+ }
+
  return logEntry, nil
 }
 
+func publishEvent(ctx context.Context, publisher pubsub.Publisher, event *events.Event, contentType events.EventContentType) {
+ err := publisher.Publish(context.WithoutCancel(ctx), event, contentType)
+ incPublishEvent(event.Type().Name(), contentType, err == nil)
+ if err != nil {
+ log.ContextLogger(ctx).Error(err)
+ }
+}
+
+func incPublishEvent(event string, contentType events.EventContentType, ok bool) {
+ status := "SUCCESS"
+ if !ok {
+ status = "ERROR"
+ }
+ labels := map[string]string{
+ "event": event,
+ "status": status,
+ "content_type": string(contentType),
+ }
+ metricPublishEvents.With(labels).Inc()
+}
+
 // CreateLogEntryHandler creates new entry into log
 func CreateLogEntryHandler(params entries.CreateLogEntryParams) middleware.Responder {
  httpReq := params.HTTPRequest

pkg/api/metrics.go

@@ -29,6 +29,11 @@ var (
  Help: "The total number of new log entries",
  })
 
+ metricPublishEvents = promauto.NewCounterVec(prometheus.CounterOpts{
+ Name: "rekor_publish_events",
+ Help: "The status of publishing events to Pub/Sub",
+ }, []string{"event", "content_type", "status"})
+
  MetricLatency = promauto.NewHistogramVec(prometheus.HistogramOpts{
  Name: "rekor_api_latency",
  Help: "Api Latency on calls",

pkg/events/doc.go

@@ -0,0 +1,19 @@
+// Copyright 2023 The Sigstore Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package events provides methods for working with CloudEvents.
+package events
+
+// The version of the CloudEvents specification the package adheres to.
+const CloudEventsSpecVersion = "1.0"