fix: Bound untrusted manifest cache

include/xrpl/server/Manifest.h

@@ -45,12 +45,15 @@ namespace xrpl {
     dynamically generates the signatureless form when it needs to verify
     the signature.
 
-    An instance of ManifestCache stores, for each trusted validator, (a) its
+    An instance of ManifestCache stores, for each known validator, (a) its
     master public key, and (b) the most senior of all valid manifests it has
     seen for that validator, if any.  On startup, the [validator_token] config
     entry (which contains the manifest for this validator) is decoded and
     added to the manifest cache.  Other manifests are added as "gossip"
-    received from xrpld peers.
+    received from xrpld peers, including ones for validators this node does not
+    list. Manifests for unlisted validators are capped (kMaxUntrustedCount)
+    so peer gossip cannot grow the cache without bound; listed validators are
+    not capped. Entries are never evicted, so a stored revocation is permanent.
 
     When an ephemeral key is compromised, a new signing key pair is created,
     along with a new manifest vouching for it (with a higher sequence number),
@@ -258,30 +261,17 @@ loadValidatorToken(
     beast::Journal journal = beast::Journal(beast::Journal::getNullSink()));
 
 enum class ManifestDisposition {
-    /**
-     * Manifest is valid
-     */
-    Accepted = 0,
+    Accepted = 0,  ///< Manifest is valid
 
-    /**
-     * Sequence is too old
-     */
-    Stale,
+    Stale,  ///< Sequence is too old
 
-    /**
-     * The master key is not acceptable to us
-     */
-    BadMasterKey,
+    BadMasterKey,  ///< The master key is not acceptable to us
 
-    /**
-     * The ephemeral key is not acceptable to us
-     */
-    BadEphemeralKey,
+    BadEphemeralKey,  ///< The ephemeral key is not acceptable to us
 
-    /**
-     * Timely, but invalid signature
-     */
-    Invalid
+    Invalid,  ///< Timely, but invalid signature
+
+    UntrustedCapacity  ///< Unlisted and limit reached
 };
 
 inline std::string
@@ -299,11 +289,25 @@ to_string(ManifestDisposition m)
             return "badEphemeralKey";
         case ManifestDisposition::Invalid:
             return "invalid";
+        case ManifestDisposition::UntrustedCapacity:
+            return "untrustedCapacity";
         default:
             return "unknown";
     }
 }
 
