diff --git a/cmd/feedwitness/configs.go b/cmd/feedwitness/configs.go new file mode 100644 index 00000000..a6bfefc3 --- /dev/null +++ b/cmd/feedwitness/configs.go @@ -0,0 +1,97 @@ +// Copyright 2022 Google LLC. All Rights Reserved. +// +// 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 main + +import ( + "context" + _ "embed" // embed is needed to embed files as constants + "fmt" + "iter" + "maps" + + logfmt "github.com/transparency-dev/formats/log" + "github.com/transparency-dev/witness/omniwitness" + "github.com/transparency-dev/formats/note" + "gopkg.in/yaml.v3" +) + +// configYAML contains a list of configuration options for known logs. +type configYAML struct { + Logs []logYAML `yaml:"Logs"` +} + +// logYAML contains the details about a log. +type logYAML struct { + omniwitness.LogYAML + + Feeder logFeeder `yaml:"Feeder"` +} + +// newStaticFeederConfig creates a new config based on the provided YAML data. +func newStaticFeederConfig(yamlCfg []byte) (*staticFeederConfig, error) { + cfg := &configYAML{} + if err := yaml.Unmarshal(yamlCfg, cfg); err != nil { + return nil, fmt.Errorf("failed to unmarshal witness config: %v", err) + } + r := &staticFeederConfig{} + for _, log := range cfg.Logs { + logV, err := note.NewVerifier(log.PublicKey) + if err != nil { + return nil, fmt.Errorf("failed to create signature verifier: %v", err) + } + logCfg := omniwitness.Log{ + VKey: log.PublicKey, + Verifier: logV, + Origin: log.Origin, + URL: log.URL, + } + if log.Origin == "" { + log.Origin = logV.Name() + } + logID := logfmt.ID(log.Origin) + if log.Feeder != None { + f := feederConfig{ + Feeder: log.Feeder, + Log: logCfg, + } + if oldFeeder, found := r.feeders[logID]; found { + return nil, fmt.Errorf("colliding feeder configs found for key %x: %+v and %+v", logID, oldFeeder, f) + } + r.feeders[logID] = f + } + } + return r, nil +} + +type staticFeederConfig struct { + feeders map[string]feederConfig +} + +func (s *staticFeederConfig) Feeders(_ context.Context) iter.Seq2[feederConfig, error] { + return func(yield func(feederConfig, error) bool) { + for _, v := range s.feeders { + if !yield(v, nil) { + return + } + } + } +} + +// Merge adds all feeders configured in other to this config. +// +// Feeders in the base config with the same ID in the config to be merged will be overridden. +func (s *staticFeederConfig) Merge(other *staticFeederConfig) { + maps.Copy(s.feeders, other.feeders) +} diff --git a/cmd/feedwitness/main.go b/cmd/feedwitness/main.go index 3b2a365a..6be6eff4 100644 --- a/cmd/feedwitness/main.go +++ b/cmd/feedwitness/main.go @@ -60,7 +60,7 @@ func main() { ctx := context.Background() - cfg, err := omniwitness.NewStaticLogConfig(omniwitness.DefaultConfigLogs) + cfg, err := newStaticFeederConfig(omniwitness.DefaultConfigLogs) if err != nil { klog.Exitf("failed to instantiate default witness config: %v", err) } @@ -86,7 +86,7 @@ func main() { httpClient := httpClientFromFlags() - witnesses := []omniwitness.Witness{} + witnesses := []targetWitness{} for _, wu := range witnessURL { u, err := url.Parse(wu) if err != nil { @@ -96,21 +96,21 @@ func main() { witness: w_http.NewWitness(u, httpClient), url: wu, } - witness := omniwitness.Witness{ + witness := targetWitness{ Name: wu, Update: lc.Update, } witnesses = append(witnesses, witness) } - rOpts := omniwitness.RunFeedOpts{ + rOpts := runFeedOpts{ Witnesses: witnesses, HTTPClient: httpClient, MaxWitnessQPS: *rateLimit, MatchLogs: *feed, FeederConfigs: cfg.Feeders, } - if err := omniwitness.RunFeeders(ctx, rOpts); err != nil { + if err := runFeeders(ctx, rOpts); err != nil { klog.Errorf("%v", err) } } diff --git a/omniwitness/run_feeders.go b/cmd/feedwitness/run_feeders.go similarity index 75% rename from omniwitness/run_feeders.go rename to cmd/feedwitness/run_feeders.go index de6a26a4..51f5909a 100644 --- a/omniwitness/run_feeders.go +++ b/cmd/feedwitness/run_feeders.go @@ -12,10 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -// TODO(al): We should remove the concept of feeding from Omniwitness now that we're moving to a -// tlog-witness world, and all the stuff in here can then be moved over to `cmd/feedwitness`. - -package omniwitness +package main import ( "context" @@ -24,13 +21,21 @@ import ( "iter" "net/http" "regexp" + "strings" "sync" "github.com/cenkalti/backoff/v5" "github.com/transparency-dev/formats/log" "github.com/transparency-dev/witness/internal/feeder" + "github.com/transparency-dev/witness/internal/feeder/pixelbt" + "github.com/transparency-dev/witness/internal/feeder/rekor_v1" + "github.com/transparency-dev/witness/internal/feeder/serverless" + "github.com/transparency-dev/witness/internal/feeder/sumdb" + "github.com/transparency-dev/witness/internal/feeder/tiles" "github.com/transparency-dev/witness/internal/witness" "github.com/transparency-dev/witness/monitoring" + "github.com/transparency-dev/witness/omniwitness" + "golang.org/x/mod/sumdb/note" "golang.org/x/sync/errgroup" "golang.org/x/time/rate" "k8s.io/klog/v2" @@ -56,17 +61,22 @@ func initFeederMetrics() { }) } -// UpdateFn is the signature of a function which knows how to update a witness. -type UpdateFn func(ctx context.Context, oldSize uint64, newCP []byte, proof [][]byte) ([]byte, uint64, error) +type feederConfig struct { + Log omniwitness.Log + Feeder logFeeder +} + +// updateFn is the signature of a function which knows how to update a witness. +type updateFn func(ctx context.Context, oldSize uint64, newCP []byte, proof [][]byte) ([]byte, uint64, error) -// Witness represents a target witness to be fed. -type Witness struct { - Update UpdateFn +// targetWitness represents a target witness to be fed. +type targetWitness struct { + Update updateFn Name string } -// RunFeedOpts is the configuration to use for RunFeeders. -type RunFeedOpts struct { +// runFeedOpts is the configuration to use for RunFeeders. +type runFeedOpts struct { // MaxWitnessQPS is the maximum number of requests to make per second to any given witness. // If unset, a default of 1 QPS will be assumed. MaxWitnessQPS float64 @@ -75,20 +85,20 @@ type RunFeedOpts struct { // MatchLogs is an optional regex to select a submet of logs to feed. MatchLogs string // FeederConfigs provides access to feeder configs. Required. - FeederConfigs func(context.Context) iter.Seq2[FeederConfig, error] + FeederConfigs func(context.Context) iter.Seq2[feederConfig, error] // Witnesses is the set of witnesses to feed to. Required. - Witnesses []Witness + Witnesses []targetWitness } type wJob struct { logOrigin string - f func(sizeHint uint64, w Witness) (uint64, error) + f func(sizeHint uint64, w targetWitness) (uint64, error) } -// RunFeeders continually feeds checkpoints from logs to witnesses according to the provided config. +// runFeeders continually feeds checkpoints from logs to witnesses according to the provided config. // // This is a long-running function which will only return when the context is done. -func RunFeeders(ctx context.Context, opts RunFeedOpts) error { +func runFeeders(ctx context.Context, opts runFeedOpts) error { initFeederMetrics() if opts.HTTPClient == nil { @@ -183,7 +193,7 @@ func RunFeeders(ctx context.Context, opts RunFeedOpts) error { select { case wc <- wJob{ logOrigin: c.Log.Origin, - f: func(sizeHint uint64, w Witness) (uint64, error) { + f: func(sizeHint uint64, w targetWitness) (uint64, error) { return feedOnce(ctx, sizeHint, w, cp, src) }, }: @@ -203,7 +213,7 @@ func RunFeeders(ctx context.Context, opts RunFeedOpts) error { // The provided sizeHint is size of the log that the caller believes is current on the target witness. // // Returns a new hint on what the current size of the log on the target witness. -func feedOnce(ctx context.Context, sizeHint uint64, w Witness, cp []byte, src feeder.Source) (uint64, error) { +func feedOnce(ctx context.Context, sizeHint uint64, w targetWitness, cp []byte, src feeder.Source) (uint64, error) { klog.V(2).Infof("CP to feed:\n%s", string(cp)) cpSubmit, _, _, err := log.ParseCheckpoint(cp, src.LogOrigin, src.LogSigVerifier) @@ -219,7 +229,7 @@ func feedOnce(ctx context.Context, sizeHint uint64, w Witness, cp []byte, src fe } // submitToWitness will submit the checkpoint to the witness, retrying up to 3 times if the local checkpoint is stale. -func submitToWitness(ctx context.Context, sizeHint uint64, cpRaw []byte, cpSubmit log.Checkpoint, fetchProof feeder.FetchProofFn, w Witness) (uint64, error) { +func submitToWitness(ctx context.Context, sizeHint uint64, cpRaw []byte, cpSubmit log.Checkpoint, fetchProof feeder.FetchProofFn, w targetWitness) (uint64, error) { // Since this func will be executed by the backoff mechanism below, we'll // log any error messages directly in here before returning the error, as // the backoff util doesn't seem to log them itself. @@ -286,3 +296,80 @@ func statusForError(e error) string { return "unknown_error" } } + +// logFeeder is an enum of the known feeder types. +type logFeeder uint8 + +const ( + Serverless logFeeder = iota + 1 + SumDB + Pixel + Rekor + Tiles + None +) + +var ( + feederByName = map[string]logFeeder{ + "serverless": Serverless, + "sumdb": SumDB, + "pixel": Pixel, + "rekor": Rekor, + "tiles": Tiles, + "none": None, + } + feederNameByID = func() map[logFeeder]string { + r := make(map[logFeeder]string) + for k, v := range feederByName { + r[v] = k + } + return r + }() +) + +// UnmarshalYAML populates the log from yaml using the unmarshal func provided. +func (f *logFeeder) UnmarshalYAML(unmarshal func(any) error) (err error) { + var raw string + if err := unmarshal(&raw); err != nil { + return err + } + if *f, err = parseFeeder(raw); err != nil { + return err + } + return nil +} + +// MarshalYAML serializes the feeder to its string representation. +func (f logFeeder) MarshalYAML() (any, error) { + return f.String(), nil +} + +func (f logFeeder) NewSourceFunc() func(origin string, v note.Verifier, url string, c *http.Client) (feeder.Source, error) { + switch f { + case Serverless: + return serverless.NewFeedSource + case SumDB: + return sumdb.NewFeedSource + case Pixel: + return pixelbt.NewFeedSource + case Rekor: + return rekor_v1.NewFeedSource + case Tiles: + return tiles.NewFeedSource + } + panic(fmt.Sprintf("unknown feeder enum: %q", f)) +} + +func (f logFeeder) String() string { + return feederNameByID[f] +} + +// ParseFeeder takes a string and returns a valid enum or an error. +func parseFeeder(f string) (logFeeder, error) { + f = strings.TrimSpace(strings.ToLower(f)) + value, ok := feederByName[f] + if !ok { + return logFeeder(0), fmt.Errorf("unknown feeder type %q", f) + } + return value, nil +} diff --git a/omniwitness/run_feeders_test.go b/cmd/feedwitness/run_feeders_test.go similarity index 96% rename from omniwitness/run_feeders_test.go rename to cmd/feedwitness/run_feeders_test.go index 2c52e8b6..2fb36d44 100644 --- a/omniwitness/run_feeders_test.go +++ b/cmd/feedwitness/run_feeders_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package omniwitness +package main import ( "context" @@ -38,7 +38,7 @@ func TestFeedOnce(t *testing.T) { for _, test := range []struct { desc string submitCP []byte - update UpdateFn + update updateFn wantErr bool }{ { @@ -89,7 +89,7 @@ func TestFeedOnce(t *testing.T) { LogSigVerifier: testdata.LogSigVerifier(t), } t.Run(test.desc, func(t *testing.T) { - _, err := feedOnce(ctx, 0, Witness{Update: test.update}, test.submitCP, src) + _, err := feedOnce(ctx, 0, targetWitness{Update: test.update}, test.submitCP, src) gotErr := err != nil if test.wantErr != gotErr { t.Fatalf("Got err %v, want err %t", err, test.wantErr) diff --git a/cmd/loadtest/loadtest.go b/cmd/loadtest/loadtest.go index 407eac31..18e030b6 100644 --- a/cmd/loadtest/loadtest.go +++ b/cmd/loadtest/loadtest.go @@ -175,7 +175,6 @@ func (ls inMemoryLogs) config() string { Origin: l.o, URL: fmt.Sprintf("http://%s/", l.o), PublicKey: l.vkey, - Feeder: omniwitness.None, } } out, err := yaml.Marshal(cfg) diff --git a/cmd/omniwitness/monolith.go b/cmd/omniwitness/monolith.go index ed8a3634..c3c3b42e 100644 --- a/cmd/omniwitness/monolith.go +++ b/cmd/omniwitness/monolith.go @@ -12,8 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -// omniwitness is a single executable that runs all of the feeders and witness -// in a single process. +// omniwitness is a single executable that runs a pre-configured witness. package main import ( @@ -65,8 +64,6 @@ var ( additionalLogYaml = flag.String("additional_logs", "", "The path to an optional addition logs YAML file. Entries in this file will be *added* to the logs configured by default") publicWitnessConfigs multiStringFlag publicWitnessConfigInterval = flag.Duration("public_witness_config_poll_interval", 1*time.Minute, "Interval between checking the public witness config for new logs to add.") - - pollInterval = flag.Duration("poll_interval", 1*time.Minute, "Time to wait between polling logs for new checkpoints. Set to 0 to disable polling logs.") ) type logAndConfigPersistence interface { @@ -181,9 +178,7 @@ func main() { BastionKey: bastionKey, RateLimit: *rateLimit, DistributeRateLimit: *distributeRateLimit, - FeedInterval: *pollInterval, Logs: p, - Feeders: l.Feeders, WitnessNetworkConfigURLs: publicWitnessConfigs, WitnessNetworkConfigInterval: *publicWitnessConfigInterval, } diff --git a/omniwitness/configs.go b/omniwitness/configs.go index 1d5a55f0..5cd50ec4 100644 --- a/omniwitness/configs.go +++ b/omniwitness/configs.go @@ -49,7 +49,6 @@ type LogYAML struct { Origin string `yaml:"Origin"` PublicKey string `yaml:"PublicKey"` URL string `yaml:"URL"` - Feeder Feeder `yaml:"Feeder"` } // NewStaticLogConfig creates a new LogConfig based on the provided YAML data. @@ -60,7 +59,6 @@ func NewStaticLogConfig(yamlCfg []byte) (*staticLogConfig, error) { } r := &staticLogConfig{ logs: make(map[string]Log), - feeders: make(map[string]FeederConfig), } for _, log := range cfg.Logs { logV, err := note.NewVerifier(log.PublicKey) @@ -77,16 +75,6 @@ func NewStaticLogConfig(yamlCfg []byte) (*staticLogConfig, error) { log.Origin = logV.Name() } logID := logfmt.ID(log.Origin) - if log.Feeder != None { - f := FeederConfig{ - Feeder: log.Feeder, - Log: logCfg, - } - if oldFeeder, found := r.feeders[logID]; found { - return nil, fmt.Errorf("colliding feeder configs found for key %x: %+v and %+v", logID, oldFeeder, f) - } - r.feeders[logID] = f - } if oldLog, found := r.logs[logID]; found { return nil, fmt.Errorf("colliding log configs found for key %x: %+v and %+v", logID, oldLog, logCfg) } @@ -97,7 +85,6 @@ func NewStaticLogConfig(yamlCfg []byte) (*staticLogConfig, error) { type staticLogConfig struct { logs map[string]Log - feeders map[string]FeederConfig } func (s *staticLogConfig) Logs(_ context.Context) iter.Seq2[Log, error] { @@ -110,16 +97,6 @@ func (s *staticLogConfig) Logs(_ context.Context) iter.Seq2[Log, error] { } } -func (s *staticLogConfig) Feeders(_ context.Context) iter.Seq2[FeederConfig, error] { - return func(yield func(FeederConfig, error) bool) { - for _, v := range s.feeders { - if !yield(v, nil) { - return - } - } - } -} - func (s *staticLogConfig) Log(_ context.Context, origin string) (Log, bool, error) { logID := logfmt.ID(origin) l, ok := s.logs[logID] diff --git a/omniwitness/configs_test.go b/omniwitness/configs_test.go index 61c84036..9363ed24 100644 --- a/omniwitness/configs_test.go +++ b/omniwitness/configs_test.go @@ -52,22 +52,6 @@ func testConfig(t *testing.T, cfg []byte) { t.Fatal("no logs defined in config") } } - - { - c := 0 - for f, err := range logCfg.Feeders(t.Context()) { - if err != nil { - t.Fatalf("Failed to iterate over feeders: %v", err) - } - if f.Feeder == omniwitness.None { - t.Errorf("log %q has unknown feeder", f.Log.Origin) - } - c++ - } - if c == 0 { - t.Fatal("no feeders defined in config") - } - } } func TestProdConfig(t *testing.T) { @@ -84,7 +68,6 @@ Logs: - Origin: go.sum database tree URL: https://sum.golang.org PublicKey: sum.golang.org+033de0ae+Ac4zctda0e5eza+HJyk9SxEdh+s3Ux18htTTAD8OuAn8 - Feeder: sumdb `)) if err != nil { t.Fatalf("Failed to parse base config: %v", err) @@ -95,7 +78,6 @@ Logs: - Origin: Armory Drive Prod 2 URL: https://raw.githubusercontent.com/f-secure-foundry/armory-drive-log/master/log/ PublicKey: armory-drive-log+16541b8f+AYDPmG5pQp4Bgu0a1mr5uDZ196+t8lIVIfWQSPWmP+Jv - Feeder: serverless `)) if err != nil { t.Fatalf("Failed to parse extra config: %v", err) diff --git a/omniwitness/omniwitness.go b/omniwitness/omniwitness.go index 9cb43c50..82fe9424 100644 --- a/omniwitness/omniwitness.go +++ b/omniwitness/omniwitness.go @@ -25,11 +25,9 @@ import ( "iter" "net" "net/http" - "strings" "time" "github.com/transparency-dev/witness/api" - "github.com/transparency-dev/witness/internal/feeder" "github.com/transparency-dev/witness/internal/persistence" "github.com/transparency-dev/witness/internal/witness" "golang.org/x/mod/sumdb/note" @@ -38,11 +36,6 @@ import ( "k8s.io/klog/v2" "github.com/transparency-dev/witness/internal/bastion" - "github.com/transparency-dev/witness/internal/feeder/pixelbt" - "github.com/transparency-dev/witness/internal/feeder/rekor_v1" - "github.com/transparency-dev/witness/internal/feeder/serverless" - "github.com/transparency-dev/witness/internal/feeder/sumdb" - "github.com/transparency-dev/witness/internal/feeder/tiles" ) // LogStatePersistence describes functionality the omniwitness requires @@ -93,7 +86,6 @@ type OperatorConfig struct { // TODO(mhutchinson): This should be baked into the code when there is a public distributor. RestDistributorBaseURL string - FeedInterval time.Duration DistributeInterval time.Duration // DistributeRateLimit is the maximum number of calls per second to the configured distributor. DistributeRateLimit float64 @@ -104,9 +96,6 @@ type OperatorConfig struct { // If unset, uses the embedded default config. Logs LogConfig - // Feeders provides the witness with the config for self-feeding from logs. - Feeders func(context.Context) iter.Seq2[FeederConfig, error] - // WitnessNetworkConfigURLs is optional, and may be set to one or more URLs pointing to resources // in the public witness network config format. // These resources will be periodically retrieved and incorporated into the LogConfig provided above. @@ -128,16 +117,10 @@ type LogConfig interface { AddLogs(ctx context.Context, cfg []Log) error } -type FeederConfig struct { - Log Log - Feeder Feeder -} - // Main runs the omniwitness, with the witness listening using the listener, and all // outbound HTTP calls using the client provided. func Main(ctx context.Context, operatorConfig OperatorConfig, p Persistence, httpListener net.Listener, httpClient *http.Client) error { initHTTPMetrics() - initFeederMetrics() // This error group will be used to run all top level processes. // If any process dies, then all of them will be stopped via context cancellation. @@ -188,16 +171,6 @@ func Main(ctx context.Context, operatorConfig OperatorConfig, p Persistence, htt if operatorConfig.WitnessNetworkConfigInterval == 0 && len(operatorConfig.WitnessNetworkConfigURLs) > 0 { operatorConfig.WitnessNetworkConfigInterval = defaultProvisionInterval } - if operatorConfig.FeedInterval > 0 && operatorConfig.Feeders != nil { - rOpts := RunFeedOpts{ - Witnesses: []Witness{{Name: operatorConfig.WitnessVerifier.Name(), Update: witness.Update}}, - HTTPClient: httpClient, - MaxWitnessQPS: float64(time.Second) / float64(operatorConfig.FeedInterval), - FeederConfigs: operatorConfig.Feeders, - } - g.Go(func() error { return RunFeeders(ctx, rOpts) }) - - } operatorConfig.ServeMux.Handle(api.HTTPAddCheckpoint, http.MaxBytesHandler(handler, 16*1024)) if operatorConfig.BastionAddr != "" && operatorConfig.BastionKey != nil { @@ -267,80 +240,3 @@ func runRestDistributors(ctx context.Context, g *errgroup.Group, httpClient *htt } }) } - -// Feeder is an enum of the known feeder types. -type Feeder uint8 - -const ( - Serverless Feeder = iota + 1 - SumDB - Pixel - Rekor - Tiles - None -) - -var ( - feederByName = map[string]Feeder{ - "serverless": Serverless, - "sumdb": SumDB, - "pixel": Pixel, - "rekor": Rekor, - "tiles": Tiles, - "none": None, - } - feederNameByID = func() map[Feeder]string { - r := make(map[Feeder]string) - for k, v := range feederByName { - r[v] = k - } - return r - }() -) - -// UnmarshalYAML populates the log from yaml using the unmarshal func provided. -func (f *Feeder) UnmarshalYAML(unmarshal func(any) error) (err error) { - var raw string - if err := unmarshal(&raw); err != nil { - return err - } - if *f, err = ParseFeeder(raw); err != nil { - return err - } - return nil -} - -// MarshalYAML serializes the feeder to its string representation. -func (f Feeder) MarshalYAML() (any, error) { - return f.String(), nil -} - -func (f Feeder) NewSourceFunc() func(origin string, v note.Verifier, url string, c *http.Client) (feeder.Source, error) { - switch f { - case Serverless: - return serverless.NewFeedSource - case SumDB: - return sumdb.NewFeedSource - case Pixel: - return pixelbt.NewFeedSource - case Rekor: - return rekor_v1.NewFeedSource - case Tiles: - return tiles.NewFeedSource - } - panic(fmt.Sprintf("unknown feeder enum: %q", f)) -} - -func (f Feeder) String() string { - return feederNameByID[f] -} - -// ParseFeeder takes a string and returns a valid enum or an error. -func ParseFeeder(f string) (Feeder, error) { - f = strings.TrimSpace(strings.ToLower(f)) - value, ok := feederByName[f] - if !ok { - return Feeder(0), fmt.Errorf("unknown feeder type %q", f) - } - return value, nil -}