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..e1c1e49bef0 100644 --- a/cmd/boulder-mtca/main.go +++ b/cmd/boulder-mtca/main.go @@ -4,12 +4,18 @@ package notmain import ( "context" + "database/sql" + "errors" "flag" "os" + "sync" + "time" "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 +28,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 +49,12 @@ 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") + flag.Parse() if *configFile == "" { flag.Usage() @@ -67,7 +84,33 @@ 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, c.MTCA.SequencingPeriod.Duration, 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 { + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + err = mtcaImpl.InitLog(ctx) + cmd.FailOnError(err, "Initializing log") + return + } + if *initLogForTest { + 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") + } + } srv := bgrpc.NewServer(c.MTCA.GRPCMTCA, logger).Add( &mtcapb.MTCA_ServiceDesc, mtcaImpl) @@ -75,6 +118,19 @@ func main() { start, err := srv.Build(tlsConfig, scope, clk) cmd.FailOnError(err, "Unable to setup MTCA gRPC server") + // 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. + 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 c40eb1df21c..49ac955cf1d 100644 --- a/mtca/mtca.go +++ b/mtca/mtca.go @@ -4,32 +4,54 @@ package mtca import ( "context" + "crypto/rand" + "crypto/sha256" + "crypto/x509" + "database/sql" "encoding/asn1" + "errors" "fmt" "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" ) +var ErrIssuanceLogAlreadyInitialized = errors.New("issuance log already initialized") +var ErrCheckpointNotReady = errors.New("not ready - no mirror signature") + var _ mtcapb.MTCAServer = &mtca{} -func New(issuer *issuance.Issuer) *mtca { - var mtcaID string - testingTrustAnchorIDOID := asn1.ObjectIdentifier{1, 3, 6, 1, 4, 1, 44363, 47, 1} - for _, attribute := range issuer.Cert.Subject.Names { - if attribute.Type.Equal(testingTrustAnchorIDOID) { - mtcaID, _ = attribute.Value.(string) - break - } +// New creates a new MTCA service. +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, // TODO: collect this from config - logNumber: 0, - latestEntryIndex: 0, - } + logNumber: 44, + pool: &pool{maxSize: 100}, + + sequencingPeriod: sequencingPeriod, + + db: initDB(dbMap), + log: logger, + }, nil } type mtca struct { @@ -39,10 +61,153 @@ type mtca struct { mtcaID string logNumber uint16 - // This is just a dummy for testing; in reality this will come from the DB. - latestEntryIndex int64 + 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 + + 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) +} + +// 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 := 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 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) + } + + if numCheckpoints > 0 || numLatestCheckpoints > 0 { + if numLatestCheckpoints == 1 { + return nil, ErrIssuanceLogAlreadyInitialized + } + + 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 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 + // 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[:], + } + + message, err := m.cosignedMessage(&firstCheckpoint) + if err != nil { + return nil, err + } + + sig, err := m.sign(message) + 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.latestCheckpoint(ctx) + if err != nil { + return fmt.Errorf("fetching first checkpoint: %s", err) + } + + return err +} + +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 +} - sequencing sync.Mutex +func (p *pool) take() []entry { + p.Lock() + defer p.Unlock() + ret := p.entries + p.entries = nil + return ret +} + +func (p *pool) len() int { + p.RLock() + defer p.RUnlock() + return len(p.entries) +} + +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. @@ -52,13 +217,347 @@ 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) { - m.sequencing.Lock() - defer m.sequencing.Unlock() - m.latestEntryIndex++ + // 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, + identifiers: req.Identifiers, + ch: ch, + }) + if err != nil { + return nil, err + } - return &mtcapb.IssueResponse{ - MtcLogID: m.mtcLogID(), - MtcEntryIndex: m.latestEntryIndex, + 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) { + go m.fakePublisher(ctx) + + since := time.Now() + ticker := time.NewTicker(m.sequencingPeriod) + defer ticker.Stop() + 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*m.sequencingPeriod { + m.log.Errf("after %s: %s", time.Since(since).Round(time.Millisecond), err) + } + continue + } + 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) + } + 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) + defer ticker.Stop() + for { + select { + case <-ticker.C: + latest, err := m.latestCheckpoint(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 + } + } +} + +// 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.latestCheckpoint(ctx) + if err != nil { + return err + } + + if !latest.mirrored() { + return fmt.Errorf("temporary: checkpoint ID %d (tree size %d): %w", + latest.ID, latest.TreeSize, ErrCheckpointNotReady) + } + + // 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[:], + } + + 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. + // + // 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.sign(message) + 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 deferred error path doesn't try to notify them. + entries = nil + + 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"` + 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) mirrored() 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) latestCheckpoint(ctx context.Context) (*checkpoint, error) { + var latest checkpoint + err := m.db.SelectOne(ctx, &latest, + `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 &latest, nil +} + +func (m *mtca) cosignedMessage(c *checkpoint) (*cosigned.Message, 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") + } + + 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 } + +// 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 { + 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..0bb96eef6de --- /dev/null +++ b/mtca/mtca_test.go @@ -0,0 +1,319 @@ +//go:build go1.27 + +package mtca + +import ( + "bytes" + "context" + "crypto/mldsa" + "crypto/x509" + "database/sql" + "encoding/base64" + "encoding/hex" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "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" +) + +// 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 + for i := 0; i < 20; i++ { + wg.Go(func() { + err := p.append(entry{}) + if err != nil { + t.Errorf("appending entry: %s", err) + } + }) + } + wg.Wait() + + 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 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) + } +} + +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) + } + t.Cleanup(cleanup) + // An empty pool is a no-op regardless of checkpoint state. + err = mtca.sequence(t.Context()) + if err != nil { + 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 < 5; 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} + }() + } + // 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{} + for i := 0; i < 5; i++ { + res := <-ch + if res.error != nil { + t.Errorf("Issue: %s", res.error) + continue + } + 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 + } + // 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) +} + +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.latestCheckpoint(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) + } +} + +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) + if err != nil { + t.Fatal(err) + } + 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) + } +} 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..58639a01925 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,23 @@ +-- 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 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` ( + -- 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`) +); + CREATE TABLE `checkpoints` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, -- ASCII-format OID relative to 1.3.6.1.4.1 @@ -8,7 +26,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. @@ -20,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, 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..394264981cd 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", 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..b3eb48a5a4f --- /dev/null +++ b/test/mtca/README.md @@ -0,0 +1,5 @@ +This directory contains a handful of utility scripts for manually interacting +with boulder-mtca. + + - 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 new file mode 100755 index 00000000000..d040375fd0d --- /dev/null +++ b/test/mtca/clear.sh @@ -0,0 +1,6 @@ +#!/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/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..b7795ada850 100644 --- a/test/startservers.py +++ b/test/startservers.py @@ -149,7 +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'), + ('./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,