+/**
+ * Whether a manifest counts against the 'untrusted' cache cap.
+ *
+ * Passed to `ManifestCache::applyManifest` with no default, so every caller
+ * must choose. `Capped` is the safe, flood-resistant value; only listed or
+ * configured keys should use `Uncapped`.
+ */
+enum class ManifestRateLimitCap : std::uint8_t {
+    Capped,   ///< Subject to the untrusted cap (unlisted peer gossip)
+    Uncapped  ///< Bypasses the cap (listed/trusted or config manifests)
+};
+
 class DatabaseCon;
 
 /**
@@ -327,6 +331,38 @@ private:
 
     std::atomic<std::uint32_t> seq_{0};
 
+    /**
+     * Master keys of cached manifests for validators this node does not list.
+     *
+     * One entry per capped key in `map_`; its size enforces the cap below.
+     * A key is added when first cached under `Capped` and removed when it
+     * becomes listed (see `promoteToTrusted`) or an `Uncapped` update arrives,
+     * never re-added on de-listing. Uncapped keys are not tracked here.
+     */
+    hash_set<PublicKey> untrustedKeys_;
+
+    /**
+     * Maximum number of untrusted master keys kept in the cache.
+     *
+     * Once reached, a manifest for a brand-new unlisted key is rejected.
+     */
+    static constexpr std::size_t kMaxUntrustedCount = 50000;
+
+    /**
+     * Running count of manifests rejected because the untrusted cap was full.
+     *
+     * Drives throttled logging (see `kUntrustedRejectCount`). Atomic because
+     * `applyManifest` may run concurrently.
+     */
+    std::atomic<std::uint64_t> untrustedRejectCount_{0};
+
+    /**
+     * Number of cap rejections between summary warnings.
+     *
+     * @see untrustedRejectCount_
+     */
+    static constexpr std::uint64_t kUntrustedRejectCount = 10000;
+
 public:
     explicit ManifestCache(beast::Journal j = beast::Journal(beast::Journal::getNullSink())) : j_(j)
     {
@@ -411,17 +447,44 @@ public:
     /**
      * Add manifest to cache.
      *
+     * A brand-new unlisted key is rejected once the untrusted cap is full;
+     * updates to a cached key and `Uncapped` manifests bypass the cap. The
+     * caller decides `cap` before calling so the cache lock is not held while
+     * consulting the validator list, which would risk a lock-ordering deadlock.
+     *
      * @param m Manifest to add
      *
-     * @return `ManifestDisposition::accepted` if successful, or
-     *         `stale` or `invalid` otherwise
+     * @param cap `Uncapped` skips the untrusted cap; use it for keys that are
+     *     listed, configured, or loaded from the DB. Note `Uncapped` does not
+     *     assert the key is currently trusted (a DB entry may predate a
+     *     de-listing). Callers must state this explicitly so a manifest is
+     *     never left uncapped by omission.
+     *
+     * @return `Accepted` if stored, `Stale` if superseded, `Invalid`/
+     *         `BadEphemeralKey` if malformed, or `UntrustedCapacity` if the
+     *         untrusted cap is full.
      *
      * @par Thread Safety
      *
      * May be called concurrently
      */
     ManifestDisposition
-    applyManifest(Manifest m);
+    applyManifest(Manifest m, ManifestRateLimitCap cap);
+
+    /**
+     * Stop counting a master key against the untrusted cap.
+     *
+     * Called when a cached untrusted key becomes listed, freeing its slot.
+     * Idempotent and a no-op for keys that were never counted.
+     *
+     * @param pk Master public key that is now listed/trusted
+     *
+     * @par Thread Safety
+     *
+     * May be called concurrently
+     */
+    void
+    promoteToTrusted(PublicKey const& pk);
 
     /**
      * Populate manifest cache with manifests in database and config.

src/libxrpl/server/Manifest.cpp

@@ -382,16 +382,20 @@ ManifestCache::revoked(PublicKey const& pk) const
 }
 
 ManifestDisposition
-ManifestCache::applyManifest(Manifest m)
+ManifestCache::applyManifest(Manifest m, ManifestRateLimitCap const cap)
 {
+    bool const uncapped = cap == ManifestRateLimitCap::Uncapped;
+
+    // The signature is checked only on the first `prewriteCheck` run (under the
+    // read lock). It is expensive, so `checkSignature` is cleared the first
+    // time it is read; the second run (under the write lock) skips it.
+    bool checkSignature = true;
+
     // Check the manifest against the conditions that do not require a
-    // `unique_lock` (write lock) on the `mutex_`. Since the signature can be
-    // relatively expensive, the `checkSignature` parameter determines if the
-    // signature should be checked. Since `prewriteCheck` is run twice (see
-    // comment below), `checkSignature` only needs to be set to true on the
-    // first run.
-    auto prewriteCheck = [this, &m](auto const& iter, bool checkSignature, auto const& lock)
-        -> std::optional<ManifestDisposition> {
+    // `unique_lock` (write lock) on the `mutex_`.
+    auto prewriteCheck = [this, &m, &checkSignature](
+                             auto const& iter,
+                             auto const& lock) -> std::optional<ManifestDisposition> {
         XRPL_ASSERT(lock.owns_lock(), "xrpl::ManifestCache::applyManifest::prewriteCheck : locked");
         (void)lock;  // not used. parameter is present to ensure the mutex is
                      // locked when the lambda is called.
@@ -406,11 +410,15 @@ ManifestCache::applyManifest(Manifest m)
             return ManifestDisposition::Stale;
         }
 
-        if (checkSignature && !m.verify())
+        if (checkSignature)
         {
-            if (auto stream = j_.warn())
-                logMftAct(stream, "Invalid", m.masterKey, m.sequence);
-            return ManifestDisposition::Invalid;
+            checkSignature = false;
+            if (!m.verify())
+            {
+                if (auto stream = j_.warn())
+                    logMftAct(stream, "Invalid", m.masterKey, m.sequence);
+                return ManifestDisposition::Invalid;
+            }
         }
 
         // If the master key associated with a manifest is or might be
@@ -470,14 +478,51 @@ ManifestCache::applyManifest(Manifest m)
         return std::nullopt;
     };
 
+    // Reject a brand-new manifest for an unlisted key once the untrusted cap
+    // is full. Updates to a cached key and uncapped manifests always pass.
+    // Called under both the read and write lock, since the cap can be reached
+    // between the two. The lock param enforces that.
+    auto atUntrustedCap = [this, &m, uncapped](auto const& iter, auto const& lock) {
+        XRPL_ASSERT(
+            lock.owns_lock(), "xrpl::ManifestCache::applyManifest::atUntrustedCap : locked");
+        (void)lock;  // not used. parameter is present to ensure the mutex is
+                     // locked when the lambda is called.
+        if (iter == map_.end() && !uncapped && untrustedKeys_.size() >= kMaxUntrustedCount)
+        {
+            // Log each rejection at debug, but warn only once per interval so a
+            // flood does not fill the log.
+            if (auto stream = j_.debug())
+                logMftAct(stream, "UntrustedCapacity", m.masterKey, m.sequence);
+            if (auto const n = untrustedRejectCount_.fetch_add(1) + 1;
+                n % kUntrustedRejectCount == 0)
+            {
+                JLOG(j_.warn()) << "Untrusted manifest cap reached; " << n
+                                << " manifests rejected so far";
+            }
+            return true;
+        }
+        return false;
+    };
+
     {
         std::shared_lock const sl{mutex_};
-        if (auto d = prewriteCheck(map_.find(m.masterKey), /*checkSig*/ true, sl))
+        auto const iter = map_.find(m.masterKey);
+
+        if (atUntrustedCap(iter, sl))
+            return ManifestDisposition::UntrustedCapacity;
+
+        if (auto d = prewriteCheck(iter, sl); d.has_value())
             return *d;
     }
 
     std::unique_lock const sl{mutex_};
     auto const iter = map_.find(m.masterKey);
