From b9f5ab718ca46b0f376694e50eb6672f48cf6c2a Mon Sep 17 00:00:00 2001 From: Jacob Hoffman-Andrews Date: Tue, 9 Jun 2026 18:37:30 -0700 Subject: [PATCH 1/7] mtca: add sequencing This is a partial implementation: it doesn't really write to tile storage, or calculate realistic hashes. In Loop(), take the current contents of the pool, "sequence" them by increasing the tree size and generating a fake root hash randomly, and write a new checkpoint row. Add a new table, `latestCheckpoints`. This has a single row per log, referring to the latest checkpoint. This allows us to take an exclusive lock while signing to prevent signing split views. --- .gitignore | 1 + cmd/boulder-mtca/main.go | 39 +- mtca/mtca.go | 496 +++++++++++++++++- mtca/mtca_test.go | 281 ++++++++++ sa/db/01-mtcmeta_44947_4_1_0_44.sql | 14 +- sa/db/02-users_next.sql | 5 + .../mtcmeta_44947_4_1_0_44/vschema.json | 7 +- test/certs/genmtpki/genmtpki.go | 6 +- test/config-next/mtca.json | 6 +- test/config-next/proxysql/mtca1_dburl | 1 + test/config-next/vitess/mtca1_dburl | 1 + test/config/proxysql/mtca1_dburl | 1 + test/config/vitess/mtca1_dburl | 1 + test/entrypoint.sh | 1 + test/mtca/README.md | 6 + test/mtca/clear.sh | 5 + test/mtca/start.sh | 6 + test/proxysql/proxysql.cnf | 3 + test/startservers.py | 3 +- 19 files changed, 856 insertions(+), 27 deletions(-) create mode 100644 mtca/mtca_test.go create mode 100644 test/config-next/proxysql/mtca1_dburl create mode 100644 test/config-next/vitess/mtca1_dburl create mode 100644 test/config/proxysql/mtca1_dburl create mode 100644 test/config/vitess/mtca1_dburl create mode 100644 test/mtca/README.md create mode 100755 test/mtca/clear.sh create mode 100755 test/mtca/start.sh diff --git a/.gitignore b/.gitignore index d3b34996c98..eac8ac895c9 100644 --- a/.gitignore +++ b/.gitignore @@ -55,3 +55,4 @@ test/secrets/mtpublisher_dburl test/secrets/revoker_dburl test/secrets/sa_dburl test/secrets/sa_ro_dburl +test/secrets/mtca1_dburl diff --git a/cmd/boulder-mtca/main.go b/cmd/boulder-mtca/main.go index 34bd6594d92..dfaa822446d 100644 --- a/cmd/boulder-mtca/main.go +++ b/cmd/boulder-mtca/main.go @@ -4,12 +4,16 @@ package notmain import ( "context" + "database/sql" + "errors" "flag" "os" "github.com/jmhodges/clock" + "github.com/letsencrypt/borp" "github.com/letsencrypt/boulder/cmd" + "github.com/letsencrypt/boulder/config" bgrpc "github.com/letsencrypt/boulder/grpc" "github.com/letsencrypt/boulder/issuance" mtca "github.com/letsencrypt/boulder/mtca" @@ -22,12 +26,17 @@ type Config struct { GRPCMTCA *cmd.GRPCServerConfig + DB cmd.DBConfig + // Issuer holds the configuration for a single MTCA instance with a single mtcaID. // We run a separate process for each issuer. // TODO: the issuance package parses the CA certificate as a self-signed X.509 // certificate, but per MTC draft, a CA SHOULD be represented by an RFC 9925 // unsigned certificate: https://www.rfc-editor.org/rfc/rfc9925.html. Issuer issuance.IssuerConfig + + // SequencingPeriod controls how frequently the MTCA sequences a batch and signs a checkpoint. + SequencingPeriod config.Duration `validate:"required"` } Syslog cmd.SyslogConfig @@ -38,6 +47,9 @@ func main() { grpcAddr := flag.String("addr", "", "gRPC listen address override") debugAddr := flag.String("debug-addr", "", "Debug server address override") configFile := flag.String("config", "", "File path to the configuration file for this service") + initLog := flag.Bool("init-log", false, "Initialize log metadata in the database and exit") + initLogForTest := flag.Bool("init-log-for-test", false, "For testing: initialize log metadata if not already initialized, then serve") + flag.Parse() if *configFile == "" { flag.Usage() @@ -67,7 +79,26 @@ func main() { issuer, err := issuance.LoadIssuer(c.MTCA.Issuer, clk) cmd.FailOnError(err, "Loading issuer") - mtcaImpl := mtca.New(issuer) + url, err := c.MTCA.DB.URL() + cmd.FailOnError(err, "Reading DB URL") + db, err := sql.Open("mysql", url) + cmd.FailOnError(err, "Opening DB") + dbMap := &borp.DbMap{Db: db, Dialect: borp.MySQLDialect{}} + + mtcaImpl, err := mtca.New(issuer, dbMap, logger) + cmd.FailOnError(err, "Building MTCA") + + if *initLog { + err = mtcaImpl.InitLog(context.Background()) + cmd.FailOnError(err, "Initializing log") + return + } + if *initLogForTest { + err = mtcaImpl.InitLog(context.Background()) + if err != nil && !errors.Is(err, mtca.ErrIssuanceLogAlreadyInitialized) { + cmd.FailOnError(err, "Initializing MTC log DB for test") + } + } srv := bgrpc.NewServer(c.MTCA.GRPCMTCA, logger).Add( &mtcapb.MTCA_ServiceDesc, mtcaImpl) @@ -75,6 +106,12 @@ func main() { start, err := srv.Build(tlsConfig, scope, clk) cmd.FailOnError(err, "Unable to setup MTCA gRPC server") + ctx, cancel := context.WithCancel(context.Background()) + // Cancel will be called after start() returns, which happens after GracefulStop() returns. + // That means all inflight RPCs will be done, which means the last of the pool has been sequenced. + defer cancel() + go mtcaImpl.Loop(ctx, c.MTCA.SequencingPeriod.Duration) + cmd.FailOnError(start(), "MTCA gRPC service failed") } diff --git a/mtca/mtca.go b/mtca/mtca.go index c40eb1df21c..4b2878ac4f9 100644 --- a/mtca/mtca.go +++ b/mtca/mtca.go @@ -4,32 +4,147 @@ package mtca import ( "context" + "crypto/rand" + "crypto/sha256" + "crypto/x509" + "database/sql" "encoding/asn1" + "errors" "fmt" "sync" + "time" + corepb "github.com/letsencrypt/boulder/core/proto" + "github.com/letsencrypt/boulder/db" "github.com/letsencrypt/boulder/issuance" mtcapb "github.com/letsencrypt/boulder/mtca/proto" + "github.com/letsencrypt/boulder/trees/cosigned" + + "github.com/letsencrypt/borp" + blog "github.com/letsencrypt/boulder/log" ) var _ mtcapb.MTCAServer = &mtca{} -func New(issuer *issuance.Issuer) *mtca { - var mtcaID string +func getMTCAID(issuerCert *x509.Certificate) (string, error) { testingTrustAnchorIDOID := asn1.ObjectIdentifier{1, 3, 6, 1, 4, 1, 44363, 47, 1} - for _, attribute := range issuer.Cert.Subject.Names { + for _, attribute := range issuerCert.Subject.Names { if attribute.Type.Equal(testingTrustAnchorIDOID) { - mtcaID, _ = attribute.Value.(string) - break + mtcaID, ok := attribute.Value.(string) + if !ok { + return "", fmt.Errorf("invalid trust anchor attribute type %T", attribute.Value) + } + return mtcaID, nil } } + + return "", fmt.Errorf("issuer subject %q did not contain trust anchor ID OID %q", + issuerCert.Subject, testingTrustAnchorIDOID) +} + +// New creates a new MTCA service. +func New(issuer *issuance.Issuer, dbMap *borp.DbMap, logger blog.Logger) (*mtca, error) { + mtcaID, err := getMTCAID(issuer.Cert.Certificate) + if err != nil { + return nil, err + } + return &mtca{ + log: logger, + db: initDB(dbMap), issuer: issuer, mtcaID: mtcaID, // TODO: collect this from config - logNumber: 0, - latestEntryIndex: 0, + logNumber: 0, + pool: &pool{maxSize: 100}, + }, nil +} + +func initDB(dbMap *borp.DbMap) *db.WrappedMap { + dbMap.AddTableWithName(checkpoint{}, "checkpoints").SetKeys(true, "ID") + return db.NewWrappedMap(dbMap) +} + +var ErrIssuanceLogAlreadyInitialized = errors.New("issuance log already initialized") +var ErrCheckpointNotReady = errors.New("not ready - no mirror signature") + +// InitLog creates the database metadata for a new, empty log: one checkpoint and the row +// in `latestCheckpoint` that refers to it. Should only be run once in a log's lifetime. +func (m *mtca) InitLog(ctx context.Context) error { + _, err := m.latest(ctx) + if err == nil { + return ErrIssuanceLogAlreadyInitialized + } + + var numLatestCheckpoints int64 + err = m.db.SelectOne(ctx, &numLatestCheckpoints, "SELECT COUNT(*) FROM latestCheckpoint WHERE mtcLogID = ?", + m.mtcLogID()) + if err != nil { + return fmt.Errorf("getting latestCheckpoint: %s", err) + } + + var numCheckpoints int64 + err = m.db.SelectOne(ctx, &numCheckpoints, "SELECT COUNT(*) FROM checkpoints WHERE mtcLogID = ?", + m.mtcLogID()) + if err != nil { + return fmt.Errorf("getting checkpoints: %s", err) + } + + if numCheckpoints > 0 || numLatestCheckpoints > 0 { + return fmt.Errorf("initializing issuance log for %s: already has %d checkpoints and %d latestCheckpoint rows", + m.mtcLogID(), numCheckpoints, numLatestCheckpoints) + } + + // null_entry has empty extensions and a MerkleTreeCertEntryType of 0. Since extensions can be up to 2^16 long + // there's two bytes of length prefix. Since MerkleTreeCertEntryType can have up to 2^16 values, it's also two bytes. + // All the bytes are zero: empty extensions, null_entry type is enum value zero. + // https://ietf-plants-wg.github.io/merkle-tree-certs/draft-ietf-plants-merkle-tree-certs.html#name-log-entries + // To calculate the Merkle Tree Hash of a single-entry list, we prepend 0x00 (as compared with 0x01 when hashing + // two nodes). So five zeroes total. + // https://www.rfc-editor.org/info/rfc9162/#name-definition-of-the-merkle-tr + nullEntry := []byte{0, 0, 0, 0, 0} + rootHash := sha256.Sum256(nullEntry) + + firstCheckpoint := checkpoint{ + MTCLogID: m.mtcLogID(), + MTCASignature: nil, + MirrorID: "", + MirrorSignature: nil, + TreeSize: 1, + RootHash: rootHash[:], + } + + _, err = db.WithTransaction(ctx, m.db, func(tx db.Executor) (any, error) { + sig, err := m.signCheckpoint(&firstCheckpoint) + if err != nil { + return nil, err + } + + firstCheckpoint.MTCASignature = sig + + err = tx.Insert(ctx, &firstCheckpoint) + if err != nil { + return nil, err + } + + _, err = tx.ExecContext(ctx, "INSERT INTO latestCheckpoint (id, mtcLogID) VALUES (?, ?)", + firstCheckpoint.ID, m.mtcLogID()) + if err != nil { + return nil, fmt.Errorf("inserting latestCheckpoint: %s", err) + } + + return nil, nil + }) + if err != nil { + return err + } + + _, err = m.latest(ctx) + if err != nil { + return fmt.Errorf("fetching first checkpoint: %s", err) } + + return err } type mtca struct { @@ -39,10 +154,48 @@ type mtca struct { mtcaID string logNumber uint16 - // This is just a dummy for testing; in reality this will come from the DB. - latestEntryIndex int64 + db *db.WrappedMap + log blog.Logger + + pool *pool + + sequencingMu sync.Mutex +} + +type entry struct { + pubkey []byte + identifiers []*corepb.Identifier + ch chan<- int64 +} + +type pool struct { + sync.Mutex + entries []entry + maxSize int +} + +func (p *pool) take() []entry { + p.Lock() + defer p.Unlock() + ret := p.entries + p.entries = nil + return ret +} + +func (p *pool) len() int { + p.Lock() + defer p.Unlock() + return len(p.entries) +} - sequencing sync.Mutex +func (p *pool) append(e entry) error { + p.Lock() + defer p.Unlock() + if len(p.entries) >= p.maxSize { + return fmt.Errorf("pool is full") + } + p.entries = append(p.entries, e) + return nil } // mtcLogID returns the string-formatted relative OID for this log. @@ -53,12 +206,321 @@ func (m *mtca) mtcLogID() string { } func (m *mtca) Issue(ctx context.Context, req *mtcapb.IssueRequest) (*mtcapb.IssueResponse, error) { - m.sequencing.Lock() - defer m.sequencing.Unlock() - m.latestEntryIndex++ + ch := make(chan int64, 1) + err := m.pool.append(entry{ + pubkey: req.Pubkey, + identifiers: req.Identifiers, + ch: ch, + }) + if err != nil { + return nil, err + } - return &mtcapb.IssueResponse{ - MtcLogID: m.mtcLogID(), - MtcEntryIndex: m.latestEntryIndex, - }, nil + select { + case <-ctx.Done(): + return nil, ctx.Err() + case entryIndex := <-ch: + if entryIndex < 0 { + return nil, errors.New("error during sequencing") + } + return &mtcapb.IssueResponse{ + MtcLogID: m.mtcLogID(), + MtcEntryIndex: entryIndex, + }, nil + } +} + +// Loop periodically sequences all entries in the pool and sends notifications to the waiting RPCs. +// +// At process shutdown, this context should be canceled _after_ GracefulStop returns. That ensures +// there are no inflight RPCs from clients, which in turn ensures that we have sequenced everything +// had in the pool. +func (m *mtca) Loop(ctx context.Context, sequencingPeriod time.Duration) { + if sequencingPeriod <= 0 { + sequencingPeriod = 100 * time.Millisecond + } + + go m.fakePublisher(ctx) + + since := time.Now() + ticker := time.NewTicker(sequencingPeriod) + for { + select { + case <-ticker.C: + err := m.sequence(ctx) + if err != nil { + if !errors.Is(err, ErrCheckpointNotReady) { + m.log.Errf("sequencing: %s", err) + } else if time.Since(since) > 10*sequencingPeriod { + m.log.Errf("after %s: %s", time.Since(since).Round(time.Millisecond), err) + } + continue + } + since = time.Now() + case <-ctx.Done(): + poolSize := m.pool.len() + if poolSize != 0 { + m.log.Errf("shutting down loop: pool has %d entries left. ungraceful stop?", poolSize) + } + return + } + } +} + +// fakePublisher simulates the role of the mtpublisher by finding checkpoints with no +// mirrorSignature and writing a fake signature to them. +// +// TODO: remove once a real publisher is available in integration. +func (m *mtca) fakePublisher(ctx context.Context) { + ticker := time.NewTicker(37 * time.Millisecond) + for { + select { + case <-ticker.C: + latest, err := m.latest(ctx) + if err != nil { + m.log.Errf("getting latest checkpoint for fake publisher: %s", err) + continue + } + _, err = m.db.ExecContext(ctx, ` + UPDATE checkpoints SET mirrorID = ?, mirrorSignature = ? + WHERE id = ? AND mtcLogID = ?`, + "fake mirror ID", []byte("fake mirror signature"), + latest.ID, m.mtcLogID()) + if err != nil { + m.log.Errf("updating latest checkpoint with fake signature: %s", err) + continue + } + case <-ctx.Done(): + return + } + } +} + +func (m *mtca) sequence(ctx context.Context) error { + if m.pool.len() == 0 { + return nil + } + + latest, err := m.latest(ctx) + if err != nil { + return err + } + + if !latest.sequencingReady() { + return fmt.Errorf("temporary: checkpoint ID %d (tree size %d): %w", + latest.ID, latest.TreeSize, ErrCheckpointNotReady) + } + + m.sequencingMu.Lock() + defer m.sequencingMu.Unlock() + + // Pull the contents of the pool. + entries := m.pool.take() + if len(entries) == 0 { + return nil + } + + // Since we've taken ownership of the previously-pooled entries, make sure we notify + // the waiting RPCs of either a success or a failure. + defer func() { + for _, e := range entries { + e.ch <- -1 + } + }() + + // Simulate writing to tile storage + latestTreeSize := latest.TreeSize + var entryIndexes []int64 + for range entries { + entryIndexes = append(entryIndexes, latestTreeSize) + latestTreeSize++ + } + + // TODO: calculate new root hash for real + var newRootHash [sha256.Size]byte + rand.Read(newRootHash[:]) + + newCheckpoint := checkpoint{ + ID: 0, + MTCLogID: m.mtcLogID(), + MTCASignature: nil, + MirrorID: "", + MirrorSignature: nil, + TreeSize: latestTreeSize, + RootHash: newRootHash[:], + } + + // Precommit to the new checkpoint. This will allow us to do recovery if we crash between signing + // the new checkpoint and writing it to the database. + // + // TODO: crash recovery. When MTCA starts up, if there is a checkpoint with no MTCA signature, MTCA + // should check for staged tiles. Assuming the staged tiles and the checkpoint are consistent with + // the previous, signed, checkpoint, MTCA should try to re-sign the checkpoint and proceed from there. + // + // Note: Insert() updates the ID field of its parameter due to SetKeys(true, "ID") + err = m.db.Insert(ctx, &newCheckpoint) + if err != nil { + return err + } + + _, err = db.WithTransaction(ctx, m.db, func(tx db.Executor) (any, error) { + var latestID int64 + // Lock the latestCheckpoint to make sure there is no concurrent signer/writer, avoiding signing a split view. + // The FOR UPDATE does the heavy lifting here. + // https://mariadb.com/docs/server/reference/sql-statements/data-manipulation/selecting-data/for-update + err := tx.SelectOne(ctx, &latestID, + `SELECT id from latestCheckpoint WHERE mtcLogID = ? FOR UPDATE`, + m.mtcLogID()) + if err != nil { + return nil, err + } + if latestID != latest.ID { + return nil, fmt.Errorf("latestCheckpoint changed during sequencing from %d to %d. multiple writers?", + latest.ID, latestID) + } + + // Note that we're doing HSM work while holding a database lock. That's intentional; the database lock + // is to prevent the possibility of a concurrent signer on the same tree. + sig, err := m.signCheckpoint(&newCheckpoint) + if err != nil { + return nil, err + } + + result, err := tx.ExecContext(ctx, "UPDATE checkpoints SET mtcaSignature = ? WHERE mtcLogID = ? AND id = ?", + sig, m.mtcLogID(), newCheckpoint.ID) + if err != nil { + return nil, fmt.Errorf("updating checkpoint: %s", err) + } + rowsAffected, err := result.RowsAffected() + if err != nil { + return nil, fmt.Errorf("updating checkpoint, getting rows affected: %s", err) + } + if rowsAffected != 1 { + return nil, fmt.Errorf("updating checkpoint: %d rows updated, rolling back", rowsAffected) + } + + result, err = tx.ExecContext(ctx, "UPDATE latestCheckpoint SET id = ? WHERE mtcLogID = ? AND id = ?", + newCheckpoint.ID, m.mtcLogID(), latestID) + if err != nil { + return nil, fmt.Errorf("updating latestCheckpoint: %s", err) + } + rowsAffected, err = result.RowsAffected() + if err != nil { + return nil, fmt.Errorf("updating latestCheckpoint, getting rows affected: %s", err) + } + if rowsAffected != 1 { + return nil, fmt.Errorf("updating latestCheckpoint: %d rows updated, rolling back", rowsAffected) + } + + return nil, nil + }) + if err != nil { + return err + } + + // Notify waiting RPCs. + for i, e := range entries { + e.ch <- entryIndexes[i] + } + // Empty out the entries list so the defered error path doesn't try to notify them. + entries = nil + + return nil +} + +type checkpoint struct { + ID int64 `db:"id"` + MTCLogID string `db:"mtcLogID"` + MTCASignature []byte `db:"mtcaSignature"` + MirrorID string `db:"mirrorID"` + MirrorSignature []byte `db:"mirrorSignature"` + TreeSize int64 `db:"treeSize"` + RootHash []byte `db:"rootHash"` +} + +func (c *checkpoint) valid() error { + if len(c.MTCLogID) == 0 { + return errors.New("MTCLogID is empty") + } + if c.TreeSize == 0 { + return errors.New("TreeSize is 0") + } + if len(c.RootHash) == 0 { + return errors.New("RootHash is empty") + } + if len(c.RootHash) != sha256.Size { + return fmt.Errorf("RootHash is %d bytes", len(c.RootHash)) + } + + return nil +} + +func (c *checkpoint) sequencingReady() bool { + return len(c.MTCASignature) > 0 && len(c.MirrorSignature) > 0 +} + +// String returns a string that is reasonable to print in logs, omitting the (large) signatures. +func (c *checkpoint) String() string { + caSig := "empty" + if len(c.MTCASignature) > 0 { + caSig = "non-empty" + } + mirrorSig := "empty" + if len(c.MirrorSignature) > 0 { + mirrorSig = "non-empty" + } + return fmt.Sprintf("ID:%d MTCLogID:%s MTCASignature:%s MirrorID:%s MirrorSignature:%s TreeSize:%d RootHash:%x", + c.ID, c.MTCLogID, caSig, c.MirrorID, mirrorSig, c.TreeSize, c.RootHash) +} + +func (m *mtca) latest(ctx context.Context) (*checkpoint, error) { + var latestCheckpoint checkpoint + err := m.db.SelectOne(ctx, &latestCheckpoint, + `SELECT id, checkpoints.mtcLogID, mtcaSignature, mirrorID, + mirrorSignature, treeSize, rootHash + FROM latestCheckpoint JOIN checkpoints + USING(id) + WHERE latestCheckpoint.mtcLogID = ? AND + checkpoints.mtcLogID = ?`, + m.mtcLogID(), + m.mtcLogID()) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, fmt.Errorf("getting latest checkpoint for %q: issuance log DB is not initialized", m.mtcLogID()) + } + return nil, fmt.Errorf("getting latest checkpoint for %q: %w", m.mtcLogID(), err) + } + + return &latestCheckpoint, nil +} + +// signCheckpoint signs the checkpoint contents and returns the signature bytes. +func (m *mtca) signCheckpoint(c *checkpoint) ([]byte, error) { + err := c.valid() + if err != nil { + return nil, fmt.Errorf("validating checkpoint: %s", err) + } + + if len(c.MTCASignature) > 0 { + return nil, errors.New("already MTCA-signed") + } + if len(c.MirrorSignature) > 0 { + return nil, errors.New("already mirror-signed") + } + + message := cosigned.Message{ + CosignerName: fmt.Sprintf("oid/1.3.6.1.4.1.%s", m.mtcaID), + Timestamp: 0, + LogOrigin: fmt.Sprintf("oid/1.3.6.1.4.1.%s", m.mtcLogID()), + Start: 0, + End: uint64(c.TreeSize), + SubtreeHash: [32]byte(c.RootHash), + } + + marshaled, err := message.Marshal() + if err != nil { + return nil, err + } + + return m.issuer.Signer.Sign(nil, marshaled, nil) } diff --git a/mtca/mtca_test.go b/mtca/mtca_test.go new file mode 100644 index 00000000000..bd07869f927 --- /dev/null +++ b/mtca/mtca_test.go @@ -0,0 +1,281 @@ +//go:build go1.27 + +package mtca + +import ( + "bytes" + "context" + "crypto/mldsa" + "database/sql" + "encoding/hex" + "fmt" + "strings" + "testing" + + "github.com/jmhodges/clock" + "github.com/letsencrypt/borp" + corepb "github.com/letsencrypt/boulder/core/proto" + "github.com/letsencrypt/boulder/issuance" + blog "github.com/letsencrypt/boulder/log" + "github.com/letsencrypt/boulder/mtca/proto" + "github.com/letsencrypt/boulder/test/vars" + "github.com/letsencrypt/boulder/trees/cosigned" +) + +func TestPool(t *testing.T) { + p := &pool{maxSize: 20} + for i := 0; i < 20; i++ { + err := p.append(entry{}) + if err != nil { + t.Fatal(err) + } + } + + err := p.append(entry{}) + if err == nil { + t.Errorf("append to full pool: got nil, want err") + } + + length := p.len() + if length != 20 { + t.Errorf("p.len(): got %d, want 20", length) + } + + entries := p.take() + if len(entries) != 20 { + t.Errorf("p.take(): got %d entries, want 20", len(entries)) + } + + length = p.len() + if length != 0 { + t.Errorf("p.len(): got %d after take, want 0", length) + } +} + +func TestCheckpointValid(t *testing.T) { + type testCase struct { + name string + value checkpoint + } + + rootHash := [32]byte{} + + testCases := []testCase{ + {"no ID", checkpoint{MTCLogID: "TestLog", TreeSize: 9, RootHash: rootHash[:]}}, + {"no MTCLogID", checkpoint{ID: 7, TreeSize: 9, RootHash: rootHash[:]}}, + {"no TreeSize", checkpoint{ID: 7, MTCLogID: "TestLog", RootHash: rootHash[:]}}, + {"short RootHash", checkpoint{ID: 7, MTCLogID: "TestLog", TreeSize: 9, RootHash: rootHash[:4]}}, + {"no RootHash", checkpoint{ID: 7, MTCLogID: "TestLog", TreeSize: 9}}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + err := tc.value.valid() + if err == nil { + t.Errorf("checkpoint.valid(): got nil, want error") + } + }) + } + + goodCheckpoint := checkpoint{ + ID: 7, + MTCLogID: "TestLog", + TreeSize: 9, + RootHash: rootHash[:], + } + + err := goodCheckpoint.valid() + if err != nil { + t.Errorf("goodCheckpoint.valid(): got %q, want no error", err) + } +} + +// setup returns a working mtca plus a cleanup function, or an error. +func setup() (*mtca, func(), error) { + issuer, err := issuance.LoadIssuer(issuance.IssuerConfig{ + Profiles: []string{}, + IssuerURL: "http://ignored.letsencrypt.org", + CRLURLBase: "http://ignored.letsencrypt.org/", + CRLShards: 1, + Location: issuance.IssuerLoc{ + File: "../test/certs/mtpki/mtca1.key.pem", + CertFile: "../test/certs/mtpki/mtca1.cert.pem", + }, + }, clock.NewFake()) + if err != nil { + return nil, nil, err + } + + db, err := sql.Open("mysql", vars.DBConnMTCMeta_44947_4_1_0_44FullPerms) + if err != nil { + return nil, nil, err + } + dbMap := &borp.DbMap{Db: db, Dialect: borp.MySQLDialect{}} + truncateTables(db) + + logger := blog.NewMock() + + mtca, err := New(issuer, dbMap, logger) + if err != nil { + return nil, nil, err + } + + mtca.InitLog(context.Background()) + + cleanup := func() { + truncateTables(db) + } + + return mtca, cleanup, nil +} + +func truncateTables(db *sql.DB) { + db.Exec("TRUNCATE TABLE checkpoints") + db.Exec("TRUNCATE TABLE latestCheckpoint") +} + +func TestSequence(t *testing.T) { + mtca, cleanup, err := setup() + if err != nil { + t.Fatalf("setting up mtca: %s", err) + } + defer cleanup() + + err = mtca.sequence(t.Context()) + if err == nil { + t.Fatalf("sequencing with an unready checkpoint: got nil error, want error") + } + if !strings.Contains(err.Error(), "not ready") { + t.Errorf("sequencing with an unready checkpoint: expected 'not ready', got %q", err) + } + + // Fake publication + latest, err := mtca.latest(t.Context()) + if err != nil { + t.Fatalf("getting latest: %s", err) + } + latest.MirrorID = "fake" + latest.MirrorSignature = []byte("fake") + + _, err = mtca.db.Update(t.Context(), latest) + if err != nil { + t.Fatalf("updating checkpoint with fake mirror signature: %s", err) + } + + err = mtca.sequence(t.Context()) + if err != nil { + t.Fatalf("sequencing with ready checkpoint and empty pool: %s", err) + } + + mtca.pool.maxSize = 5 + + type result struct { + *proto.IssueResponse + error + } + ch := make(chan result) + for i := 0; i < 6; i++ { + go func() { + resp, err := mtca.Issue(t.Context(), &proto.IssueRequest{ + Pubkey: []byte("abc"), + Identifiers: []*corepb.Identifier{ + {Type: "dns", Value: "example.com"}, + }, + Profile: "mtcExample", + }) + ch <- result{IssueResponse: resp, error: err} + }() + } + + err = mtca.sequence(t.Context()) + if err != nil { + t.Fatalf("sequencing with waiting entries: %s", err) + } + + seenIDs := map[int64]bool{} + var seenError error + for i := 0; i < 6; i++ { + res := <-ch + if res.error != nil { + if seenError != nil { + t.Errorf("too many errors: %s", res.error) + } + seenError = res.error + continue + } + if seenIDs[res.MtcEntryIndex] { + t.Errorf("entryIndex %d seen twice", res.MtcEntryIndex) + } + seenIDs[res.MtcEntryIndex] = true + } + + if seenError == nil { + t.Errorf("putting 6 entries in a pool of size 5: expected error, got none") + } + if !strings.Contains(seenError.Error(), "pool is full") { + t.Errorf("putting 6 entries in a pool of size 5: expected 'pool is full', got %q", seenError) + } + + latest, err = mtca.latest(t.Context()) + if err != nil { + t.Fatalf("getting latest: %s", err) + } + + verify(t, mtca, latest) +} + +func TestInitLog(t *testing.T) { + mtca, cleanup, err := setup() + if err != nil { + t.Fatalf("setting up mtca: %s", err) + } + defer cleanup() + + // InitLog is called once by setup. A second time should fail. + err = mtca.InitLog(t.Context()) + if err == nil { + t.Errorf("second InitLog: got nil error, want error") + } + + latest, err := mtca.latest(t.Context()) + if err != nil { + t.Fatalf("getting latest: %s", err) + } + + if latest.TreeSize != 1 { + t.Errorf("just-initialized log: got TreeSize %d, want 1", latest.TreeSize) + } + expected, _ := hex.DecodeString("8855508aade16ec573d21e6a485dfd0a7624085c1a14b5ecdd6485de0c6839a4") + if !bytes.Equal(latest.RootHash, expected) { + t.Errorf("just-initialized log: got RootHash %x, want %x", latest.RootHash, expected) + } + + verify(t, mtca, latest) +} + +func verify(t *testing.T, mtca *mtca, checkpoint *checkpoint) { + t.Helper() + message := cosigned.Message{ + CosignerName: fmt.Sprintf("oid/1.3.6.1.4.1.%s", mtca.mtcaID), + Timestamp: 0, + LogOrigin: fmt.Sprintf("oid/1.3.6.1.4.1.%s", mtca.mtcLogID()), + Start: 0, + End: uint64(checkpoint.TreeSize), + SubtreeHash: [32]byte(checkpoint.RootHash), + } + + marshaled, err := message.Marshal() + if err != nil { + t.Fatalf("marshaling cosigned.Message: %s", err) + } + + pubkey, ok := mtca.issuer.Signer.Public().(*mldsa.PublicKey) + if !ok { + t.Fatalf("issuer pubkey: got %T, want %T", mtca.issuer.Signer.Public(), &mldsa.PublicKey{}) + } + + err = mldsa.Verify(pubkey, marshaled, checkpoint.MTCASignature, nil) + if err != nil { + t.Errorf("verifying MTCASignature: %s", err) + } +} diff --git a/sa/db/01-mtcmeta_44947_4_1_0_44.sql b/sa/db/01-mtcmeta_44947_4_1_0_44.sql index 425d4961f09..266ecbf4cd0 100644 --- a/sa/db/01-mtcmeta_44947_4_1_0_44.sql +++ b/sa/db/01-mtcmeta_44947_4_1_0_44.sql @@ -1,5 +1,17 @@ +-- This schema is written to work either in a DB-per-issuance-log configuration (current MariaDB), +-- or optionally as a Vitess sharded keyspace where each issuance log is in a separate shard. USE mtcmeta_44947_4_1_0_44; +-- latestCheckpoint is a single-row table pointing to the id of the latest checkpoints. +-- Its single row is locked with SELECT ... FOR UPDATE before signing happens, and then +-- updated after signing happens. +CREATE TABLE `latestCheckpoint` ( + `id` bigint(20) NOT NULL, + -- ASCII-format OID relative to 1.3.6.1.4.1 + `mtcLogID` varchar(255) NOT NULL, + PRIMARY KEY (`mtcLogID`) +); + CREATE TABLE `checkpoints` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, -- ASCII-format OID relative to 1.3.6.1.4.1 @@ -8,7 +20,7 @@ CREATE TABLE `checkpoints` ( -- This is redundant with the database/keyspace name and will be used for extra checks to ensure -- configuration errors can't result in using the wrong database/keyspace. `mtcLogID` varchar(255) NOT NULL, - `mtcaSignature` mediumblob NOT NULL, + `mtcaSignature` mediumblob, -- For simplicity we start out with a hardcoded assumption of one mirror signature, -- the planned CQRP requirement. If requirements increase we can add more fields. -- `mirrorID` is an ASCII-format OID relative to 1.3.6.1.4.1. diff --git a/sa/db/02-users_next.sql b/sa/db/02-users_next.sql index 986a7110b59..90b83ff0f66 100644 --- a/sa/db/02-users_next.sql +++ b/sa/db/02-users_next.sql @@ -98,9 +98,14 @@ GRANT ALL PRIVILEGES ON * to 'test_setup'@'%'; USE mtcmeta_44947_4_1_0_44; CREATE USER IF NOT EXISTS 'mtpublisher'@'%'; +CREATE USER IF NOT EXISTS 'mtca'@'%'; -- MTPublisher stub: reads checkpoints awaiting a cosignature and writes one. GRANT SELECT,UPDATE ON checkpoints TO 'mtpublisher'@'%'; +-- MTCA +GRANT SELECT,INSERT,UPDATE ON checkpoints TO 'mtca'@'%'; +GRANT SELECT,INSERT,UPDATE ON latestCheckpoint TO 'mtca'@'%'; + -- Test setup and teardown GRANT ALL PRIVILEGES ON * to 'test_setup'@'%'; diff --git a/sa/vtschema/mtcmeta_44947_4_1_0_44/vschema.json b/sa/vtschema/mtcmeta_44947_4_1_0_44/vschema.json index a7e1d84d71a..67aaf2e440e 100644 --- a/sa/vtschema/mtcmeta_44947_4_1_0_44/vschema.json +++ b/sa/vtschema/mtcmeta_44947_4_1_0_44/vschema.json @@ -1,12 +1,13 @@ { - "sharded":true, + "sharded": true, "vindexes": { "xxhash": { "type": "xxhash" } }, "tables": { - "checkpoints": { "column_vindexes": [ { "column": "id", "name": "xxhash" } ] }, - "landmarks": { "column_vindexes": [ { "column": "id", "name": "xxhash" } ] } + "checkpoints": { "column_vindexes": [ { "column": "mtcLogID", "name": "xxhash" } ] }, + "latestCheckpoint": { "column_vindexes": [ { "column": "mtcLogID", "name": "xxhash" } ] }, + "landmarks": { "column_vindexes": [ { "column": "mtcLogID", "name": "xxhash" } ] } } } diff --git a/test/certs/genmtpki/genmtpki.go b/test/certs/genmtpki/genmtpki.go index a72ebf407ee..e71f6d3b7d3 100644 --- a/test/certs/genmtpki/genmtpki.go +++ b/test/certs/genmtpki/genmtpki.go @@ -112,10 +112,10 @@ func mtcaSubject() pkix.Name { // idRDNATrustAnchorID := asn1.ObjectIdentifier{ 1, 3, 6, 1, 5, 5, 7, 25 } idRDNATrustAnchorIDExperimental := asn1.ObjectIdentifier{1, 3, 6, 1, 4, 1, 44363, 47, 1} - // https://www.alvestrand.no/objectid/1.3.6.1.4.1.44947.1.html - // 44947.1 is Let's Encrypt; 44947.1.2 will be temporarily for our prototype MTC implementation, with ".1" + // https://letsencrypt.org/docs/oids/ + // 44947 is ISRG; 44947.4.1 will be temporarily for our prototype MTC implementation, with ".1" // representing one CA instance. - mtcaID := "44947.1.2.1" + mtcaID := "44947.4.1" attributes := []pkix.AttributeTypeAndValue{ { Type: idRDNATrustAnchorIDExperimental, diff --git a/test/config-next/mtca.json b/test/config-next/mtca.json index 679125a63e8..4303fb9ed3e 100644 --- a/test/config-next/mtca.json +++ b/test/config-next/mtca.json @@ -1,5 +1,9 @@ { "mtca": { + "sequencingPeriod": "100ms", + "db": { + "dbConnectFile": "test/secrets/mtca1_dburl" + }, "tls": { "caCertFile": "test/certs/ipki/minica.pem", "certFile": "test/certs/ipki/mtca.boulder/cert.pem", @@ -31,7 +35,7 @@ } }, "syslog": { - "stdoutlevel": 4, + "stdoutlevel": 7, "sysloglevel": -1 }, "openTelemetry": { diff --git a/test/config-next/proxysql/mtca1_dburl b/test/config-next/proxysql/mtca1_dburl new file mode 100644 index 00000000000..b3c930ad2e3 --- /dev/null +++ b/test/config-next/proxysql/mtca1_dburl @@ -0,0 +1 @@ +mtca@tcp(boulder-proxysql:6033)/mtcmeta_44947_4_1_0_44 diff --git a/test/config-next/vitess/mtca1_dburl b/test/config-next/vitess/mtca1_dburl new file mode 100644 index 00000000000..117eb22c209 --- /dev/null +++ b/test/config-next/vitess/mtca1_dburl @@ -0,0 +1 @@ +mtca@tcp(boulder-vitess:33577)/mtcmeta_44947_4_1_0_44 diff --git a/test/config/proxysql/mtca1_dburl b/test/config/proxysql/mtca1_dburl new file mode 100644 index 00000000000..b3c930ad2e3 --- /dev/null +++ b/test/config/proxysql/mtca1_dburl @@ -0,0 +1 @@ +mtca@tcp(boulder-proxysql:6033)/mtcmeta_44947_4_1_0_44 diff --git a/test/config/vitess/mtca1_dburl b/test/config/vitess/mtca1_dburl new file mode 100644 index 00000000000..117eb22c209 --- /dev/null +++ b/test/config/vitess/mtca1_dburl @@ -0,0 +1 @@ +mtca@tcp(boulder-vitess:33577)/mtcmeta_44947_4_1_0_44 diff --git a/test/entrypoint.sh b/test/entrypoint.sh index 741805f7b99..396065231ef 100755 --- a/test/entrypoint.sh +++ b/test/entrypoint.sh @@ -16,6 +16,7 @@ DB_URL_FILES=( incidents_dburl incidents_admin_dburl mtpublisher_dburl + mtca1_dburl revoker_dburl sa_dburl sa_ro_dburl diff --git a/test/mtca/README.md b/test/mtca/README.md new file mode 100644 index 00000000000..f1d9807af03 --- /dev/null +++ b/test/mtca/README.md @@ -0,0 +1,6 @@ +This directory contains a handful of utility scripts for manually interacting +with boulder-mtca. + + - init.sh: initialize the demo log SQL DB. + - clear.sh: clear the demo log SQL DB. + - start.sh: run boulder-mtca as a single component. diff --git a/test/mtca/clear.sh b/test/mtca/clear.sh new file mode 100755 index 00000000000..6edda5abeec --- /dev/null +++ b/test/mtca/clear.sh @@ -0,0 +1,5 @@ +#!/bin/bash +set -feuxo pipefail +mysql -h boulder-mariadb -u root -D mtcmeta_44947_4_1_0_44 \ + -e "TRUNCATE TABLE checkpoints; TRUNCATE TABLE latestCheckpoint; TRUNCATE TABLE landmarks" + diff --git a/test/mtca/start.sh b/test/mtca/start.sh new file mode 100755 index 00000000000..ef6c94967f5 --- /dev/null +++ b/test/mtca/start.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash + +set -feuxo pipefail + +make GO=gotip GO_BUILD_FLAGS= +exec ./bin/boulder boulder-mtca -config test/config-next/mtca.json -addr :9396 "$@" diff --git a/test/proxysql/proxysql.cnf b/test/proxysql/proxysql.cnf index 16001f98546..f9ee171a403 100644 --- a/test/proxysql/proxysql.cnf +++ b/test/proxysql/proxysql.cnf @@ -103,6 +103,9 @@ mysql_users = }, { username = "mtpublisher"; + }, + { + username = "mtca"; } ); mysql_query_rules = diff --git a/test/startservers.py b/test/startservers.py index c87354005c9..1b612f77c3a 100644 --- a/test/startservers.py +++ b/test/startservers.py @@ -149,7 +149,8 @@ SERVICES.extend([ Service('boulder-mtca-1', 8010, 9396, 'mtca.boulder', - ('./bin/boulder', 'boulder-mtca', '--config', os.path.join(config_dir, 'mtca.json'), '--addr', ':9396', '--debug-addr', ':8010'), + ('./bin/boulder', 'boulder-mtca', '--config', os.path.join(config_dir, 'mtca.json'), '--addr', ':9396', '--debug-addr', ':8010', + '-init-log-for-test'), None), Service('boulder-mtpublisher-1', 8025, None, None, From 62d8bada033e99356451484fd90a6247660c3edb Mon Sep 17 00:00:00 2001 From: Jacob Hoffman-Andrews Date: Wed, 8 Jul 2026 09:35:06 -0700 Subject: [PATCH 2/7] typo --- mtca/mtca.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mtca/mtca.go b/mtca/mtca.go index 4b2878ac4f9..4c78f4a73c6 100644 --- a/mtca/mtca.go +++ b/mtca/mtca.go @@ -422,7 +422,7 @@ func (m *mtca) sequence(ctx context.Context) error { for i, e := range entries { e.ch <- entryIndexes[i] } - // Empty out the entries list so the defered error path doesn't try to notify them. + // Empty out the entries list so the deferred error path doesn't try to notify them. entries = nil return nil From 1b10613c8f84d294a614fb48d972edd62a31269b Mon Sep 17 00:00:00 2001 From: Jacob Hoffman-Andrews Date: Tue, 14 Jul 2026 12:55:38 -0700 Subject: [PATCH 3/7] review feedback --- cmd/boulder-mtca/main.go | 13 +++++++++++-- mtca/mtca.go | 2 ++ mtca/mtca_test.go | 16 +++++++++++----- sa/db/01-mtcmeta_44947_4_1_0_44.sql | 14 ++++++++++---- test/config-next/mtca.json | 2 +- test/mtca/README.md | 3 +-- test/mtca/clear.sh | 1 + test/startservers.py | 3 +-- 8 files changed, 38 insertions(+), 16 deletions(-) diff --git a/cmd/boulder-mtca/main.go b/cmd/boulder-mtca/main.go index dfaa822446d..78f548691e9 100644 --- a/cmd/boulder-mtca/main.go +++ b/cmd/boulder-mtca/main.go @@ -8,6 +8,7 @@ import ( "errors" "flag" "os" + "time" "github.com/jmhodges/clock" @@ -47,6 +48,9 @@ func main() { grpcAddr := flag.String("addr", "", "gRPC listen address override") debugAddr := flag.String("debug-addr", "", "Debug server address override") configFile := flag.String("config", "", "File path to the configuration file for this service") + // We require an explicit flag to initialize a log because this is a rare operation and we want + // to make sure it's intentional. We exit after initializing the log to make sure we don't + // accidentally include `-init-log` in the command intended for general server operation. initLog := flag.Bool("init-log", false, "Initialize log metadata in the database and exit") initLogForTest := flag.Bool("init-log-for-test", false, "For testing: initialize log metadata if not already initialized, then serve") @@ -88,13 +92,18 @@ func main() { mtcaImpl, err := mtca.New(issuer, dbMap, logger) cmd.FailOnError(err, "Building MTCA") + if *initLog && *initLogForTest { + cmd.Fail("only one of -init-log and -init-log-for-test may happen") + } if *initLog { - err = mtcaImpl.InitLog(context.Background()) + ctx, _ := context.WithTimeout(context.Background(), 15*time.Second) + err = mtcaImpl.InitLog(ctx) cmd.FailOnError(err, "Initializing log") return } if *initLogForTest { - err = mtcaImpl.InitLog(context.Background()) + ctx, _ := context.WithTimeout(context.Background(), 15*time.Second) + err = mtcaImpl.InitLog(ctx) if err != nil && !errors.Is(err, mtca.ErrIssuanceLogAlreadyInitialized) { cmd.FailOnError(err, "Initializing MTC log DB for test") } diff --git a/mtca/mtca.go b/mtca/mtca.go index 4c78f4a73c6..c5bb60cfcb8 100644 --- a/mtca/mtca.go +++ b/mtca/mtca.go @@ -236,6 +236,8 @@ func (m *mtca) Issue(ctx context.Context, req *mtcapb.IssueRequest) (*mtcapb.Iss // there are no inflight RPCs from clients, which in turn ensures that we have sequenced everything // had in the pool. func (m *mtca) Loop(ctx context.Context, sequencingPeriod time.Duration) { + // In cmd/boulder-mtca/main.go, the SequencingPeriod field of config is `required`, so this + // shouldn't happen, but fall back on a reasonable value just in case. if sequencingPeriod <= 0 { sequencingPeriod = 100 * time.Millisecond } diff --git a/mtca/mtca_test.go b/mtca/mtca_test.go index bd07869f927..59bebe81c8e 100644 --- a/mtca/mtca_test.go +++ b/mtca/mtca_test.go @@ -10,10 +10,12 @@ import ( "encoding/hex" "fmt" "strings" + "sync" "testing" "github.com/jmhodges/clock" "github.com/letsencrypt/borp" + corepb "github.com/letsencrypt/boulder/core/proto" "github.com/letsencrypt/boulder/issuance" blog "github.com/letsencrypt/boulder/log" @@ -24,12 +26,16 @@ import ( func TestPool(t *testing.T) { p := &pool{maxSize: 20} + var wg sync.WaitGroup for i := 0; i < 20; i++ { - err := p.append(entry{}) - if err != nil { - t.Fatal(err) - } + wg.Go(func() { + err := p.append(entry{}) + if err != nil { + t.Fatal(err) + } + }) } + wg.Wait() err := p.append(entry{}) if err == nil { @@ -139,7 +145,7 @@ func TestSequence(t *testing.T) { if err != nil { t.Fatalf("setting up mtca: %s", err) } - defer cleanup() + t.Cleanup(cleanup) err = mtca.sequence(t.Context()) if err == nil { diff --git a/sa/db/01-mtcmeta_44947_4_1_0_44.sql b/sa/db/01-mtcmeta_44947_4_1_0_44.sql index 266ecbf4cd0..397a25eab72 100644 --- a/sa/db/01-mtcmeta_44947_4_1_0_44.sql +++ b/sa/db/01-mtcmeta_44947_4_1_0_44.sql @@ -2,13 +2,19 @@ -- or optionally as a Vitess sharded keyspace where each issuance log is in a separate shard. USE mtcmeta_44947_4_1_0_44; --- latestCheckpoint is a single-row table pointing to the id of the latest checkpoints. --- Its single row is locked with SELECT ... FOR UPDATE before signing happens, and then --- updated after signing happens. +-- latestCheckpoint contains the latest checkpoint for a specific MTC log ID. +-- +-- In MariaDB, it will only ever contain a single row. In Vitess it could contain a row +-- per mtcLogID and be sharded on mtcLogID. To ensure the "single row per MTC log ID" +-- property, we make mtcLogID a primary key. +-- +-- The row for a given mtcLogID is locked with SELECT ... FOR UPDATE before signing +-- happens, and then updated after signing happens. CREATE TABLE `latestCheckpoint` ( - `id` bigint(20) NOT NULL, -- ASCII-format OID relative to 1.3.6.1.4.1 `mtcLogID` varchar(255) NOT NULL, + -- ID of a checkpoint in the `checkpoints` table. + `id` bigint(20) NOT NULL, PRIMARY KEY (`mtcLogID`) ); diff --git a/test/config-next/mtca.json b/test/config-next/mtca.json index 4303fb9ed3e..394264981cd 100644 --- a/test/config-next/mtca.json +++ b/test/config-next/mtca.json @@ -35,7 +35,7 @@ } }, "syslog": { - "stdoutlevel": 7, + "stdoutlevel": 4, "sysloglevel": -1 }, "openTelemetry": { diff --git a/test/mtca/README.md b/test/mtca/README.md index f1d9807af03..b3eb48a5a4f 100644 --- a/test/mtca/README.md +++ b/test/mtca/README.md @@ -1,6 +1,5 @@ This directory contains a handful of utility scripts for manually interacting with boulder-mtca. - - init.sh: initialize the demo log SQL DB. - - clear.sh: clear the demo log SQL DB. + - clear.sh: clear the demo log SQL DB (only works for MariaDB). - start.sh: run boulder-mtca as a single component. diff --git a/test/mtca/clear.sh b/test/mtca/clear.sh index 6edda5abeec..d040375fd0d 100755 --- a/test/mtca/clear.sh +++ b/test/mtca/clear.sh @@ -1,4 +1,5 @@ #!/bin/bash +# Note: only works for mariadb, not vitess. set -feuxo pipefail mysql -h boulder-mariadb -u root -D mtcmeta_44947_4_1_0_44 \ -e "TRUNCATE TABLE checkpoints; TRUNCATE TABLE latestCheckpoint; TRUNCATE TABLE landmarks" diff --git a/test/startservers.py b/test/startservers.py index 1b612f77c3a..b7795ada850 100644 --- a/test/startservers.py +++ b/test/startservers.py @@ -149,8 +149,7 @@ SERVICES.extend([ Service('boulder-mtca-1', 8010, 9396, 'mtca.boulder', - ('./bin/boulder', 'boulder-mtca', '--config', os.path.join(config_dir, 'mtca.json'), '--addr', ':9396', '--debug-addr', ':8010', - '-init-log-for-test'), + ('./bin/boulder', 'boulder-mtca', '--config', os.path.join(config_dir, 'mtca.json'), '--addr', ':9396', '--debug-addr', ':8010', '-init-log-for-test'), None), Service('boulder-mtpublisher-1', 8025, None, None, From dd31edf1bff0834385dd42b21b4b9bf63f99e0b5 Mon Sep 17 00:00:00 2001 From: Jacob Hoffman-Andrews Date: Tue, 14 Jul 2026 14:19:45 -0700 Subject: [PATCH 4/7] review feedback --- cmd/boulder-mtca/main.go | 1 + mtca/mtca.go | 199 +++++++++++++++++++++------------------ 2 files changed, 106 insertions(+), 94 deletions(-) diff --git a/cmd/boulder-mtca/main.go b/cmd/boulder-mtca/main.go index 78f548691e9..0c1ba93bc89 100644 --- a/cmd/boulder-mtca/main.go +++ b/cmd/boulder-mtca/main.go @@ -118,6 +118,7 @@ func main() { ctx, cancel := context.WithCancel(context.Background()) // Cancel will be called after start() returns, which happens after GracefulStop() returns. // That means all inflight RPCs will be done, which means the last of the pool has been sequenced. + // GracefulStop() is registered as part of srv.Build() above. defer cancel() go mtcaImpl.Loop(ctx, c.MTCA.SequencingPeriod.Duration) diff --git a/mtca/mtca.go b/mtca/mtca.go index c5bb60cfcb8..4f0c358ec03 100644 --- a/mtca/mtca.go +++ b/mtca/mtca.go @@ -14,33 +14,20 @@ import ( "sync" "time" + "github.com/letsencrypt/borp" + corepb "github.com/letsencrypt/boulder/core/proto" "github.com/letsencrypt/boulder/db" "github.com/letsencrypt/boulder/issuance" + blog "github.com/letsencrypt/boulder/log" mtcapb "github.com/letsencrypt/boulder/mtca/proto" "github.com/letsencrypt/boulder/trees/cosigned" - - "github.com/letsencrypt/borp" - blog "github.com/letsencrypt/boulder/log" ) -var _ mtcapb.MTCAServer = &mtca{} - -func getMTCAID(issuerCert *x509.Certificate) (string, error) { - testingTrustAnchorIDOID := asn1.ObjectIdentifier{1, 3, 6, 1, 4, 1, 44363, 47, 1} - for _, attribute := range issuerCert.Subject.Names { - if attribute.Type.Equal(testingTrustAnchorIDOID) { - mtcaID, ok := attribute.Value.(string) - if !ok { - return "", fmt.Errorf("invalid trust anchor attribute type %T", attribute.Value) - } - return mtcaID, nil - } - } +var ErrIssuanceLogAlreadyInitialized = errors.New("issuance log already initialized") +var ErrCheckpointNotReady = errors.New("not ready - no mirror signature") - return "", fmt.Errorf("issuer subject %q did not contain trust anchor ID OID %q", - issuerCert.Subject, testingTrustAnchorIDOID) -} +var _ mtcapb.MTCAServer = &mtca{} // New creates a new MTCA service. func New(issuer *issuance.Issuer, dbMap *borp.DbMap, logger blog.Logger) (*mtca, error) { @@ -50,71 +37,94 @@ func New(issuer *issuance.Issuer, dbMap *borp.DbMap, logger blog.Logger) (*mtca, } return &mtca{ - log: logger, - db: initDB(dbMap), issuer: issuer, mtcaID: mtcaID, // TODO: collect this from config logNumber: 0, pool: &pool{maxSize: 100}, + + db: initDB(dbMap), + log: logger, }, nil } +type mtca struct { + mtcapb.UnimplementedMTCAServer + + issuer *issuance.Issuer + mtcaID string + logNumber uint16 + + db *db.WrappedMap + log blog.Logger + + pool *pool +} + +func getMTCAID(issuerCert *x509.Certificate) (string, error) { + testingTrustAnchorIDOID := asn1.ObjectIdentifier{1, 3, 6, 1, 4, 1, 44363, 47, 1} + for _, attribute := range issuerCert.Subject.Names { + if attribute.Type.Equal(testingTrustAnchorIDOID) { + mtcaID, ok := attribute.Value.(string) + if !ok { + return "", fmt.Errorf("invalid trust anchor attribute type %T", attribute.Value) + } + return mtcaID, nil + } + } + + return "", fmt.Errorf("issuer subject %q did not contain trust anchor ID OID %q", + issuerCert.Subject, testingTrustAnchorIDOID) +} + func initDB(dbMap *borp.DbMap) *db.WrappedMap { dbMap.AddTableWithName(checkpoint{}, "checkpoints").SetKeys(true, "ID") return db.NewWrappedMap(dbMap) } -var ErrIssuanceLogAlreadyInitialized = errors.New("issuance log already initialized") -var ErrCheckpointNotReady = errors.New("not ready - no mirror signature") - // InitLog creates the database metadata for a new, empty log: one checkpoint and the row // in `latestCheckpoint` that refers to it. Should only be run once in a log's lifetime. func (m *mtca) InitLog(ctx context.Context) error { - _, err := m.latest(ctx) - if err == nil { - return ErrIssuanceLogAlreadyInitialized - } + _, err := db.WithTransaction(ctx, m.db, func(tx db.Executor) (any, error) { + var numLatestCheckpoints int64 + err := tx.SelectOne(ctx, &numLatestCheckpoints, "SELECT COUNT(*) FROM latestCheckpoint WHERE mtcLogID = ?", + m.mtcLogID()) + if err != nil { + return nil, fmt.Errorf("getting latestCheckpoint: %s", err) + } - var numLatestCheckpoints int64 - err = m.db.SelectOne(ctx, &numLatestCheckpoints, "SELECT COUNT(*) FROM latestCheckpoint WHERE mtcLogID = ?", - m.mtcLogID()) - if err != nil { - return fmt.Errorf("getting latestCheckpoint: %s", err) - } + var numCheckpoints int64 + err = tx.SelectOne(ctx, &numCheckpoints, "SELECT COUNT(*) FROM checkpoints WHERE mtcLogID = ?", + m.mtcLogID()) + if err != nil { + return nil, fmt.Errorf("getting checkpoints: %s", err) + } - var numCheckpoints int64 - err = m.db.SelectOne(ctx, &numCheckpoints, "SELECT COUNT(*) FROM checkpoints WHERE mtcLogID = ?", - m.mtcLogID()) - if err != nil { - return fmt.Errorf("getting checkpoints: %s", err) - } + if numCheckpoints > 0 || numLatestCheckpoints > 0 { + if numLatestCheckpoints == 1 { + return nil, ErrIssuanceLogAlreadyInitialized + } - if numCheckpoints > 0 || numLatestCheckpoints > 0 { - return fmt.Errorf("initializing issuance log for %s: already has %d checkpoints and %d latestCheckpoint rows", - m.mtcLogID(), numCheckpoints, numLatestCheckpoints) - } + return nil, fmt.Errorf("initializing issuance log for %s: already has %d checkpoints and %d latestCheckpoint rows", + m.mtcLogID(), numCheckpoints, numLatestCheckpoints) + } - // null_entry has empty extensions and a MerkleTreeCertEntryType of 0. Since extensions can be up to 2^16 long - // there's two bytes of length prefix. Since MerkleTreeCertEntryType can have up to 2^16 values, it's also two bytes. - // All the bytes are zero: empty extensions, null_entry type is enum value zero. - // https://ietf-plants-wg.github.io/merkle-tree-certs/draft-ietf-plants-merkle-tree-certs.html#name-log-entries - // To calculate the Merkle Tree Hash of a single-entry list, we prepend 0x00 (as compared with 0x01 when hashing - // two nodes). So five zeroes total. - // https://www.rfc-editor.org/info/rfc9162/#name-definition-of-the-merkle-tr - nullEntry := []byte{0, 0, 0, 0, 0} - rootHash := sha256.Sum256(nullEntry) - - firstCheckpoint := checkpoint{ - MTCLogID: m.mtcLogID(), - MTCASignature: nil, - MirrorID: "", - MirrorSignature: nil, - TreeSize: 1, - RootHash: rootHash[:], - } + // null_entry has empty extensions and a MerkleTreeCertEntryType of 0. Since extensions can be up to 2^16 long + // there's two bytes of length prefix. Since MerkleTreeCertEntryType can have up to 2^16 values, it's also two bytes. + // All the bytes are zero: empty extensions, null_entry type is enum value zero. + // https://ietf-plants-wg.github.io/merkle-tree-certs/draft-ietf-plants-merkle-tree-certs.html#name-log-entries + // To calculate the Merkle Tree Hash of a single-entry list, we prepend 0x00 (as compared with 0x01 when hashing + // two nodes). So five zeroes total. + // https://www.rfc-editor.org/info/rfc9162/#name-definition-of-the-merkle-tr + nullEntry := []byte{0, 0, 0, 0, 0} + rootHash := sha256.Sum256(nullEntry) + + firstCheckpoint := checkpoint{ + MTCLogID: m.mtcLogID(), + TreeSize: 1, + RootHash: rootHash[:], + } - _, err = db.WithTransaction(ctx, m.db, func(tx db.Executor) (any, error) { sig, err := m.signCheckpoint(&firstCheckpoint) if err != nil { return nil, err @@ -139,7 +149,7 @@ func (m *mtca) InitLog(ctx context.Context) error { return err } - _, err = m.latest(ctx) + _, err = m.latestCheckpoint(ctx) if err != nil { return fmt.Errorf("fetching first checkpoint: %s", err) } @@ -147,33 +157,19 @@ func (m *mtca) InitLog(ctx context.Context) error { return err } -type mtca struct { - mtcapb.UnimplementedMTCAServer - - issuer *issuance.Issuer - mtcaID string - logNumber uint16 - - db *db.WrappedMap - log blog.Logger - - pool *pool - - sequencingMu sync.Mutex +type pool struct { + sync.RWMutex + entries []entry + maxSize int } +// entry represents an entry in the pool, along with a channel to notify a pending RPC. type entry struct { pubkey []byte identifiers []*corepb.Identifier ch chan<- int64 } -type pool struct { - sync.Mutex - entries []entry - maxSize int -} - func (p *pool) take() []entry { p.Lock() defer p.Unlock() @@ -183,8 +179,8 @@ func (p *pool) take() []entry { } func (p *pool) len() int { - p.Lock() - defer p.Unlock() + p.RLock() + defer p.RUnlock() return len(p.entries) } @@ -205,7 +201,11 @@ func (m *mtca) mtcLogID() string { return fmt.Sprintf("%s.0.%d", m.mtcaID, m.logNumber) } +// Issue requests a TBSCertificateLogEntry be issued and returns after it's been sequenced into the log +// and a new checkpoint signed by the CA. It does not wait for a mirror cosignature. func (m *mtca) Issue(ctx context.Context, req *mtcapb.IssueRequest) (*mtcapb.IssueResponse, error) { + // We'll get notification of sequencing on this channel. Buffer it so `sequence()` doesn't + // block if this method has already returned (e.g. due to timeout). ch := make(chan int64, 1) err := m.pool.append(entry{ pubkey: req.Pubkey, @@ -246,6 +246,7 @@ func (m *mtca) Loop(ctx context.Context, sequencingPeriod time.Duration) { since := time.Now() ticker := time.NewTicker(sequencingPeriod) + defer ticker.Stop() for { select { case <-ticker.C: @@ -262,7 +263,7 @@ func (m *mtca) Loop(ctx context.Context, sequencingPeriod time.Duration) { case <-ctx.Done(): poolSize := m.pool.len() if poolSize != 0 { - m.log.Errf("shutting down loop: pool has %d entries left. ungraceful stop?", poolSize) + m.log.Errf("shouldn't happen: pool has %d entries left after Loop() context canceled. ungraceful stop?", poolSize) } return } @@ -278,7 +279,7 @@ func (m *mtca) fakePublisher(ctx context.Context) { for { select { case <-ticker.C: - latest, err := m.latest(ctx) + latest, err := m.latestCheckpoint(ctx) if err != nil { m.log.Errf("getting latest checkpoint for fake publisher: %s", err) continue @@ -298,24 +299,31 @@ func (m *mtca) fakePublisher(ctx context.Context) { } } +// sequence takes all entries from the pool, simulates writing them to tile storage, signs +// and stores a new checkpoint, and notifies waiting RPCs. +// +// If the pool is empty, nothing happens. +// If the pool is non-empty, but the previous checkpoint doesn't have a mirror signature, +// returns an error that wraps ErrCheckpointNotReady (without taking entries from the pool). +// This is expected to be a common occurrence. +// +// Each entry in the pool will get a notification on its channel: either the index at which +// it was sequenced, or -1 if there was an error during sequencing. func (m *mtca) sequence(ctx context.Context) error { if m.pool.len() == 0 { return nil } - latest, err := m.latest(ctx) + latest, err := m.latestCheckpoint(ctx) if err != nil { return err } - if !latest.sequencingReady() { + if !latest.mirrored() { return fmt.Errorf("temporary: checkpoint ID %d (tree size %d): %w", latest.ID, latest.TreeSize, ErrCheckpointNotReady) } - m.sequencingMu.Lock() - defer m.sequencingMu.Unlock() - // Pull the contents of the pool. entries := m.pool.take() if len(entries) == 0 { @@ -430,6 +438,9 @@ func (m *mtca) sequence(ctx context.Context) error { return nil } +// checkpoint represents the database storage of a checkpoint and associated signatures. +// +// For signing, the TreeSize and RootHash fields are incorporated into a `cosigned.Message`. type checkpoint struct { ID int64 `db:"id"` MTCLogID string `db:"mtcLogID"` @@ -457,7 +468,7 @@ func (c *checkpoint) valid() error { return nil } -func (c *checkpoint) sequencingReady() bool { +func (c *checkpoint) mirrored() bool { return len(c.MTCASignature) > 0 && len(c.MirrorSignature) > 0 } @@ -475,7 +486,7 @@ func (c *checkpoint) String() string { c.ID, c.MTCLogID, caSig, c.MirrorID, mirrorSig, c.TreeSize, c.RootHash) } -func (m *mtca) latest(ctx context.Context) (*checkpoint, error) { +func (m *mtca) latestCheckpoint(ctx context.Context) (*checkpoint, error) { var latestCheckpoint checkpoint err := m.db.SelectOne(ctx, &latestCheckpoint, `SELECT id, checkpoints.mtcLogID, mtcaSignature, mirrorID, From bf888cf5d06ebc281db61375b300e1494ea9aef8 Mon Sep 17 00:00:00 2001 From: Jacob Hoffman-Andrews Date: Tue, 14 Jul 2026 15:29:21 -0700 Subject: [PATCH 5/7] split sign / cosignedMessage; add test --- mtca/mtca.go | 28 ++++++++++++++++++++++------ mtca/mtca_test.go | 34 +++++++++++++++++++++++++++++++--- 2 files changed, 53 insertions(+), 9 deletions(-) diff --git a/mtca/mtca.go b/mtca/mtca.go index 4f0c358ec03..d38740a374c 100644 --- a/mtca/mtca.go +++ b/mtca/mtca.go @@ -125,7 +125,12 @@ func (m *mtca) InitLog(ctx context.Context) error { RootHash: rootHash[:], } - sig, err := m.signCheckpoint(&firstCheckpoint) + message, err := m.cosignedMessage(&firstCheckpoint) + if err != nil { + return nil, err + } + + sig, err := m.sign(message) if err != nil { return nil, err } @@ -261,6 +266,10 @@ func (m *mtca) Loop(ctx context.Context, sequencingPeriod time.Duration) { } since = time.Now() case <-ctx.Done(): + // Given the structure of main(), this context will only be cancelled once + // GracefulStop has finished. That means all in-flight RPCs have returned, + // which in turn means that their certificate requests were sequenced (or + // they timed out, in which case emitting this error is appropriate). poolSize := m.pool.len() if poolSize != 0 { m.log.Errf("shouldn't happen: pool has %d entries left after Loop() context canceled. ungraceful stop?", poolSize) @@ -373,6 +382,11 @@ func (m *mtca) sequence(ctx context.Context) error { return err } + message, err := m.cosignedMessage(&newCheckpoint) + if err != nil { + return err + } + _, err = db.WithTransaction(ctx, m.db, func(tx db.Executor) (any, error) { var latestID int64 // Lock the latestCheckpoint to make sure there is no concurrent signer/writer, avoiding signing a split view. @@ -391,7 +405,7 @@ func (m *mtca) sequence(ctx context.Context) error { // Note that we're doing HSM work while holding a database lock. That's intentional; the database lock // is to prevent the possibility of a concurrent signer on the same tree. - sig, err := m.signCheckpoint(&newCheckpoint) + sig, err := m.sign(message) if err != nil { return nil, err } @@ -507,8 +521,7 @@ func (m *mtca) latestCheckpoint(ctx context.Context) (*checkpoint, error) { return &latestCheckpoint, nil } -// signCheckpoint signs the checkpoint contents and returns the signature bytes. -func (m *mtca) signCheckpoint(c *checkpoint) ([]byte, error) { +func (m *mtca) cosignedMessage(c *checkpoint) (*cosigned.Message, error) { err := c.valid() if err != nil { return nil, fmt.Errorf("validating checkpoint: %s", err) @@ -521,15 +534,18 @@ func (m *mtca) signCheckpoint(c *checkpoint) ([]byte, error) { return nil, errors.New("already mirror-signed") } - message := cosigned.Message{ + return &cosigned.Message{ CosignerName: fmt.Sprintf("oid/1.3.6.1.4.1.%s", m.mtcaID), Timestamp: 0, LogOrigin: fmt.Sprintf("oid/1.3.6.1.4.1.%s", m.mtcLogID()), Start: 0, End: uint64(c.TreeSize), SubtreeHash: [32]byte(c.RootHash), - } + }, nil +} +// signCheckpoint signs the checkpoint contents and returns the signature bytes. +func (m *mtca) sign(message *cosigned.Message) ([]byte, error) { marshaled, err := message.Marshal() if err != nil { return nil, err diff --git a/mtca/mtca_test.go b/mtca/mtca_test.go index 59bebe81c8e..fc3c9faa66d 100644 --- a/mtca/mtca_test.go +++ b/mtca/mtca_test.go @@ -6,7 +6,9 @@ import ( "bytes" "context" "crypto/mldsa" + "crypto/x509" "database/sql" + "encoding/base64" "encoding/hex" "fmt" "strings" @@ -156,7 +158,7 @@ func TestSequence(t *testing.T) { } // Fake publication - latest, err := mtca.latest(t.Context()) + latest, err := mtca.latestCheckpoint(t.Context()) if err != nil { t.Fatalf("getting latest: %s", err) } @@ -222,7 +224,7 @@ func TestSequence(t *testing.T) { t.Errorf("putting 6 entries in a pool of size 5: expected 'pool is full', got %q", seenError) } - latest, err = mtca.latest(t.Context()) + latest, err = mtca.latestCheckpoint(t.Context()) if err != nil { t.Fatalf("getting latest: %s", err) } @@ -243,7 +245,7 @@ func TestInitLog(t *testing.T) { t.Errorf("second InitLog: got nil error, want error") } - latest, err := mtca.latest(t.Context()) + latest, err := mtca.latestCheckpoint(t.Context()) if err != nil { t.Fatalf("getting latest: %s", err) } @@ -285,3 +287,29 @@ func verify(t *testing.T, mtca *mtca, checkpoint *checkpoint) { t.Errorf("verifying MTCASignature: %s", err) } } + +func TestGetMTCAID(t *testing.T) { + certBytes, err := base64.StdEncoding.DecodeString(strings.ReplaceAll(` +MIIBRjCB9KADAgECAgF7MAoGCCqGSM49BAMCMBsxGTAXBgorBgEEAYLaSy8BDAk0 +NDk0Ny40LjEwHhcNMjYwNzE0MjIyNjIwWhcNMzYwNzExMjIyNjIwWjAbMRkwFwYK +KwYBBAGC2ksvAQwJNDQ5NDcuNC4xME4wEAYHKoZIzj0CAQYFK4EEACEDOgAERbiP +RTb8x/eav43juNzWZLId2Wl5TzmTsG5iRf+CiB+rn+TXnuUbWDIuIi/kYs3USANm +LUyLxH+jNDAyMA4GA1UdDwEB/wQEAwIBBjAPBgNVHRMBAf8EBTADAQH/MA8GA1Ud +DgQIBAaC3xMBAgEwCgYIKoZIzj0EAwIDQQAwPgIdAMebuq7759hyFC3hjrVUEaXk +2TewRlXg+ohJvFoCHQCTMjnYvLIvTCqF3gZm38+h1iShEgMfMT522d60 +`, "\n", "")) + if err != nil { + t.Fatal(err) + } + + cert, err := x509.ParseCertificate(certBytes) + mtcaID, err := getMTCAID(cert) + if err != nil { + t.Fatal(err) + } + + expected := "44947.4.1" + if mtcaID != expected { + t.Errorf("getMTCAID(): got %s, want %s", mtcaID, expected) + } +} From 8ce8a23de3b447245dc0ebdd97b06e6157340a36 Mon Sep 17 00:00:00 2001 From: Jacob Hoffman-Andrews Date: Wed, 15 Jul 2026 15:44:20 -0700 Subject: [PATCH 6/7] Move sequencingPeriod to constructor --- cmd/boulder-mtca/main.go | 4 ++-- mtca/mtca.go | 22 ++++++++++++---------- mtca/mtca_test.go | 3 ++- 3 files changed, 16 insertions(+), 13 deletions(-) diff --git a/cmd/boulder-mtca/main.go b/cmd/boulder-mtca/main.go index 0c1ba93bc89..4978e173a3f 100644 --- a/cmd/boulder-mtca/main.go +++ b/cmd/boulder-mtca/main.go @@ -89,7 +89,7 @@ func main() { cmd.FailOnError(err, "Opening DB") dbMap := &borp.DbMap{Db: db, Dialect: borp.MySQLDialect{}} - mtcaImpl, err := mtca.New(issuer, dbMap, logger) + mtcaImpl, err := mtca.New(issuer, c.MTCA.SequencingPeriod.Duration, dbMap, logger) cmd.FailOnError(err, "Building MTCA") if *initLog && *initLogForTest { @@ -120,7 +120,7 @@ func main() { // That means all inflight RPCs will be done, which means the last of the pool has been sequenced. // GracefulStop() is registered as part of srv.Build() above. defer cancel() - go mtcaImpl.Loop(ctx, c.MTCA.SequencingPeriod.Duration) + go mtcaImpl.Loop(ctx) cmd.FailOnError(start(), "MTCA gRPC service failed") } diff --git a/mtca/mtca.go b/mtca/mtca.go index d38740a374c..9f867dd1055 100644 --- a/mtca/mtca.go +++ b/mtca/mtca.go @@ -30,12 +30,16 @@ var ErrCheckpointNotReady = errors.New("not ready - no mirror signature") var _ mtcapb.MTCAServer = &mtca{} // New creates a new MTCA service. -func New(issuer *issuance.Issuer, dbMap *borp.DbMap, logger blog.Logger) (*mtca, error) { +func New(issuer *issuance.Issuer, sequencingPeriod time.Duration, dbMap *borp.DbMap, logger blog.Logger) (*mtca, error) { mtcaID, err := getMTCAID(issuer.Cert.Certificate) if err != nil { return nil, err } + if sequencingPeriod == 0 { + return nil, errors.New("sequencingPeriod must be non-zero") + } + return &mtca{ issuer: issuer, mtcaID: mtcaID, @@ -43,6 +47,8 @@ func New(issuer *issuance.Issuer, dbMap *borp.DbMap, logger blog.Logger) (*mtca, logNumber: 0, pool: &pool{maxSize: 100}, + sequencingPeriod: sequencingPeriod, + db: initDB(dbMap), log: logger, }, nil @@ -55,6 +61,8 @@ type mtca struct { mtcaID string logNumber uint16 + sequencingPeriod time.Duration + db *db.WrappedMap log blog.Logger @@ -240,17 +248,11 @@ func (m *mtca) Issue(ctx context.Context, req *mtcapb.IssueRequest) (*mtcapb.Iss // At process shutdown, this context should be canceled _after_ GracefulStop returns. That ensures // there are no inflight RPCs from clients, which in turn ensures that we have sequenced everything // had in the pool. -func (m *mtca) Loop(ctx context.Context, sequencingPeriod time.Duration) { - // In cmd/boulder-mtca/main.go, the SequencingPeriod field of config is `required`, so this - // shouldn't happen, but fall back on a reasonable value just in case. - if sequencingPeriod <= 0 { - sequencingPeriod = 100 * time.Millisecond - } - +func (m *mtca) Loop(ctx context.Context) { go m.fakePublisher(ctx) since := time.Now() - ticker := time.NewTicker(sequencingPeriod) + ticker := time.NewTicker(m.sequencingPeriod) defer ticker.Stop() for { select { @@ -259,7 +261,7 @@ func (m *mtca) Loop(ctx context.Context, sequencingPeriod time.Duration) { if err != nil { if !errors.Is(err, ErrCheckpointNotReady) { m.log.Errf("sequencing: %s", err) - } else if time.Since(since) > 10*sequencingPeriod { + } else if time.Since(since) > 10*m.sequencingPeriod { m.log.Errf("after %s: %s", time.Since(since).Round(time.Millisecond), err) } continue diff --git a/mtca/mtca_test.go b/mtca/mtca_test.go index fc3c9faa66d..064beb662a6 100644 --- a/mtca/mtca_test.go +++ b/mtca/mtca_test.go @@ -14,6 +14,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/jmhodges/clock" "github.com/letsencrypt/borp" @@ -123,7 +124,7 @@ func setup() (*mtca, func(), error) { logger := blog.NewMock() - mtca, err := New(issuer, dbMap, logger) + mtca, err := New(issuer, 100*time.Millisecond, dbMap, logger) if err != nil { return nil, nil, err } From 941b30a3c9866f89efa6c8a18bc2a062c0048d58 Mon Sep 17 00:00:00 2001 From: Jacob Hoffman-Andrews Date: Thu, 16 Jul 2026 10:27:58 -0700 Subject: [PATCH 7/7] review feedback --- cmd/boulder-mtca/main.go | 19 ++- mtca/mtca.go | 30 +++-- mtca/mtca_test.go | 175 ++++++++++++++-------------- sa/db/01-mtcmeta_44947_4_1_0_44.sql | 2 - 4 files changed, 121 insertions(+), 105 deletions(-) diff --git a/cmd/boulder-mtca/main.go b/cmd/boulder-mtca/main.go index 4978e173a3f..e1c1e49bef0 100644 --- a/cmd/boulder-mtca/main.go +++ b/cmd/boulder-mtca/main.go @@ -8,6 +8,7 @@ import ( "errors" "flag" "os" + "sync" "time" "github.com/jmhodges/clock" @@ -96,13 +97,15 @@ func main() { cmd.Fail("only one of -init-log and -init-log-for-test may happen") } if *initLog { - ctx, _ := context.WithTimeout(context.Background(), 15*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() err = mtcaImpl.InitLog(ctx) cmd.FailOnError(err, "Initializing log") return } if *initLogForTest { - ctx, _ := context.WithTimeout(context.Background(), 15*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() err = mtcaImpl.InitLog(ctx) if err != nil && !errors.Is(err, mtca.ErrIssuanceLogAlreadyInitialized) { cmd.FailOnError(err, "Initializing MTC log DB for test") @@ -115,12 +118,18 @@ func main() { start, err := srv.Build(tlsConfig, scope, clk) cmd.FailOnError(err, "Unable to setup MTCA gRPC server") - ctx, cancel := context.WithCancel(context.Background()) // Cancel will be called after start() returns, which happens after GracefulStop() returns. // That means all inflight RPCs will be done, which means the last of the pool has been sequenced. // GracefulStop() is registered as part of srv.Build() above. - defer cancel() - go mtcaImpl.Loop(ctx) + ctx, cancel := context.WithCancel(context.Background()) + var loopWG sync.WaitGroup + loopWG.Go(func() { + mtcaImpl.Loop(ctx) + }) + defer func() { + cancel() + loopWG.Wait() + }() cmd.FailOnError(start(), "MTCA gRPC service failed") } diff --git a/mtca/mtca.go b/mtca/mtca.go index 9f867dd1055..49ac955cf1d 100644 --- a/mtca/mtca.go +++ b/mtca/mtca.go @@ -44,7 +44,7 @@ func New(issuer *issuance.Issuer, sequencingPeriod time.Duration, dbMap *borp.Db issuer: issuer, mtcaID: mtcaID, // TODO: collect this from config - logNumber: 0, + logNumber: 44, pool: &pool{maxSize: 100}, sequencingPeriod: sequencingPeriod, @@ -63,6 +63,9 @@ type mtca struct { sequencingPeriod time.Duration + // TODO: factor our sa.InitWrappedDb() so we get metrics and other goodies. + // TODO: decide whether we want to route this through the SA or an SA-like object, + // or keep a direct DB connection from the MTCA. db *db.WrappedMap log blog.Logger @@ -117,8 +120,8 @@ func (m *mtca) InitLog(ctx context.Context) error { m.mtcLogID(), numCheckpoints, numLatestCheckpoints) } - // null_entry has empty extensions and a MerkleTreeCertEntryType of 0. Since extensions can be up to 2^16 long - // there's two bytes of length prefix. Since MerkleTreeCertEntryType can have up to 2^16 values, it's also two bytes. + // null_entry has empty extensions and a MTCLogEntryType of 0. Since extensions can be up to 2^16 long + // there's two bytes of length prefix. Since MTCLogEntryType can have up to 2^16 values, it's also two bytes. // All the bytes are zero: empty extensions, null_entry type is enum value zero. // https://ietf-plants-wg.github.io/merkle-tree-certs/draft-ietf-plants-merkle-tree-certs.html#name-log-entries // To calculate the Merkle Tree Hash of a single-entry list, we prepend 0x00 (as compared with 0x01 when hashing @@ -287,6 +290,7 @@ func (m *mtca) Loop(ctx context.Context) { // TODO: remove once a real publisher is available in integration. func (m *mtca) fakePublisher(ctx context.Context) { ticker := time.NewTicker(37 * time.Millisecond) + defer ticker.Stop() for { select { case <-ticker.C: @@ -371,6 +375,11 @@ func (m *mtca) sequence(ctx context.Context) error { RootHash: newRootHash[:], } + message, err := m.cosignedMessage(&newCheckpoint) + if err != nil { + return err + } + // Precommit to the new checkpoint. This will allow us to do recovery if we crash between signing // the new checkpoint and writing it to the database. // @@ -384,11 +393,6 @@ func (m *mtca) sequence(ctx context.Context) error { return err } - message, err := m.cosignedMessage(&newCheckpoint) - if err != nil { - return err - } - _, err = db.WithTransaction(ctx, m.db, func(tx db.Executor) (any, error) { var latestID int64 // Lock the latestCheckpoint to make sure there is no concurrent signer/writer, avoiding signing a split view. @@ -503,8 +507,8 @@ func (c *checkpoint) String() string { } func (m *mtca) latestCheckpoint(ctx context.Context) (*checkpoint, error) { - var latestCheckpoint checkpoint - err := m.db.SelectOne(ctx, &latestCheckpoint, + var latest checkpoint + err := m.db.SelectOne(ctx, &latest, `SELECT id, checkpoints.mtcLogID, mtcaSignature, mirrorID, mirrorSignature, treeSize, rootHash FROM latestCheckpoint JOIN checkpoints @@ -520,7 +524,7 @@ func (m *mtca) latestCheckpoint(ctx context.Context) (*checkpoint, error) { return nil, fmt.Errorf("getting latest checkpoint for %q: %w", m.mtcLogID(), err) } - return &latestCheckpoint, nil + return &latest, nil } func (m *mtca) cosignedMessage(c *checkpoint) (*cosigned.Message, error) { @@ -546,7 +550,9 @@ func (m *mtca) cosignedMessage(c *checkpoint) (*cosigned.Message, error) { }, nil } -// signCheckpoint signs the checkpoint contents and returns the signature bytes. +// sign marshals a *cosigned.Message and signs its bytes. +// +// Returns the signature bytes. func (m *mtca) sign(message *cosigned.Message) ([]byte, error) { marshaled, err := message.Marshal() if err != nil { diff --git a/mtca/mtca_test.go b/mtca/mtca_test.go index 064beb662a6..0bb96eef6de 100644 --- a/mtca/mtca_test.go +++ b/mtca/mtca_test.go @@ -10,6 +10,7 @@ import ( "database/sql" "encoding/base64" "encoding/hex" + "errors" "fmt" "strings" "sync" @@ -27,6 +28,45 @@ import ( "github.com/letsencrypt/boulder/trees/cosigned" ) +// setup returns a working mtca plus a cleanup function, or an error. +func setup() (*mtca, func(), error) { + issuer, err := issuance.LoadIssuer(issuance.IssuerConfig{ + Profiles: []string{}, + IssuerURL: "http://ignored.letsencrypt.org", + CRLURLBase: "http://ignored.letsencrypt.org/", + CRLShards: 1, + Location: issuance.IssuerLoc{ + File: "../test/certs/mtpki/mtca1.key.pem", + CertFile: "../test/certs/mtpki/mtca1.cert.pem", + }, + }, clock.NewFake()) + if err != nil { + return nil, nil, err + } + + db, err := sql.Open("mysql", vars.DBConnMTCMeta_44947_4_1_0_44FullPerms) + if err != nil { + return nil, nil, err + } + dbMap := &borp.DbMap{Db: db, Dialect: borp.MySQLDialect{}} + truncateTables(db) + + logger := blog.NewMock() + + mtca, err := New(issuer, 100*time.Millisecond, dbMap, logger) + if err != nil { + return nil, nil, err + } + + mtca.InitLog(context.Background()) + + cleanup := func() { + truncateTables(db) + } + + return mtca, cleanup, nil +} + func TestPool(t *testing.T) { p := &pool{maxSize: 20} var wg sync.WaitGroup @@ -34,7 +74,7 @@ func TestPool(t *testing.T) { wg.Go(func() { err := p.append(entry{}) if err != nil { - t.Fatal(err) + t.Errorf("appending entry: %s", err) } }) } @@ -70,7 +110,6 @@ func TestCheckpointValid(t *testing.T) { rootHash := [32]byte{} testCases := []testCase{ - {"no ID", checkpoint{MTCLogID: "TestLog", TreeSize: 9, RootHash: rootHash[:]}}, {"no MTCLogID", checkpoint{ID: 7, TreeSize: 9, RootHash: rootHash[:]}}, {"no TreeSize", checkpoint{ID: 7, MTCLogID: "TestLog", RootHash: rootHash[:]}}, {"short RootHash", checkpoint{ID: 7, MTCLogID: "TestLog", TreeSize: 9, RootHash: rootHash[:4]}}, @@ -99,45 +138,6 @@ func TestCheckpointValid(t *testing.T) { } } -// setup returns a working mtca plus a cleanup function, or an error. -func setup() (*mtca, func(), error) { - issuer, err := issuance.LoadIssuer(issuance.IssuerConfig{ - Profiles: []string{}, - IssuerURL: "http://ignored.letsencrypt.org", - CRLURLBase: "http://ignored.letsencrypt.org/", - CRLShards: 1, - Location: issuance.IssuerLoc{ - File: "../test/certs/mtpki/mtca1.key.pem", - CertFile: "../test/certs/mtpki/mtca1.cert.pem", - }, - }, clock.NewFake()) - if err != nil { - return nil, nil, err - } - - db, err := sql.Open("mysql", vars.DBConnMTCMeta_44947_4_1_0_44FullPerms) - if err != nil { - return nil, nil, err - } - dbMap := &borp.DbMap{Db: db, Dialect: borp.MySQLDialect{}} - truncateTables(db) - - logger := blog.NewMock() - - mtca, err := New(issuer, 100*time.Millisecond, dbMap, logger) - if err != nil { - return nil, nil, err - } - - mtca.InitLog(context.Background()) - - cleanup := func() { - truncateTables(db) - } - - return mtca, cleanup, nil -} - func truncateTables(db *sql.DB) { db.Exec("TRUNCATE TABLE checkpoints") db.Exec("TRUNCATE TABLE latestCheckpoint") @@ -149,41 +149,19 @@ func TestSequence(t *testing.T) { t.Fatalf("setting up mtca: %s", err) } t.Cleanup(cleanup) - - err = mtca.sequence(t.Context()) - if err == nil { - t.Fatalf("sequencing with an unready checkpoint: got nil error, want error") - } - if !strings.Contains(err.Error(), "not ready") { - t.Errorf("sequencing with an unready checkpoint: expected 'not ready', got %q", err) - } - - // Fake publication - latest, err := mtca.latestCheckpoint(t.Context()) - if err != nil { - t.Fatalf("getting latest: %s", err) - } - latest.MirrorID = "fake" - latest.MirrorSignature = []byte("fake") - - _, err = mtca.db.Update(t.Context(), latest) - if err != nil { - t.Fatalf("updating checkpoint with fake mirror signature: %s", err) - } - + // An empty pool is a no-op regardless of checkpoint state. err = mtca.sequence(t.Context()) if err != nil { - t.Fatalf("sequencing with ready checkpoint and empty pool: %s", err) + t.Fatalf("sequencing with empty pool: %s", err) } - + // Fill the pool with five concurrent requests. mtca.pool.maxSize = 5 - type result struct { *proto.IssueResponse error } ch := make(chan result) - for i := 0; i < 6; i++ { + for i := 0; i < 5; i++ { go func() { resp, err := mtca.Issue(t.Context(), &proto.IssueRequest{ Pubkey: []byte("abc"), @@ -195,41 +173,63 @@ func TestSequence(t *testing.T) { ch <- result{IssueResponse: resp, error: err} }() } - + // Issue blocks until sequencing, so wait for all five to be pooled. + for mtca.pool.len() < 5 { + time.Sleep(time.Millisecond) + } + // With the pool full, a sixth request should fail. + _, err = mtca.Issue(t.Context(), &proto.IssueRequest{Pubkey: []byte("abc")}) + if err == nil { + t.Fatal("Issue with a full pool: got nil error, want error") + } + if !strings.Contains(err.Error(), "pool is full") { + t.Errorf("Issue with a full pool: expected 'pool is full', got %q", err) + } + // The checkpoint has no mirror signature yet, so sequencing must fail. + err = mtca.sequence(t.Context()) + if err == nil { + t.Fatalf("sequencing with an unready checkpoint: got nil error, want error") + } + if !errors.Is(err, ErrCheckpointNotReady) { + t.Errorf("sequencing with an unready checkpoint: want ErrCheckpointNotReady, got %q", err) + } + if mtca.pool.len() != 5 { + t.Errorf("pool after refused sequencing: got len %d, want 5", mtca.pool.len()) + } + // Fake publication + latest, err := mtca.latestCheckpoint(t.Context()) + if err != nil { + t.Fatalf("getting latest: %s", err) + } + latest.MirrorID = "fake" + latest.MirrorSignature = []byte("fake") + _, err = mtca.db.Update(t.Context(), latest) + if err != nil { + t.Fatalf("updating checkpoint with fake mirror signature: %s", err) + } + // Now sequencing should succeed. err = mtca.sequence(t.Context()) if err != nil { t.Fatalf("sequencing with waiting entries: %s", err) } - + // The five requests should now be unblocked and return results. seenIDs := map[int64]bool{} - var seenError error - for i := 0; i < 6; i++ { + for i := 0; i < 5; i++ { res := <-ch if res.error != nil { - if seenError != nil { - t.Errorf("too many errors: %s", res.error) - } - seenError = res.error + t.Errorf("Issue: %s", res.error) continue } - if seenIDs[res.MtcEntryIndex] { - t.Errorf("entryIndex %d seen twice", res.MtcEntryIndex) + if res.MtcEntryIndex < 1 || res.MtcEntryIndex > 5 || seenIDs[res.MtcEntryIndex] { + t.Errorf("entryIndex %d out of range or seen twice", res.MtcEntryIndex) } seenIDs[res.MtcEntryIndex] = true } - - if seenError == nil { - t.Errorf("putting 6 entries in a pool of size 5: expected error, got none") - } - if !strings.Contains(seenError.Error(), "pool is full") { - t.Errorf("putting 6 entries in a pool of size 5: expected 'pool is full', got %q", seenError) - } - + // Lastly, verify the resulting checkpoint is valid. latest, err = mtca.latestCheckpoint(t.Context()) if err != nil { t.Fatalf("getting latest: %s", err) } - verify(t, mtca, latest) } @@ -304,6 +304,9 @@ DgQIBAaC3xMBAgEwCgYIKoZIzj0EAwIDQQAwPgIdAMebuq7759hyFC3hjrVUEaXk } cert, err := x509.ParseCertificate(certBytes) + if err != nil { + t.Fatal(err) + } mtcaID, err := getMTCAID(cert) if err != nil { t.Fatal(err) diff --git a/sa/db/01-mtcmeta_44947_4_1_0_44.sql b/sa/db/01-mtcmeta_44947_4_1_0_44.sql index 397a25eab72..58639a01925 100644 --- a/sa/db/01-mtcmeta_44947_4_1_0_44.sql +++ b/sa/db/01-mtcmeta_44947_4_1_0_44.sql @@ -38,8 +38,6 @@ CREATE TABLE `checkpoints` ( -- Note that `log_origin` and `cosigner_name` in the link above are derived from `mtcLogID` and `mirrorID` respectively. -- Also, for checkpoint signatures start == 0 and end == tree size. -- `treeSize` will be strictly increasing over time, enforced by the application. - -- Appending to this table will involve a check (in a transaction) that the treeSize and root hash of the - -- highest-sized checkpoint match what the CA expects. `treeSize` bigint(20) unsigned NOT NULL, `rootHash` binary(32) NOT NULL,