+
+    // Re-check the cap under the write lock: the cache may have grown while the
+    // read lock above was released.
+    if (atUntrustedCap(iter, sl))
+        return ManifestDisposition::UntrustedCapacity;
+
     // Since we released the previously held read lock, it's possible that the
     // collections have been written to. This means we need to run
     // `prewriteCheck` again. This re-does work, but `prewriteCheck` is
@@ -487,7 +532,7 @@ ManifestCache::applyManifest(Manifest m)
     // doesn't need to happen again (signature checks are somewhat expensive).
     // Note: It's a mistake to use an upgradable lock. This is a recipe for
     // deadlock.
-    if (auto d = prewriteCheck(iter, /*checkSig*/ false, sl))
+    if (auto d = prewriteCheck(iter, sl); d.has_value())
         return *d;
 
     bool const revoked = m.revoked();
@@ -506,6 +551,12 @@ ManifestCache::applyManifest(Manifest m)
         }
 
         auto masterKey = m.masterKey;
+
+        // Count this key against the untrusted cap. Uncapped keys (listed,
+        // configured, or DB-loaded) are not tracked.
+        if (!uncapped)
+            untrustedKeys_.insert(masterKey);
+
         map_.emplace(std::move(masterKey), std::move(m));
 
         // Something has changed. Keep track of it.
@@ -519,6 +570,11 @@ ManifestCache::applyManifest(Manifest m)
     if (auto stream = j_.info())
         logMftAct(stream, "AcceptedUpdate", m.masterKey, m.sequence, iter->second.sequence);
 
+    // If this key was counted against the cap but now arrives uncapped, free
+    // its slot without waiting for promoteToTrusted.
+    if (uncapped)
+        untrustedKeys_.erase(m.masterKey);
+
     signingToMasterKeys_.erase(
         *iter->second.signingKey);  // NOLINT(bugprone-unchecked-optional-access) prewriteCheck
                                     // ensures old manifest is not revoked
@@ -526,8 +582,8 @@ ManifestCache::applyManifest(Manifest m)
     if (!revoked)
     {
         signingToMasterKeys_.emplace(
-            *m.signingKey, m.masterKey);  // NOLINT(bugprone-unchecked-optional-access) non-revoked
-                                          // manifest always has signingKey
+            *m.signingKey, m.masterKey);  // NOLINT(bugprone-unchecked-optional-access)
+                                          // non-revoked manifest always has signingKey
     }
 
     iter->second = std::move(m);
@@ -538,6 +594,16 @@ ManifestCache::applyManifest(Manifest m)
     return ManifestDisposition::Accepted;
 }
 
+void
+ManifestCache::promoteToTrusted(PublicKey const& pk)
+{
+    // Frees the key's untrusted slot; a no-op (and idempotent) if the key was
+    // never counted. Not re-added on de-listing, so list/de-list cannot grow
+    // the count.
+    std::unique_lock const sl{mutex_};
+    untrustedKeys_.erase(pk);
+}
+
 void
 ManifestCache::load(DatabaseCon& dbCon, std::string const& dbTable)
 {
@@ -568,7 +634,8 @@ ManifestCache::load(
             JLOG(j_.warn()) << "Configured manifest revokes public key";
         }
 
-        if (applyManifest(std::move(*mo)) == ManifestDisposition::Invalid)
+        if (applyManifest(std::move(*mo), ManifestRateLimitCap::Uncapped) ==
+            ManifestDisposition::Invalid)
         {
             JLOG(j_.error()) << "Manifest in config was rejected";
             return false;
@@ -590,7 +657,9 @@ ManifestCache::load(
 
         auto mo = deserializeManifest(base64Decode(revocationStr));
 
-        if (!mo || !mo->revoked() || applyManifest(std::move(*mo)) == ManifestDisposition::Invalid)
+        if (!mo || !mo->revoked() ||
+            applyManifest(std::move(*mo), ManifestRateLimitCap::Uncapped) ==
+                ManifestDisposition::Invalid)
         {
             JLOG(j_.error()) << "Invalid validator key revocation in config";
             return false;

src/libxrpl/server/Wallet.cpp

@@ -29,6 +29,7 @@
 #include <soci/use.h>
 
 #include <array>
+#include <cstddef>
 #include <functional>
 #include <memory>
 #include <string>
@@ -77,7 +78,9 @@ getManifests(
                 continue;
             }
 
-            cache.applyManifest(std::move(*mo));
+            // Only trusted manifests are persisted (see saveManifests), so
+            // anything loaded from the DB bypasses the untrusted cap.
+            cache.applyManifest(std::move(*mo), ManifestRateLimitCap::Uncapped);
         }
         else
         {
@@ -107,19 +110,27 @@ saveManifests(
 {
     soci::transaction tr(session);
     session << "DELETE FROM " << dbTable;
+    // Count skipped untrusted manifests and log one summary afterwards, since
+    // the cache can hold many and per-entry logging would flood at shutdown.
+    std::size_t skipped = 0;
     for (auto const& v : map)
     {
-        // Save all revocation manifests,
-        // but only save trusted non-revocation manifests.
-        if (!v.second.revoked() && !isTrusted(v.second.masterKey))
+        // Persist only trusted keys. Untrusted gossip is left out so a flood
+        // cannot survive a restart on disk.
+        if (!isTrusted(v.second.masterKey))
         {
-            JLOG(j.info()) << "Untrusted manifest in cache not saved to db";
+            ++skipped;
             continue;
         }
 
         saveManifest(session, dbTable, v.second.serialized);
     }
     tr.commit();
+
+    if (skipped != 0)
+    {
+        JLOG(j.info()) << skipped << " untrusted manifest(s) in cache not saved to db";
+    }
 }
 
 void

src/test/app/Manifest_test.cpp

@@ -399,7 +399,8 @@ public:
         BEAST_EXPECT(
             ManifestDisposition::Accepted ==
             cache.applyManifest(
-                makeManifest(sk, KeyType::Ed25519, kp0.second, KeyType::Secp256k1, 0)));
+                makeManifest(sk, KeyType::Ed25519, kp0.second, KeyType::Secp256k1, 0),
+                ManifestRateLimitCap::Capped));
         BEAST_EXPECT(cache.getSigningKey(pk) == kp0.first);
         BEAST_EXPECT(cache.getMasterKey(kp0.first) == pk);
 
@@ -411,7 +412,8 @@ public:
         BEAST_EXPECT(
             ManifestDisposition::Accepted ==
             cache.applyManifest(
-                makeManifest(sk, KeyType::Ed25519, kp1.second, KeyType::Secp256k1, 1)));
+                makeManifest(sk, KeyType::Ed25519, kp1.second, KeyType::Secp256k1, 1),
+                ManifestRateLimitCap::Capped));
         BEAST_EXPECT(cache.getSigningKey(pk) == kp1.first);
         BEAST_EXPECT(cache.getMasterKey(kp1.first) == pk);
         BEAST_EXPECT(cache.getMasterKey(kp0.first) == kp0.first);
@@ -421,7 +423,8 @@ public:
         BEAST_EXPECT(
             ManifestDisposition::BadEphemeralKey ==
             cache.applyManifest(
-                makeManifest(sk, KeyType::Ed25519, kp1.second, KeyType::Secp256k1, 2)));
+                makeManifest(sk, KeyType::Ed25519, kp1.second, KeyType::Secp256k1, 2),
+                ManifestRateLimitCap::Capped));
         BEAST_EXPECT(cache.getSigningKey(pk) == kp1.first);
         BEAST_EXPECT(cache.getMasterKey(kp1.first) == pk);
         BEAST_EXPECT(cache.getMasterKey(kp0.first) == kp0.first);
@@ -431,7 +434,8 @@ public:
         // key from a revoked master public key
         BEAST_EXPECT(
             ManifestDisposition::Accepted ==
-            cache.applyManifest(makeRevocation(sk, KeyType::Ed25519)));
+            cache.applyManifest(
+                makeRevocation(sk, KeyType::Ed25519), ManifestRateLimitCap::Capped));
         BEAST_EXPECT(cache.revoked(pk));
         BEAST_EXPECT(cache.getSigningKey(pk) == pk);
         BEAST_EXPECT(cache.getMasterKey(kp0.first) == kp0.first);
@@ -902,39 +906,69 @@ public:
             // applyManifest should accept new manifests with
             // higher sequence numbers
             auto const seq0 = cache.sequence();
-            BEAST_EXPECT(cache.applyManifest(clone(sA0)) == ManifestDisposition::Accepted);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sA0), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Accepted);
             BEAST_EXPECT(cache.sequence() > seq0);
 
             auto const seq1 = cache.sequence();
-            BEAST_EXPECT(cache.applyManifest(clone(sA0)) == ManifestDisposition::Stale);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sA0), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Stale);
             BEAST_EXPECT(cache.sequence() == seq1);
 
-            BEAST_EXPECT(cache.applyManifest(clone(sA1)) == ManifestDisposition::Accepted);
-            BEAST_EXPECT(cache.applyManifest(clone(sA1)) == ManifestDisposition::Stale);
-            BEAST_EXPECT(cache.applyManifest(clone(sA0)) == ManifestDisposition::Stale);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sA1), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Accepted);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sA1), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Stale);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sA0), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Stale);
 
-            BEAST_EXPECT(cache.applyManifest(clone(sA2)) == ManifestDisposition::BadEphemeralKey);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sA2), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::BadEphemeralKey);
 
             // applyManifest should accept manifests with max sequence numbers
             // that revoke the master public key
             BEAST_EXPECT(!cache.revoked(pkA));
             BEAST_EXPECT(sAMax.revoked());
-            BEAST_EXPECT(cache.applyManifest(clone(sAMax)) == ManifestDisposition::Accepted);
-            BEAST_EXPECT(cache.applyManifest(clone(sAMax)) == ManifestDisposition::Stale);
-            BEAST_EXPECT(cache.applyManifest(clone(sA1)) == ManifestDisposition::Stale);
-            BEAST_EXPECT(cache.applyManifest(clone(sA0)) == ManifestDisposition::Stale);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sAMax), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Accepted);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sAMax), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Stale);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sA1), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Stale);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sA0), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Stale);
             BEAST_EXPECT(cache.revoked(pkA));
 
             // applyManifest should reject manifests with invalid signatures
-            BEAST_EXPECT(cache.applyManifest(clone(sB0)) == ManifestDisposition::Accepted);
-            BEAST_EXPECT(cache.applyManifest(clone(sB0)) == ManifestDisposition::Stale);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sB0), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Accepted);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sB0), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Stale);
             BEAST_EXPECT(!deserializeManifest(fake));
-            BEAST_EXPECT(cache.applyManifest(clone(sB1)) == ManifestDisposition::Invalid);
-            BEAST_EXPECT(cache.applyManifest(clone(sB2)) == ManifestDisposition::Accepted);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sB1), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Invalid);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sB2), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Accepted);
 
             auto const sC0 = makeManifest(
                 kpB2.second, KeyType::Ed25519, randomSecretKey(), KeyType::Ed25519, 47);
-            BEAST_EXPECT(cache.applyManifest(clone(sC0)) == ManifestDisposition::BadMasterKey);
+            BEAST_EXPECT(
+                cache.applyManifest(clone(sC0), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::BadMasterKey);
         }
 
         testLoadStore(cache);

src/test/app/ValidatorList_test.cpp

@@ -278,8 +278,10 @@ private:
                 trustedKeys->load(localSigningPublicOuter, emptyCfgKeys, emptyCfgPublishers));
             BEAST_EXPECT(trustedKeys->listed(localSigningPublicOuter));
 
-            // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
-            manifests.applyManifest(*deserializeManifest(cfgManifest));
+            // NOLINTBEGIN(bugprone-unchecked-optional-access)
+            manifests.applyManifest(
+                *deserializeManifest(cfgManifest), ManifestRateLimitCap::Capped);
+            // NOLINTEND(bugprone-unchecked-optional-access)
             BEAST_EXPECT(
                 trustedKeys->load(localSigningPublicOuter, emptyCfgKeys, emptyCfgPublishers));
 
@@ -369,8 +371,10 @@ private:
                 app.config().legacy(Sections::kDatabasePath),
                 env.journal);
 
-            // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
-            manifests.applyManifest(*deserializeManifest(cfgManifest));
+            // NOLINTBEGIN(bugprone-unchecked-optional-access)
+            manifests.applyManifest(
+                *deserializeManifest(cfgManifest), ManifestRateLimitCap::Capped);
+            // NOLINTEND(bugprone-unchecked-optional-access)
 
             BEAST_EXPECT(trustedKeys->load(localSigningPublicOuter, cfgKeys, emptyCfgPublishers));
 
@@ -455,13 +459,16 @@ private:
             auto const pubRevokedSigning = randomKeyPair(KeyType::Secp256k1);
             // make this manifest revoked (seq num = max)
             //  -- thus should not be loaded
-            // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
-            pubManifests.applyManifest(*deserializeManifest(makeManifestString(
-                pubRevokedPublic,
-                pubRevokedSecret,
-                pubRevokedSigning.first,
-                pubRevokedSigning.second,
-                std::numeric_limits<std::uint32_t>::max())));
+            // NOLINTBEGIN(bugprone-unchecked-optional-access)
+            pubManifests.applyManifest(
+                *deserializeManifest(makeManifestString(
+                    pubRevokedPublic,
+                    pubRevokedSecret,
+                    pubRevokedSigning.first,
+                    pubRevokedSigning.second,
+                    std::numeric_limits<std::uint32_t>::max())),
+                ManifestRateLimitCap::Capped);
+            // NOLINTEND(bugprone-unchecked-optional-access)
 
             // these two are not revoked (and not in the manifest cache at all.)
             auto legitKey1 = randomMasterKey();
@@ -494,13 +501,16 @@ private:
             auto const pubRevokedSigning = randomKeyPair(KeyType::Secp256k1);
             // make this manifest revoked (seq num = max)
             //  -- thus should not be loaded
-            // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
-            pubManifests.applyManifest(*deserializeManifest(makeManifestString(
-                pubRevokedPublic,
-                pubRevokedSecret,
-                pubRevokedSigning.first,
-                pubRevokedSigning.second,
-                std::numeric_limits<std::uint32_t>::max())));
+            // NOLINTBEGIN(bugprone-unchecked-optional-access)
+            pubManifests.applyManifest(
+                *deserializeManifest(makeManifestString(
+                    pubRevokedPublic,
+                    pubRevokedSecret,
+                    pubRevokedSigning.first,
+                    pubRevokedSigning.second,
+                    std::numeric_limits<std::uint32_t>::max())),
+                ManifestRateLimitCap::Capped);
+            // NOLINTEND(bugprone-unchecked-optional-access)
 
             // this one is not revoked (and not in the manifest cache at all.)
             auto legitKey = randomMasterKey();
@@ -1218,7 +1228,8 @@ private:
 
             BEAST_EXPECT(
                 // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
-                manifestsOuter.applyManifest(std::move(*m1)) == ManifestDisposition::Accepted);
+                manifestsOuter.applyManifest(std::move(*m1), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Accepted);
             BEAST_EXPECT(trustedKeysOuter->listed(masterPublic));
             BEAST_EXPECT(trustedKeysOuter->trusted(masterPublic));
             BEAST_EXPECT(trustedKeysOuter->listed(signingPublic1));
@@ -1232,7 +1243,8 @@ private:
                 masterPublic, masterPrivate, signingPublic2, signingKeys2.second, 2));
             BEAST_EXPECT(
                 // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
-                manifestsOuter.applyManifest(std::move(*m2)) == ManifestDisposition::Accepted);
+                manifestsOuter.applyManifest(std::move(*m2), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Accepted);
             BEAST_EXPECT(trustedKeysOuter->listed(masterPublic));
             BEAST_EXPECT(trustedKeysOuter->trusted(masterPublic));
             BEAST_EXPECT(trustedKeysOuter->listed(signingPublic2));
@@ -1249,7 +1261,8 @@ private:
             // NOLINTBEGIN(bugprone-unchecked-optional-access)
             BEAST_EXPECT(max->revoked());
             BEAST_EXPECT(
-                manifestsOuter.applyManifest(std::move(*max)) == ManifestDisposition::Accepted);
+                manifestsOuter.applyManifest(std::move(*max), ManifestRateLimitCap::Capped) ==
+                ManifestDisposition::Accepted);
             // NOLINTEND(bugprone-unchecked-optional-access)
 
             BEAST_EXPECT(manifestsOuter.getSigningKey(masterPublic) == masterPublic);
@@ -2668,7 +2681,9 @@ private:
             auto threshold = listThreshold > 0 ? std::optional(listThreshold) : std::nullopt;
             if (self)
             {
-                valManifests.applyManifest(*deserializeManifest(base64Decode(self->manifest)));
+                valManifests.applyManifest(
+                    *deserializeManifest(base64Decode(self->manifest)),
+                    ManifestRateLimitCap::Capped);
                 BEAST_EXPECT(
                     result->load(self->signingPublic, emptyCfgKeys, cfgPublishers, threshold));
             }

src/xrpld/app/misc/detail/ValidatorList.cpp

@@ -1065,6 +1065,8 @@ ValidatorList::updatePublisherList(
         {
             // Increment list count for added keys
             ++keyListings_[*iNew];
+            // Key is now listed: free its untrusted slot if it had one.
+            validatorManifests_.promoteToTrusted(*iNew);
             ++iNew;
         }
         else if (iNew == publisherList.end() || (iOld != oldList.end() && *iOld < *iNew))
@@ -1103,7 +1105,8 @@ ValidatorList::updatePublisherList(
             continue;
         }
 
-        if (auto const r = validatorManifests_.applyManifest(std::move(*m));
+        if (auto const r =
+                validatorManifests_.applyManifest(std::move(*m), ManifestRateLimitCap::Uncapped);
             r == ManifestDisposition::Invalid)
         {
             JLOG(j_.warn()) << "List for " << strHex(pubKey)
@@ -1357,7 +1360,10 @@ ValidatorList::verify(
     PublicKey masterPubKey = manifest.masterKey;
     auto const revoked = manifest.revoked();
 
-    auto const result = publisherManifests_.applyManifest(std::move(manifest));
+    // Publisher keys are configured/trusted (checked above), so bypass the
+    // untrusted cap.
+    auto const result =
+        publisherManifests_.applyManifest(std::move(manifest), ManifestRateLimitCap::Uncapped);
 
     if (revoked && result == ManifestDisposition::Accepted)
     {

src/xrpld/overlay/detail/OverlayImpl.cpp

@@ -675,13 +675,22 @@ OverlayImpl::onManifests(
         if (auto mo = deserializeManifest(s))
         {
             auto const serialized = mo->serialized;
+            // Resolve trust before applyManifest takes the manifest-cache
+            // lock: listed() takes the validator-list lock, so ordering it
+            // first avoids holding the two locks in opposite orders.
+            bool const isTrusted = app_.getValidators().listed(mo->masterKey);
+            // Updates to a known key are relayed even when untrusted. Use
+            // getSequence, not getManifest, to avoid copying the cached payload
+            // on this hot path.
+            bool const isKnown =
+                app_.getValidatorManifests().getSequence(mo->masterKey).has_value();
 
-            auto const result = app_.getValidatorManifests().applyManifest(std::move(*mo));
+            auto const result = app_.getValidatorManifests().applyManifest(
+                std::move(*mo),
+                isTrusted ? ManifestRateLimitCap::Uncapped : ManifestRateLimitCap::Capped);
 
             if (result == ManifestDisposition::Accepted)
             {
-                relay.add_list()->set_stobject(s);
-
                 // N.B.: this is important; the applyManifest call above moves
                 //       the loaded Manifest out of the optional so we need to
                 //       reload it here.
@@ -693,10 +702,19 @@ OverlayImpl::onManifests(
                 // NOLINTBEGIN(bugprone-unchecked-optional-access) assert above
                 app_.getOPs().pubManifest(*mo);
 
-                if (app_.getValidators().listed(mo->masterKey))
+                // Relay only trusted manifests or updates to known keys, so
+                // untrusted gossip for a brand-new key cannot be amplified.
+                // Persist to the wallet DB only for trusted keys, so untrusted
+                // gossip never survives a restart.
+                if (isTrusted || isKnown)
                 {
-                    auto db = app_.getWalletDB().checkoutDb();
-                    addValidatorManifest(*db, serialized);
+                    relay.add_list()->set_stobject(s);
+
+                    if (isTrusted)
+                    {
+                        auto db = app_.getWalletDB().checkoutDb();
+                        addValidatorManifest(*db, serialized);
+                    }
                 }
                 // NOLINTEND(bugprone-unchecked-optional-access)
             }