diff --git a/src/atomdb/AtomDB.h b/src/atomdb/AtomDB.h
index a8056c08..f36dd86a 100644
--- a/src/atomdb/AtomDB.h
+++ b/src/atomdb/AtomDB.h
@@ -52,6 +52,7 @@ class AtomDB : public HandleDecoder {
 
     virtual bool allow_nested_indexing() = 0;
     virtual bool composite_type_enabled() const = 0;
+    virtual atomdb_api_types::ProtectionMode get_protection_mode() const = 0;
 
     virtual shared_ptr<Atom> get_atom(const string& handle) = 0;  // HandleDecoder interface
     virtual shared_ptr<Node> get_node(const string& handle) = 0;
diff --git a/src/atomdb/AtomDBAPITypes.h b/src/atomdb/AtomDBAPITypes.h
index 7563ca8b..306052b6 100644
--- a/src/atomdb/AtomDBAPITypes.h
+++ b/src/atomdb/AtomDBAPITypes.h
@@ -120,5 +120,14 @@ class AccessPermissionDocument {
     }
 };
 
+/**
+ * @brief How an AtomDB participates in protected access.
+ *
+ * - UNPROTECTED: no authorization wrapper; open access.
+ * - PROTECTED: wrap and apply authorization post-processing (filter) after queries.
+ * - FORWARD: wrap and pass access keys through, but do not post-process locally
+ */
+enum class ProtectionMode { UNPROTECTED = 0, FORWARD, PROTECTED };
+
 }  // namespace atomdb_api_types
 }  // namespace atomdb
diff --git a/src/atomdb/AtomDBFactory.cc b/src/atomdb/AtomDBFactory.cc
index be3c07b0..f13080da 100644
--- a/src/atomdb/AtomDBFactory.cc
+++ b/src/atomdb/AtomDBFactory.cc
@@ -3,6 +3,7 @@
 #include "AdapterDB.h"
 #include "InMemoryDB.h"
 #include "MorkDB.h"
+#include "ProtectedAtomDB.h"
 #include "RedisMongoDB.h"
 #include "RemoteAtomDB.h"
 #include "Utils.h"
@@ -103,6 +104,26 @@ shared_ptr<AtomDB> AtomDBFactory::create_composite_atomdb(const JsonConfig& conf
 }
 
 shared_ptr<AtomDB> AtomDBFactory::wrap_if_protected(shared_ptr<AtomDB> atomdb) {
-    // AtomDBFactory::wrap_if_protected() is not implemented yet.
-    return atomdb;
+    if (!atomdb) {
+        RAISE_ERROR("AtomDBFactory::wrap_if_protected() received null atomdb");
+    }
+
+    const auto mode = atomdb->get_protection_mode();
+
+    // Interim: ProtectedAtomDB wrapping is disabled; pass UNPROTECTED backends through unchanged.
+    if (mode == atomdb_api_types::ProtectionMode::UNPROTECTED ||
+        dynamic_pointer_cast<ProtectedAtomDB>(atomdb)) {
+        return atomdb;
+    }
+
+    // Interim fail-closed: PROTECTED and FORWARD both require ProtectedAtomDB, which is not
+    // enabled yet. Federation (RemoteAtomDB) may report FORWARD when peers are protected;
+    // that case is rejected here until wrapping is restored.
+    // TODO: return make_shared<ProtectedAtomDB>(atomdb) when authorization integration is complete.
+    if (mode == atomdb_api_types::ProtectionMode::PROTECTED ||
+        mode == atomdb_api_types::ProtectionMode::FORWARD) {
+        RAISE_ERROR("Protected AtomDB support is not available");
+    }
+
+    RAISE_ERROR("AtomDBFactory::wrap_if_protected() encountered unknown protection mode");
 }
\ No newline at end of file
diff --git a/src/atomdb/AtomDBFactory.h b/src/atomdb/AtomDBFactory.h
index 7f3b663b..5f7ebb3d 100644
--- a/src/atomdb/AtomDBFactory.h
+++ b/src/atomdb/AtomDBFactory.h
@@ -27,7 +27,10 @@ namespace atomdb {
 class AtomDBFactory {
    public:
     /**
-     * @brief Creates a AtomDB and wraps it with ProtectedAtomDB when is applyable.
+     * @brief Creates an AtomDB from config.
+     *
+     * Interim behavior: ProtectedAtomDB wrapping is disabled. Configurations whose
+     * resulting AtomDB reports PROTECTED or FORWARD raise at the end of create().
      */
     static shared_ptr<AtomDB> create(const JsonConfig& config, const string& context = "");
 
@@ -40,7 +43,11 @@ class AtomDBFactory {
                                                       const string& context = "");
 
     /**
-     * @brief Wraps an AtomDB with ProtectedAtomDB when protected and not already wrapped.
+     * @brief Applies protection wrapping when enabled.
+     *
+     * Interim behavior: returns atomdb unchanged when mode is UNPROTECTED (or already
+     * wrapped). Raises when mode is PROTECTED or FORWARD because ProtectedAtomDB
+     * integration is not yet available.
      */
     static shared_ptr<AtomDB> wrap_if_protected(shared_ptr<AtomDB> atomdb);
 };
diff --git a/src/atomdb/BUILD b/src/atomdb/BUILD
index ed66ba36..95015e91 100644
--- a/src/atomdb/BUILD
+++ b/src/atomdb/BUILD
@@ -11,6 +11,12 @@ cc_library(
         ":atomdb_factory",
         ":atomdb_singleton",
         ":atomdbutils",
+        ":protected_atomdb",
+        "//atomdb/adapterdb:adapterdb_lib",
+        "//atomdb/inmemorydb:inmemorydb_lib",
+        "//atomdb/morkdb:morkdb_lib",
+        "//atomdb/redis_mongodb:redis_mongodb_lib",
+        "//atomdb/remotedb:remotedb_lib",
     ],
 )
 
@@ -21,6 +27,7 @@ cc_library(
     includes = ["."],
     deps = [
         ":atomdb",
+        ":protected_atomdb",
         "//atomdb/adapterdb:adapterdb_lib",
         "//atomdb/inmemorydb:inmemorydb_lib",
         "//atomdb/morkdb:morkdb_lib",
@@ -73,3 +80,16 @@ cc_library(
         "//commons:commons_lib",
     ],
 )
+
+cc_library(
+    name = "protected_atomdb",
+    srcs = ["ProtectedAtomDB.cc"],
+    hdrs = ["ProtectedAtomDB.h"],
+    includes = ["."],
+    deps = [
+        ":atomdb",
+        ":atomdb_api_types",
+        "//commons:commons_lib",
+        "//commons/atoms:atoms_lib",
+    ],
+)
diff --git a/src/atomdb/ProtectedAtomDB.cc b/src/atomdb/ProtectedAtomDB.cc
new file mode 100644
index 00000000..e9d9c6b6
--- /dev/null
+++ b/src/atomdb/ProtectedAtomDB.cc
@@ -0,0 +1,307 @@
+#include "ProtectedAtomDB.h"
+
+#define LOG_LEVEL INFO_LEVEL
+#include "Logger.h"
+#include "Utils.h"
+
+using namespace std;
+using namespace atomdb;
+using namespace commons;
+
+// --------------------------------------------------------------------------------
+// Constructors and destructors
+
+ProtectedAtomDB::ProtectedAtomDB(shared_ptr<AtomDB> backend) : backend(backend) {
+    if (this->backend == nullptr) {
+        RAISE_ERROR("ProtectedAtomDB requires a non-null backend AtomDB");
+    }
+    LOG_INFO("ProtectedAtomDB initialized");
+}
+
+// --------------------------------------------------------------------------------
+// Public methods
+
+shared_ptr<Atom> ProtectedAtomDB::get_atom(const string& handle,
+                                           const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::get_atom(handle, public_key) is not implemented yet");
+}
+
+shared_ptr<Node> ProtectedAtomDB::get_node(const string& handle,
+                                           const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::get_node(handle, public_key) is not implemented yet");
+}
+
+shared_ptr<Link> ProtectedAtomDB::get_link(const string& handle,
+                                           const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::get_link(handle, public_key) is not implemented yet");
+}
+
+vector<shared_ptr<Atom>> ProtectedAtomDB::get_matching_atoms(
+    bool is_toplevel, Atom& key, const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::get_matching_atoms(..., public_key) is not implemented yet");
+}
+
+shared_ptr<atomdb_api_types::HandleSet> ProtectedAtomDB::query_for_pattern(
+    const LinkSchema& link_schema, const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::query_for_pattern(link_schema, public_key) is not implemented yet");
+}
+
+shared_ptr<atomdb_api_types::HandleList> ProtectedAtomDB::query_for_targets(
+    const string& handle, const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::query_for_targets(handle, public_key) is not implemented yet");
+}
+
+shared_ptr<atomdb_api_types::HandleSet> ProtectedAtomDB::query_for_incoming_set(
+    const string& handle, const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::query_for_incoming_set(handle, public_key) is not implemented yet");
+}
+
+bool ProtectedAtomDB::atom_exists(const string& handle, const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::atom_exists(handle, public_key) is not implemented yet");
+}
+
+bool ProtectedAtomDB::node_exists(const string& handle, const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::node_exists(handle, public_key) is not implemented yet");
+}
+
+bool ProtectedAtomDB::link_exists(const string& handle, const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::link_exists(handle, public_key) is not implemented yet");
+}
+
+set<string> ProtectedAtomDB::atoms_exist(const vector<string>& handles,
+                                         const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::atoms_exist(handles, public_key) is not implemented yet");
+}
+
+set<string> ProtectedAtomDB::nodes_exist(const vector<string>& handles,
+                                         const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::nodes_exist(handles, public_key) is not implemented yet");
+}
+
+set<string> ProtectedAtomDB::links_exist(const vector<string>& handles,
+                                         const atomdb_api_types::PublicKey& public_key) {
+    RAISE_ERROR("ProtectedAtomDB::links_exist(handles, public_key) is not implemented yet");
+}
+
+string ProtectedAtomDB::add_atom(const atoms::Atom* atom,
+                                 const atomdb_api_types::PublicKey& public_key,
+                                 const atoms::Merger* merger) {
+    RAISE_ERROR("ProtectedAtomDB::add_atom(atom, public_key) is not implemented yet");
+}
+
+string ProtectedAtomDB::add_node(const atoms::Node* node,
+                                 const atomdb_api_types::PublicKey& public_key,
+                                 const atoms::Merger* merger) {
+    RAISE_ERROR("ProtectedAtomDB::add_node(node, public_key) is not implemented yet");
+}
+
+string ProtectedAtomDB::add_link(const atoms::Link* link,
+                                 const atomdb_api_types::PublicKey& public_key,
+                                 const atoms::Merger* merger) {
+    RAISE_ERROR("ProtectedAtomDB::add_link(link, public_key) is not implemented yet");
+}
+
+vector<string> ProtectedAtomDB::add_atoms(const vector<atoms::Atom*>& atom_list,
+                                          const atomdb_api_types::PublicKey& public_key,
+                                          bool is_transactional,
+                                          const atoms::Merger* merger) {
+    RAISE_ERROR("ProtectedAtomDB::add_atoms(atom_list, public_key) is not implemented yet");
+}
+
+vector<string> ProtectedAtomDB::add_nodes(const vector<atoms::Node*>& nodes,
+                                          const atomdb_api_types::PublicKey& public_key,
+                                          bool is_transactional,
+                                          const atoms::Merger* merger) {
+    RAISE_ERROR("ProtectedAtomDB::add_nodes(nodes, public_key) is not implemented yet");
+}
+
+vector<string> ProtectedAtomDB::add_links(const vector<atoms::Link*>& links,
+                                          const atomdb_api_types::PublicKey& public_key,
+                                          bool is_transactional,
+                                          const atoms::Merger* merger) {
+    RAISE_ERROR("ProtectedAtomDB::add_links(links, public_key) is not implemented yet");
+}
+
+bool ProtectedAtomDB::delete_atom(const string& handle,
+                                  const atomdb_api_types::PublicKey& public_key,
+                                  bool delete_link_targets) {
+    RAISE_ERROR("ProtectedAtomDB::delete_atom(handle, public_key) is not implemented yet");
+}
+
+bool ProtectedAtomDB::delete_node(const string& handle,
+                                  const atomdb_api_types::PublicKey& public_key,
+                                  bool delete_link_targets) {
+    RAISE_ERROR("ProtectedAtomDB::delete_node(handle, public_key) is not implemented yet");
+}
+
+bool ProtectedAtomDB::delete_link(const string& handle,
+                                  const atomdb_api_types::PublicKey& public_key,
+                                  bool delete_link_targets) {
+    RAISE_ERROR("ProtectedAtomDB::delete_link(handle, public_key) is not implemented yet");
+}
+
+uint ProtectedAtomDB::delete_atoms(const vector<string>& handles,
+                                   const atomdb_api_types::PublicKey& public_key,
+                                   bool delete_link_targets) {
+    RAISE_ERROR("ProtectedAtomDB::delete_atoms(handles, public_key) is not implemented yet");
+}
+
+uint ProtectedAtomDB::delete_nodes(const vector<string>& handles,
+                                   const atomdb_api_types::PublicKey& public_key,
+                                   bool delete_link_targets) {
+    RAISE_ERROR("ProtectedAtomDB::delete_nodes(handles, public_key) is not implemented yet");
+}
+
+uint ProtectedAtomDB::delete_links(const vector<string>& handles,
+                                   const atomdb_api_types::PublicKey& public_key,
+                                   bool delete_link_targets) {
+    RAISE_ERROR("ProtectedAtomDB::delete_links(handles, public_key) is not implemented yet");
+}
+
+void ProtectedAtomDB::re_index_patterns(const atomdb_api_types::PublicKey& public_key,
+                                        bool flush_patterns) {
+    RAISE_ERROR("ProtectedAtomDB::re_index_patterns(public_key) is not implemented yet");
+}
+
+size_t ProtectedAtomDB::node_count(const atomdb_api_types::PublicKey& public_key) const {
+    RAISE_ERROR("ProtectedAtomDB::node_count(public_key) is not implemented yet");
+}
+
+size_t ProtectedAtomDB::link_count(const atomdb_api_types::PublicKey& public_key) const {
+    RAISE_ERROR("ProtectedAtomDB::link_count(public_key) is not implemented yet");
+}
+
+size_t ProtectedAtomDB::atom_count(const atomdb_api_types::PublicKey& public_key) const {
+    RAISE_ERROR("ProtectedAtomDB::atom_count(public_key) is not implemented yet");
+}
+
+bool ProtectedAtomDB::allow_nested_indexing() { return this->backend->allow_nested_indexing(); }
+
+bool ProtectedAtomDB::composite_type_enabled() const { return this->backend->composite_type_enabled(); }
+
+atomdb_api_types::ProtectionMode ProtectedAtomDB::get_protection_mode() const {
+    return this->backend->get_protection_mode();
+}
+
+// --------------------------------------------------------------------------------
+// Public methods (without public_key - reject the call)
+
+shared_ptr<Atom> ProtectedAtomDB::get_atom(const string& handle) {
+    raise_public_key_required("get_atom");
+}
+
+shared_ptr<Node> ProtectedAtomDB::get_node(const string& handle) {
+    raise_public_key_required("get_node");
+}
+
+shared_ptr<Link> ProtectedAtomDB::get_link(const string& handle) {
+    raise_public_key_required("get_link");
+}
+
+vector<shared_ptr<Atom>> ProtectedAtomDB::get_matching_atoms(bool is_toplevel, Atom& key) {
+    raise_public_key_required("get_matching_atoms");
+}
+
+shared_ptr<atomdb_api_types::HandleSet> ProtectedAtomDB::query_for_pattern(
+    const LinkSchema& link_schema) {
+    raise_public_key_required("query_for_pattern");
+}
+
+shared_ptr<atomdb_api_types::HandleList> ProtectedAtomDB::query_for_targets(const string& handle) {
+    raise_public_key_required("query_for_targets");
+}
+
+shared_ptr<atomdb_api_types::HandleSet> ProtectedAtomDB::query_for_incoming_set(const string& handle) {
+    raise_public_key_required("query_for_incoming_set");
+}
+
+bool ProtectedAtomDB::atom_exists(const string& handle) { raise_public_key_required("atom_exists"); }
+
+bool ProtectedAtomDB::node_exists(const string& handle) { raise_public_key_required("node_exists"); }
+
+bool ProtectedAtomDB::link_exists(const string& handle) { raise_public_key_required("link_exists"); }
+
+set<string> ProtectedAtomDB::atoms_exist(const vector<string>& handles) {
+    raise_public_key_required("atoms_exist");
+}
+
+set<string> ProtectedAtomDB::nodes_exist(const vector<string>& handles) {
+    raise_public_key_required("nodes_exist");
+}
+
+set<string> ProtectedAtomDB::links_exist(const vector<string>& handles) {
+    raise_public_key_required("links_exist");
+}
+
+string ProtectedAtomDB::add_atom(const atoms::Atom* atom, const atoms::Merger* merger) {
+    raise_public_key_required("add_atom");
+}
+
+string ProtectedAtomDB::add_node(const atoms::Node* node, const atoms::Merger* merger) {
+    raise_public_key_required("add_node");
+}
+
+string ProtectedAtomDB::add_link(const atoms::Link* link, const atoms::Merger* merger) {
+    raise_public_key_required("add_link");
+}
+
+vector<string> ProtectedAtomDB::add_atoms(const vector<atoms::Atom*>& atom_list,
+                                          bool is_transactional,
+                                          const atoms::Merger* merger) {
+    raise_public_key_required("add_atoms");
+}
+
+vector<string> ProtectedAtomDB::add_nodes(const vector<atoms::Node*>& nodes,
+                                          bool is_transactional,
+                                          const atoms::Merger* merger) {
+    raise_public_key_required("add_nodes");
+}
+
+vector<string> ProtectedAtomDB::add_links(const vector<atoms::Link*>& links,
+                                          bool is_transactional,
+                                          const atoms::Merger* merger) {
+    raise_public_key_required("add_links");
+}
+
+bool ProtectedAtomDB::delete_atom(const string& handle, bool delete_link_targets) {
+    raise_public_key_required("delete_atom");
+}
+
+bool ProtectedAtomDB::delete_node(const string& handle, bool delete_link_targets) {
+    raise_public_key_required("delete_node");
+}
+
+bool ProtectedAtomDB::delete_link(const string& handle, bool delete_link_targets) {
+    raise_public_key_required("delete_link");
+}
+
+uint ProtectedAtomDB::delete_atoms(const vector<string>& handles, bool delete_link_targets) {
+    raise_public_key_required("delete_atoms");
+}
+
+uint ProtectedAtomDB::delete_nodes(const vector<string>& handles, bool delete_link_targets) {
+    raise_public_key_required("delete_nodes");
+}
+
+uint ProtectedAtomDB::delete_links(const vector<string>& handles, bool delete_link_targets) {
+    raise_public_key_required("delete_links");
+}
+
+void ProtectedAtomDB::re_index_patterns(bool flush_patterns) {
+    raise_public_key_required("re_index_patterns");
+}
+
+size_t ProtectedAtomDB::node_count() const { raise_public_key_required("node_count"); }
+
+size_t ProtectedAtomDB::link_count() const { raise_public_key_required("link_count"); }
+
+size_t ProtectedAtomDB::atom_count() const { raise_public_key_required("atom_count"); }
+
+// --------------------------------------------------------------------------------
+// Private methods
+
+void ProtectedAtomDB::raise_public_key_required(const string& method_name) {
+    RAISE_ERROR("ProtectedAtomDB::" + method_name +
+                "() is unavailable in protected AtomDBs. Use the public API in ProtectedAtomDB passing "
+                "a PublicKey.");
+}
diff --git a/src/atomdb/ProtectedAtomDB.h b/src/atomdb/ProtectedAtomDB.h
new file mode 100644
index 00000000..5d52f1cc
--- /dev/null
+++ b/src/atomdb/ProtectedAtomDB.h
@@ -0,0 +1,173 @@
+#pragma once
+
+#include <memory>
+#include <set>
+#include <string>
+#include <vector>
+
+#include "AtomDB.h"
+
+using namespace std;
+using namespace atoms;
+
+namespace atomdb {
+
+/**
+ * @brief Authorization wrapper around any AtomDB backend for protected databases.
+ *
+ * Data-access methods expose two forms:
+ * - overloads without PublicKey: reject the call (protected access requires a key)
+ * - overloads with PublicKey: authorize and delegate to the backend
+ *
+ * When the backend reports ProtectionMode::FORWARD, this wrapper forwards the
+ * access key without applying local authorization post-processing.
+ *
+ * get_protection_mode() reports the backend's mode so callers can detect protected
+ * or federated persistence without inspecting the wrapper type.
+ */
+class ProtectedAtomDB : public AtomDB {
+   public:
+    /**
+     * @param backend Shared concrete AtomDB to wrap.
+     */
+    explicit ProtectedAtomDB(shared_ptr<AtomDB> backend);
+
+    bool allow_nested_indexing() override;
+    bool composite_type_enabled() const override;
+    atomdb_api_types::ProtectionMode get_protection_mode() const override;
+
+    shared_ptr<Atom> get_atom(const string& handle) override;
+    shared_ptr<Atom> get_atom(const string& handle, const atomdb_api_types::PublicKey& public_key);
+
+    shared_ptr<Node> get_node(const string& handle) override;
+    shared_ptr<Node> get_node(const string& handle, const atomdb_api_types::PublicKey& public_key);
+
+    shared_ptr<Link> get_link(const string& handle) override;
+    shared_ptr<Link> get_link(const string& handle, const atomdb_api_types::PublicKey& public_key);
+
+    vector<shared_ptr<Atom>> get_matching_atoms(bool is_toplevel, Atom& key) override;
+    vector<shared_ptr<Atom>> get_matching_atoms(bool is_toplevel,
+                                                Atom& key,
+                                                const atomdb_api_types::PublicKey& public_key);
+
+    shared_ptr<atomdb_api_types::HandleSet> query_for_pattern(const LinkSchema& link_schema) override;
+    shared_ptr<atomdb_api_types::HandleSet> query_for_pattern(
+        const LinkSchema& link_schema, const atomdb_api_types::PublicKey& public_key);
+
+    shared_ptr<atomdb_api_types::HandleList> query_for_targets(const string& handle) override;
+    shared_ptr<atomdb_api_types::HandleList> query_for_targets(
+        const string& handle, const atomdb_api_types::PublicKey& public_key);
+
+    shared_ptr<atomdb_api_types::HandleSet> query_for_incoming_set(const string& handle) override;
+    shared_ptr<atomdb_api_types::HandleSet> query_for_incoming_set(
+        const string& handle, const atomdb_api_types::PublicKey& public_key);
+
+    bool atom_exists(const string& handle) override;
+    bool atom_exists(const string& handle, const atomdb_api_types::PublicKey& public_key);
+
+    bool node_exists(const string& handle) override;
+    bool node_exists(const string& handle, const atomdb_api_types::PublicKey& public_key);
+
+    bool link_exists(const string& handle) override;
+    bool link_exists(const string& handle, const atomdb_api_types::PublicKey& public_key);
+
+    set<string> atoms_exist(const vector<string>& handles) override;
+    set<string> atoms_exist(const vector<string>& handles,
+                            const atomdb_api_types::PublicKey& public_key);
+
+    set<string> nodes_exist(const vector<string>& handles) override;
+    set<string> nodes_exist(const vector<string>& handles,
+                            const atomdb_api_types::PublicKey& public_key);
+
+    set<string> links_exist(const vector<string>& handles) override;
+    set<string> links_exist(const vector<string>& handles,
+                            const atomdb_api_types::PublicKey& public_key);
+
+    string add_atom(const atoms::Atom* atom, const atoms::Merger* merger = NULL) override;
+    string add_atom(const atoms::Atom* atom,
+                    const atomdb_api_types::PublicKey& public_key,
+                    const atoms::Merger* merger = NULL);
+
+    string add_node(const atoms::Node* node, const atoms::Merger* merger = NULL) override;
+    string add_node(const atoms::Node* node,
+                    const atomdb_api_types::PublicKey& public_key,
+                    const atoms::Merger* merger = NULL);
+
+    string add_link(const atoms::Link* link, const atoms::Merger* merger = NULL) override;
+    string add_link(const atoms::Link* link,
+                    const atomdb_api_types::PublicKey& public_key,
+                    const atoms::Merger* merger = NULL);
+
+    vector<string> add_atoms(const vector<atoms::Atom*>& atom_list,
+                             bool is_transactional = false,
+                             const atoms::Merger* merger = NULL) override;
+    vector<string> add_atoms(const vector<atoms::Atom*>& atom_list,
+                             const atomdb_api_types::PublicKey& public_key,
+                             bool is_transactional = false,
+                             const atoms::Merger* merger = NULL);
+
+    vector<string> add_nodes(const vector<atoms::Node*>& nodes,
+                             bool is_transactional = false,
+                             const atoms::Merger* merger = NULL) override;
+    vector<string> add_nodes(const vector<atoms::Node*>& nodes,
+                             const atomdb_api_types::PublicKey& public_key,
+                             bool is_transactional = false,
+                             const atoms::Merger* merger = NULL);
+
+    vector<string> add_links(const vector<atoms::Link*>& links,
+                             bool is_transactional = false,
+                             const atoms::Merger* merger = NULL) override;
+    vector<string> add_links(const vector<atoms::Link*>& links,
+                             const atomdb_api_types::PublicKey& public_key,
+                             bool is_transactional = false,
+                             const atoms::Merger* merger = NULL);
+
+    bool delete_atom(const string& handle, bool delete_link_targets = false) override;
+    bool delete_atom(const string& handle,
+                     const atomdb_api_types::PublicKey& public_key,
+                     bool delete_link_targets = false);
+
+    bool delete_node(const string& handle, bool delete_link_targets = false) override;
+    bool delete_node(const string& handle,
+                     const atomdb_api_types::PublicKey& public_key,
+                     bool delete_link_targets = false);
+
+    bool delete_link(const string& handle, bool delete_link_targets = false) override;
+    bool delete_link(const string& handle,
+                     const atomdb_api_types::PublicKey& public_key,
+                     bool delete_link_targets = false);
+
+    uint delete_atoms(const vector<string>& handles, bool delete_link_targets = false) override;
+    uint delete_atoms(const vector<string>& handles,
+                      const atomdb_api_types::PublicKey& public_key,
+                      bool delete_link_targets = false);
+
+    uint delete_nodes(const vector<string>& handles, bool delete_link_targets = false) override;
+    uint delete_nodes(const vector<string>& handles,
+                      const atomdb_api_types::PublicKey& public_key,
+                      bool delete_link_targets = false);
+
+    uint delete_links(const vector<string>& handles, bool delete_link_targets = false) override;
+    uint delete_links(const vector<string>& handles,
+                      const atomdb_api_types::PublicKey& public_key,
+                      bool delete_link_targets = false);
+
+    void re_index_patterns(bool flush_patterns = true) override;
+    void re_index_patterns(const atomdb_api_types::PublicKey& public_key, bool flush_patterns = true);
+
+    size_t node_count() const override;
+    size_t node_count(const atomdb_api_types::PublicKey& public_key) const;
+
+    size_t link_count() const override;
+    size_t link_count(const atomdb_api_types::PublicKey& public_key) const;
+
+    size_t atom_count() const override;
+    size_t atom_count(const atomdb_api_types::PublicKey& public_key) const;
+
+   private:
+    shared_ptr<AtomDB> backend;
+
+    [[noreturn]] static void raise_public_key_required(const string& method_name);
+};
+
+}  // namespace atomdb
diff --git a/src/atomdb/adapterdb/AdapterDB.cc b/src/atomdb/adapterdb/AdapterDB.cc
index aa277fed..064c55ea 100644
--- a/src/atomdb/adapterdb/AdapterDB.cc
+++ b/src/atomdb/adapterdb/AdapterDB.cc
@@ -70,6 +70,11 @@ vector<atomdb_api_types::AccessPermissionDocument> AdapterDB::get_access_permiss
     return this->atomdb_backend->get_access_permissions(public_key);
 }
 
+atomdb_api_types::ProtectionMode AdapterDB::get_protection_mode() const {
+    this->ensure_backend_ready();
+    return this->atomdb_backend->get_protection_mode();
+}
+
 shared_ptr<Atom> AdapterDB::get_atom(const string& handle) {
     this->ensure_backend_ready();
     return this->atomdb_backend->get_atom(handle);
diff --git a/src/atomdb/adapterdb/AdapterDB.h b/src/atomdb/adapterdb/AdapterDB.h
index 2e5d3ab3..464147a7 100644
--- a/src/atomdb/adapterdb/AdapterDB.h
+++ b/src/atomdb/adapterdb/AdapterDB.h
@@ -62,6 +62,8 @@ class AdapterDB : public AtomDB {
      */
     bool composite_type_enabled() const override;
 
+    atomdb_api_types::ProtectionMode get_protection_mode() const override;
+
     shared_ptr<Atom> get_atom(const string& handle) override;
     shared_ptr<Node> get_node(const string& handle) override;
     shared_ptr<Link> get_link(const string& handle) override;
diff --git a/src/atomdb/inmemorydb/InMemoryDB.h b/src/atomdb/inmemorydb/InMemoryDB.h
index 2f99641a..b1361bf1 100644
--- a/src/atomdb/inmemorydb/InMemoryDB.h
+++ b/src/atomdb/inmemorydb/InMemoryDB.h
@@ -49,6 +49,9 @@ class InMemoryDB : public AtomDB {
 
     bool allow_nested_indexing() override;
     bool composite_type_enabled() const override { return false; }
+    atomdb_api_types::ProtectionMode get_protection_mode() const override {
+        return atomdb_api_types::ProtectionMode::UNPROTECTED;
+    }
 
     shared_ptr<Atom> get_atom(const string& handle) override;
     shared_ptr<Node> get_node(const string& handle) override;
diff --git a/src/atomdb/redis_mongodb/RedisMongoDB.cc b/src/atomdb/redis_mongodb/RedisMongoDB.cc
index adc53f87..d8533d8c 100644
--- a/src/atomdb/redis_mongodb/RedisMongoDB.cc
+++ b/src/atomdb/redis_mongodb/RedisMongoDB.cc
@@ -34,6 +34,7 @@ uint RedisMongoDB::REDIS_CHUNK_SIZE;
 string RedisMongoDB::MONGODB_DB_NAME;
 string RedisMongoDB::MONGODB_NODES_COLLECTION_NAME;
 string RedisMongoDB::MONGODB_LINKS_COLLECTION_NAME;
+string RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME;
 string RedisMongoDB::MONGODB_PATTERN_INDEX_SCHEMA_COLLECTION_NAME;
 string RedisMongoDB::MONGODB_ACCESS_PERMISSIONS_COLLECTION_NAME;
 string RedisMongoDB::MONGODB_FIELD_NAME[MONGODB_FIELD::size];
@@ -43,9 +44,11 @@ RedisMongoDB::RedisMongoDB(const string& context, bool skip_redis, const JsonCon
     : context(context),
       skip_redis_(skip_redis),
       composite_type_enabled_(config.at_path("composite_type_enabled").get_or<bool>(true)),
-      cluster_flag(false) {
+      cluster_flag(false),
+      protection_mode(atomdb_api_types::ProtectionMode::PROTECTED) {
     initialize_statics(context);
     mongodb_setup(config);
+    load_protection_mode();
     load_pattern_index_schema();
     redis_setup(config);
     this->patterns_next_score.store(get_next_score(REDIS_PATTERNS_PREFIX + ":next_score"));
@@ -99,6 +102,9 @@ optional<atomdb_api_types::AccessPermissionDocument> RedisMongoDB::load_access_p
     if (!document.contains("public_key") || !document["public_key"].is_string()) {
         RAISE_ERROR("AccessPermissionDocument missing required string filed 'public_key'");
     }
+    if (document["public_key"].get<string>() != public_key) {
+        RAISE_ERROR("AccessPermissionDocument public_key does not match its lookup key");
+    }
     if (!document.contains("full_access") || !document["full_access"].is_boolean()) {
         RAISE_ERROR("AccessPermissionDocument missing required boolean field 'full_access'");
     }
@@ -142,6 +148,10 @@ optional<atomdb_api_types::AccessPermissionDocument> RedisMongoDB::load_access_p
         public_key, document["full_access"].get<bool>(), entries);
 }
 
+atomdb_api_types::ProtectionMode RedisMongoDB::get_protection_mode() const {
+    return this->protection_mode;
+}
+
 void RedisMongoDB::redis_setup(const JsonConfig& config) {
     if (skip_redis_) return;
 
@@ -1317,6 +1327,36 @@ void RedisMongoDB::add_pattern_index_schema(const string& tokens,
     this->pattern_index_schema_next_priority++;
 }
 
+void RedisMongoDB::load_protection_mode() {
+    auto conn = this->mongodb_pool->acquire();
+    auto config_collection = (*conn)[MONGODB_DB_NAME][MONGODB_CONFIG_COLLECTION_NAME];
+    auto config_doc = config_collection.find_one(
+        bsoncxx::v_noabi::builder::basic::make_document(bsoncxx::v_noabi::builder::basic::kvp(
+            MONGODB_FIELD_NAME[MONGODB_FIELD::ID], protection_config_document_id())));
+
+    if (!config_doc) {
+        this->protection_mode = atomdb_api_types::ProtectionMode::UNPROTECTED;
+        return;
+    }
+
+    const auto view = config_doc->view();
+    auto protected_it = view.find("protected");
+    if (protected_it == view.end()) {
+        RAISE_ERROR("RedisMongoDB config document missing required boolean field 'protected'");
+    }
+    if (protected_it->type() != bsoncxx::type::k_bool) {
+        RAISE_ERROR("RedisMongoDB config document field 'protected' must be a boolean");
+    }
+
+    this->protection_mode = protected_it->get_bool().value
+                                ? atomdb_api_types::ProtectionMode::PROTECTED
+                                : atomdb_api_types::ProtectionMode::UNPROTECTED;
+}
+
+string RedisMongoDB::protection_config_document_id() {
+    return Hasher::plain_string_hash(MONGODB_CONFIG_COLLECTION_NAME);
+}
+
 void RedisMongoDB::load_pattern_index_schema() {
     this->pattern_index_schema_map.clear();
     auto conn = this->mongodb_pool->acquire();
@@ -1508,10 +1548,28 @@ void RedisMongoDB::flush_redis_by_prefix(const string& prefix) {
 }
 
 void RedisMongoDB::drop_all() {
+    optional<bsoncxx::document::value> preserved_protection_config;
+    const auto protection_config_id = protection_config_document_id();
+    {
+        auto conn = this->mongodb_pool->acquire();
+        auto config_collection = (*conn)[MONGODB_DB_NAME][MONGODB_CONFIG_COLLECTION_NAME];
+        if (auto config_doc = config_collection.find_one(
+                bsoncxx::v_noabi::builder::basic::make_document(bsoncxx::v_noabi::builder::basic::kvp(
+                    MONGODB_FIELD_NAME[MONGODB_FIELD::ID], protection_config_id)))) {
+            preserved_protection_config = std::move(*config_doc);
+        }
+    }
+
     // Drop MongoDB database
     auto conn = this->mongodb_pool->acquire();
     (*conn)[MONGODB_DB_NAME].drop();
 
+    if (preserved_protection_config) {
+        auto restore_conn = this->mongodb_pool->acquire();
+        auto config_collection = (*restore_conn)[MONGODB_DB_NAME][MONGODB_CONFIG_COLLECTION_NAME];
+        config_collection.insert_one(preserved_protection_config->view());
+    }
+
     // Drop Redis database (by prefixes)
     if (!skip_redis_) {
         auto ctx = this->redis_pool->acquire();
diff --git a/src/atomdb/redis_mongodb/RedisMongoDB.h b/src/atomdb/redis_mongodb/RedisMongoDB.h
index 2a2abf27..710b682a 100644
--- a/src/atomdb/redis_mongodb/RedisMongoDB.h
+++ b/src/atomdb/redis_mongodb/RedisMongoDB.h
@@ -39,6 +39,7 @@ class RedisMongoDB : public AtomDB {
      */
     vector<atomdb_api_types::AccessPermissionDocument> get_access_permissions(
         const atomdb_api_types::PublicKey& public_key) const override;
+    atomdb_api_types::ProtectionMode get_protection_mode() const override;
 
     static string REDIS_PATTERNS_PREFIX;
     static string REDIS_OUTGOING_PREFIX;
@@ -47,6 +48,7 @@ class RedisMongoDB : public AtomDB {
     static string MONGODB_DB_NAME;
     static string MONGODB_NODES_COLLECTION_NAME;
     static string MONGODB_LINKS_COLLECTION_NAME;
+    static string MONGODB_CONFIG_COLLECTION_NAME;
     static string MONGODB_PATTERN_INDEX_SCHEMA_COLLECTION_NAME;
     static string MONGODB_ACCESS_PERMISSIONS_COLLECTION_NAME;
     static string MONGODB_FIELD_NAME[MONGODB_FIELD::size];
@@ -60,6 +62,7 @@ class RedisMongoDB : public AtomDB {
         MONGODB_DB_NAME = context + "das";
         MONGODB_NODES_COLLECTION_NAME = context + "nodes";
         MONGODB_LINKS_COLLECTION_NAME = context + "links";
+        MONGODB_CONFIG_COLLECTION_NAME = context + "config";
         MONGODB_PATTERN_INDEX_SCHEMA_COLLECTION_NAME = context + "pattern_index_schema";
         MONGODB_ACCESS_PERMISSIONS_COLLECTION_NAME = context + "access_permissions";
         MONGODB_FIELD_NAME[MONGODB_FIELD::ID] = "_id";
@@ -162,6 +165,7 @@ class RedisMongoDB : public AtomDB {
     bool skip_redis_;
     bool composite_type_enabled_;
     bool cluster_flag;
+    atomdb_api_types::ProtectionMode protection_mode;
     RedisContextPool* redis_pool;
     mongocxx::pool* mongodb_pool;
     atomic<uint> patterns_next_score{0};
@@ -211,6 +215,8 @@ class RedisMongoDB : public AtomDB {
     void update_incoming_set(const string& key, const string& value);
 
     void load_pattern_index_schema();
+    void load_protection_mode();
+    static string protection_config_document_id();
     vector<string> match_pattern_index_schema(const Link* link);
     vector<vector<string>> index_entries_combinations(unsigned int arity);
 
diff --git a/src/atomdb/remotedb/RemoteAtomDB.cc b/src/atomdb/remotedb/RemoteAtomDB.cc
index 3c2e5ad1..c0470d71 100644
--- a/src/atomdb/remotedb/RemoteAtomDB.cc
+++ b/src/atomdb/remotedb/RemoteAtomDB.cc
@@ -31,6 +31,15 @@ RemoteAtomDB::RemoteAtomDB(map<string, shared_ptr<RemoteAtomDBPeer>> peers)
 
 RemoteAtomDB::~RemoteAtomDB() = default;
 
+atomdb_api_types::ProtectionMode RemoteAtomDB::get_protection_mode() const {
+    for (auto& [uid, peer] : remote_db_) {
+        if (peer->get_protection_mode() != atomdb_api_types::ProtectionMode::UNPROTECTED) {
+            return atomdb_api_types::ProtectionMode::FORWARD;
+        }
+    }
+    return atomdb_api_types::ProtectionMode::UNPROTECTED;
+}
+
 void RemoteAtomDB::finalize_peer_lists() {
     writable_peers_.clear();
     readonly_peers_.clear();
diff --git a/src/atomdb/remotedb/RemoteAtomDB.h b/src/atomdb/remotedb/RemoteAtomDB.h
index 8a75ef3c..30bd41d1 100644
--- a/src/atomdb/remotedb/RemoteAtomDB.h
+++ b/src/atomdb/remotedb/RemoteAtomDB.h
@@ -28,6 +28,7 @@ class RemoteAtomDB : public AtomDB {
 
     bool allow_nested_indexing() override;
     bool composite_type_enabled() const override;
+    atomdb_api_types::ProtectionMode get_protection_mode() const override;
 
     shared_ptr<Atom> get_atom(const string& handle) override;
     shared_ptr<Node> get_node(const string& handle) override;
diff --git a/src/atomdb/remotedb/RemoteAtomDBPeer.cc b/src/atomdb/remotedb/RemoteAtomDBPeer.cc
index a1c7a2a0..c8384569 100644
--- a/src/atomdb/remotedb/RemoteAtomDBPeer.cc
+++ b/src/atomdb/remotedb/RemoteAtomDBPeer.cc
@@ -29,6 +29,10 @@ RemoteAtomDBPeer::RemoteAtomDBPeer(shared_ptr<AtomDB> remote_atomdb,
       read_cache_(make_shared<InMemoryDB>(uid + "_rc")),
       atomdb_(remote_atomdb),
       local_persistence_(local_persistence) {
+    if (local_persistence_ &&
+        local_persistence_->get_protection_mode() != atomdb_api_types::ProtectionMode::UNPROTECTED) {
+        RAISE_ERROR("RemoteAtomDBPeer does not support non-UNPROTECTED local persistence");
+    }
     start_cleanup_thread();
 }
 
@@ -60,6 +64,11 @@ void RemoteAtomDBPeer::invalidate_fetched_templates() {
     fetched_link_templates_.clear();
 }
 
+atomdb_api_types::ProtectionMode RemoteAtomDBPeer::get_protection_mode() const {
+    if (atomdb_) return atomdb_->get_protection_mode();
+    return atomdb_api_types::ProtectionMode::UNPROTECTED;
+}
+
 shared_ptr<Atom> RemoteAtomDBPeer::get_atom(const string& handle) {
     // Snapshot the two in-memory layers without holding the mutex across I/O.
     auto wb = write_buffer();
diff --git a/src/atomdb/remotedb/RemoteAtomDBPeer.h b/src/atomdb/remotedb/RemoteAtomDBPeer.h
index e61d932d..1debda10 100644
--- a/src/atomdb/remotedb/RemoteAtomDBPeer.h
+++ b/src/atomdb/remotedb/RemoteAtomDBPeer.h
@@ -43,6 +43,7 @@ class RemoteAtomDBPeer : public AtomDB, public processor::ThreadMethod {
 
     bool allow_nested_indexing() override;
     bool composite_type_enabled() const override;
+    atomdb_api_types::ProtectionMode get_protection_mode() const override;
 
     shared_ptr<Atom> get_atom(const string& handle) override;
     shared_ptr<Node> get_node(const string& handle) override;
diff --git a/src/tests/cpp/BUILD b/src/tests/cpp/BUILD
index fa3e1a66..74eba0a3 100644
--- a/src/tests/cpp/BUILD
+++ b/src/tests/cpp/BUILD
@@ -850,6 +850,23 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "protected_atomdb_test",
+    size = "small",
+    srcs = ["protected_atomdb_test.cc"],
+    copts = [
+        "-Iexternal/gtest/googletest/include",
+        "-Iexternal/gtest/googletest",
+    ],
+    linkstatic = 1,
+    deps = [
+        "//atomdb:protected_atomdb",
+        "//atomdb/inmemorydb:inmemorydb_lib",
+        "//commons/atoms:atoms_lib",
+        "@com_github_google_googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "atomdb_factory_test",
     size = "medium",
@@ -868,6 +885,7 @@ cc_test(
     linkstatic = 1,
     deps = [
         "//atomdb:atomdb_factory",
+        "//atomdb:protected_atomdb",
         "//atomdb/adapterdb:adapterdb_lib",
         "//atomdb/inmemorydb:inmemorydb_lib",
         "//atomdb/morkdb:morkdb_lib",
@@ -924,6 +942,7 @@ cc_test(
     linkstatic = 1,
     deps = [
         "//atomdb:atomdb_factory",
+        "//atomdb:protected_atomdb",
         "//atomdb/inmemorydb:inmemorydb_lib",
         "//atomdb/remotedb:remotedb_lib",
         "//commons/atoms:atoms_lib",
diff --git a/src/tests/cpp/atomdb_factory_test.cc b/src/tests/cpp/atomdb_factory_test.cc
index 4906e3a9..7aa20f34 100644
--- a/src/tests/cpp/atomdb_factory_test.cc
+++ b/src/tests/cpp/atomdb_factory_test.cc
@@ -11,6 +11,7 @@
 #include "JsonConfig.h"
 #include "MorkDB.h"
 #include "Node.h"
+#include "ProtectedAtomDB.h"
 #include "RemoteAtomDB.h"
 #include "TestAtomDBJsonConfig.h"
 #include "Utils.h"
@@ -21,6 +22,8 @@ using namespace atoms;
 using namespace commons;
 using namespace std;
 
+using atomdb_api_types::ProtectionMode;
+
 namespace {
 
 JsonConfig config_with_type(const string& type) {
@@ -154,3 +157,10 @@ TEST(AtomDBFactoryTest, CreateAdapterDBRequiresBackendType) {
 
     remove(mapping_path.c_str());
 }
+
+TEST(AtomDBFactoryTest, CreateInMemoryDBIsNotProtected) {
+    auto db = AtomDBFactory::create(config_with_type("inmemorydb"), "factory_unprotected_");
+    ASSERT_NE(db, nullptr);
+    EXPECT_EQ(db->get_protection_mode(), ProtectionMode::UNPROTECTED);
+    EXPECT_EQ(dynamic_pointer_cast<ProtectedAtomDB>(db), nullptr);
+}
diff --git a/src/tests/cpp/inmemorydb_test.cc b/src/tests/cpp/inmemorydb_test.cc
index 10f1eb3c..4637a65b 100644
--- a/src/tests/cpp/inmemorydb_test.cc
+++ b/src/tests/cpp/inmemorydb_test.cc
@@ -1376,6 +1376,10 @@ TEST_F(InMemoryDBTest, GetAccessPermissionsReturnsEmpty) {
     EXPECT_TRUE(permissions.empty());
 }
 
+TEST_F(InMemoryDBTest, IsUnprotected) {
+    EXPECT_EQ(db->get_protection_mode(), ProtectionMode::UNPROTECTED);
+}
+
 int main(int argc, char** argv) {
     ::testing::InitGoogleTest(&argc, argv);
     return RUN_ALL_TESTS();
diff --git a/src/tests/cpp/protected_atomdb_test.cc b/src/tests/cpp/protected_atomdb_test.cc
new file mode 100644
index 00000000..205a2bb4
--- /dev/null
+++ b/src/tests/cpp/protected_atomdb_test.cc
@@ -0,0 +1,103 @@
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "InMemoryDB.h"
+#include "Link.h"
+#include "Node.h"
+#include "ProtectedAtomDB.h"
+
+using namespace atomdb;
+using namespace atomdb_api_types;
+using namespace atoms;
+using namespace std;
+
+namespace {
+
+class ProtectedInMemoryDB : public InMemoryDB {
+   public:
+    ProtectedInMemoryDB(const string& context = "") : InMemoryDB(context) {}
+
+    atomdb_api_types::ProtectionMode get_protection_mode() const override {
+        return atomdb_api_types::ProtectionMode::PROTECTED;
+    }
+};
+
+shared_ptr<ProtectedAtomDB> make_protected_db(const string& context = "protected_atomdb_test_") {
+    return make_shared<ProtectedAtomDB>(make_shared<ProtectedInMemoryDB>(context));
+}
+
+}  // namespace
+
+TEST(ProtectedAtomDBTest, RejectsNullBackend) { EXPECT_THROW(ProtectedAtomDB(nullptr), runtime_error); }
+
+TEST(ProtectedAtomDBTest, ReportsProtectionModeThroughWrapper) {
+    auto backend = make_shared<ProtectedInMemoryDB>("protected_flags_");
+    EXPECT_EQ(backend->get_protection_mode(), ProtectionMode::PROTECTED);
+
+    ProtectedAtomDB db(backend);
+    EXPECT_EQ(db.get_protection_mode(), ProtectionMode::PROTECTED);
+}
+
+TEST(ProtectedAtomDBTest, DelegatesToBackend) {
+    auto backend = make_shared<ProtectedInMemoryDB>("protected_flags_");
+    ProtectedAtomDB db(backend);
+    EXPECT_EQ(db.allow_nested_indexing(), backend->allow_nested_indexing());
+    EXPECT_EQ(db.composite_type_enabled(), backend->composite_type_enabled());
+
+    PublicKey key("any_key");
+
+    EXPECT_EQ(db.get_access_permissions(key).size(), backend->get_access_permissions(key).size());
+}
+
+TEST(ProtectedAtomDBTest, RejectsOperationsWithoutPublicKey) {
+    auto db = make_protected_db();
+
+    Node node("Symbol", "\"node\"");
+    Link link("Expression", {"a", "b"});
+    vector<string> handles = {"a", "b"};
+    vector<Atom*> atoms = {&node};
+
+    EXPECT_THROW(db->get_atom("handle"), runtime_error);
+    EXPECT_THROW(db->get_node("handle"), runtime_error);
+    EXPECT_THROW(db->get_link("handle"), runtime_error);
+
+    EXPECT_THROW(db->get_matching_atoms(true, node), runtime_error);
+
+    LinkSchema schema("Expression", 2);
+    EXPECT_THROW(db->query_for_pattern(schema), runtime_error);
+    EXPECT_THROW(db->query_for_targets("handle"), runtime_error);
+    EXPECT_THROW(db->query_for_incoming_set("handle"), runtime_error);
+
+    EXPECT_THROW(db->atom_exists("handle"), runtime_error);
+    EXPECT_THROW(db->node_exists("handle"), runtime_error);
+    EXPECT_THROW(db->link_exists("handle"), runtime_error);
+
+    EXPECT_THROW(db->atoms_exist(handles), runtime_error);
+    EXPECT_THROW(db->nodes_exist(handles), runtime_error);
+    EXPECT_THROW(db->links_exist(handles), runtime_error);
+
+    EXPECT_THROW(db->add_atom(&node), runtime_error);
+    EXPECT_THROW(db->add_node(&node), runtime_error);
+    EXPECT_THROW(db->add_link(&link), runtime_error);
+
+    EXPECT_THROW(db->add_atoms(atoms), runtime_error);
+    EXPECT_THROW(db->add_nodes({&node}), runtime_error);
+    EXPECT_THROW(db->add_links({&link}), runtime_error);
+
+    EXPECT_THROW(db->delete_atom("handle"), runtime_error);
+    EXPECT_THROW(db->delete_node("handle"), runtime_error);
+    EXPECT_THROW(db->delete_link("handle"), runtime_error);
+
+    EXPECT_THROW(db->delete_atoms(handles), runtime_error);
+    EXPECT_THROW(db->delete_nodes(handles), runtime_error);
+    EXPECT_THROW(db->delete_links(handles), runtime_error);
+
+    EXPECT_THROW(db->re_index_patterns(), runtime_error);
+
+    EXPECT_THROW(db->node_count(), runtime_error);
+    EXPECT_THROW(db->link_count(), runtime_error);
+    EXPECT_THROW(db->atom_count(), runtime_error);
+}
diff --git a/src/tests/cpp/redis_mongodb_test.cc b/src/tests/cpp/redis_mongodb_test.cc
index 5b195f09..b208ffe8 100644
--- a/src/tests/cpp/redis_mongodb_test.cc
+++ b/src/tests/cpp/redis_mongodb_test.cc
@@ -32,6 +32,14 @@ using namespace atomdb::atomdb_api_types;
 using namespace atoms;
 using namespace std;
 
+namespace {
+
+string protection_config_document_id() {
+    return Hasher::plain_string_hash(RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME);
+}
+
+}  // namespace
+
 class MockDecoder : public HandleDecoder {
    public:
     map<string, shared_ptr<Atom>> atoms;
@@ -91,6 +99,12 @@ class LinkSchemaHandle : public LinkSchema {
     string fixed_handle;
 };
 
+class TestRedisMongoDB : public RedisMongoDB {
+   public:
+    TestRedisMongoDB(const string& context, const JsonConfig& config)
+        : RedisMongoDB(context, true, config) {}
+};
+
 TEST_F(RedisMongoDBTest, ConcurrentQueryForPattern) {
     const int num_threads = 4;
     vector<thread> threads;
@@ -1478,6 +1492,125 @@ TEST_F(RedisMongoDBTest, GetAccessPermissionsRejectsInvalidDocument) {
     collection.delete_many({});
 }
 
+TEST_F(RedisMongoDBTest, IsProtectedWhenPersistedConfigIsTrue) {
+    using bsoncxx::builder::basic::kvp;
+    using bsoncxx::builder::basic::make_document;
+
+    auto conn = db->get_mongo_pool()->acquire();
+    auto collection =
+        (*conn)[RedisMongoDB::MONGODB_DB_NAME][RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME];
+    collection.delete_many({});
+    collection.insert_one(
+        make_document(kvp("_id", protection_config_document_id()), kvp("protected", true)));
+
+    TestRedisMongoDB loaded("test_", test_atomdb_json_config());
+    EXPECT_EQ(loaded.get_protection_mode(), atomdb_api_types::ProtectionMode::PROTECTED);
+
+    collection.delete_many({});
+}
+
+TEST_F(RedisMongoDBTest, IsUnprotectedWhenPersistedConfigIsFalse) {
+    using bsoncxx::builder::basic::kvp;
+    using bsoncxx::builder::basic::make_document;
+
+    auto conn = db->get_mongo_pool()->acquire();
+    auto collection =
+        (*conn)[RedisMongoDB::MONGODB_DB_NAME][RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME];
+    collection.delete_many({});
+    collection.insert_one(
+        make_document(kvp("_id", protection_config_document_id()), kvp("protected", false)));
+
+    TestRedisMongoDB loaded("test_", test_atomdb_json_config());
+    EXPECT_EQ(loaded.get_protection_mode(), atomdb_api_types::ProtectionMode::UNPROTECTED);
+
+    collection.delete_many({});
+}
+
+TEST_F(RedisMongoDBTest, IsUnprotectedWhenPersistedConfigDocumentAbsent) {
+    auto conn = db->get_mongo_pool()->acquire();
+    auto collection =
+        (*conn)[RedisMongoDB::MONGODB_DB_NAME][RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME];
+    collection.delete_many({});
+
+    TestRedisMongoDB loaded("test_", test_atomdb_json_config());
+    EXPECT_EQ(loaded.get_protection_mode(), atomdb_api_types::ProtectionMode::UNPROTECTED);
+}
+
+TEST_F(RedisMongoDBTest, RejectsPersistedConfigMissingProtectedField) {
+    using bsoncxx::builder::basic::kvp;
+    using bsoncxx::builder::basic::make_document;
+
+    auto conn = db->get_mongo_pool()->acquire();
+    auto collection =
+        (*conn)[RedisMongoDB::MONGODB_DB_NAME][RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME];
+    collection.delete_many({});
+    collection.insert_one(
+        make_document(kvp("_id", protection_config_document_id()), kvp("other", "value")));
+
+    EXPECT_THROW({ TestRedisMongoDB loaded("test_", test_atomdb_json_config()); }, runtime_error);
+
+    collection.delete_many({});
+}
+
+TEST_F(RedisMongoDBTest, RejectsPersistedConfigInvalidProtectedFieldType) {
+    using bsoncxx::builder::basic::kvp;
+    using bsoncxx::builder::basic::make_document;
+
+    auto conn = db->get_mongo_pool()->acquire();
+    auto collection =
+        (*conn)[RedisMongoDB::MONGODB_DB_NAME][RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME];
+    collection.delete_many({});
+    collection.insert_one(
+        make_document(kvp("_id", protection_config_document_id()), kvp("protected", "yes")));
+
+    EXPECT_THROW({ TestRedisMongoDB loaded("test_", test_atomdb_json_config()); }, runtime_error);
+
+    collection.delete_many({});
+}
+
+TEST_F(RedisMongoDBTest, DropAllPreservesProtectionConfiguration) {
+    using bsoncxx::builder::basic::kvp;
+    using bsoncxx::builder::basic::make_document;
+
+    auto conn = db->get_mongo_pool()->acquire();
+    auto collection =
+        (*conn)[RedisMongoDB::MONGODB_DB_NAME][RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME];
+    collection.delete_many({});
+    collection.insert_one(
+        make_document(kvp("_id", protection_config_document_id()), kvp("protected", true)));
+
+    TestRedisMongoDB protected_db("test_", test_atomdb_json_config());
+    ASSERT_EQ(protected_db.get_protection_mode(), atomdb_api_types::ProtectionMode::PROTECTED);
+
+    protected_db.drop_all();
+
+    EXPECT_EQ(protected_db.get_protection_mode(), atomdb_api_types::ProtectionMode::PROTECTED);
+
+    TestRedisMongoDB reloaded("test_", test_atomdb_json_config());
+    EXPECT_EQ(reloaded.get_protection_mode(), atomdb_api_types::ProtectionMode::PROTECTED);
+
+    collection.delete_many({});
+}
+
+TEST_F(RedisMongoDBTest, IgnoresConflictingProtectionConfigurationDocuments) {
+    using bsoncxx::builder::basic::kvp;
+    using bsoncxx::builder::basic::make_document;
+
+    auto conn = db->get_mongo_pool()->acquire();
+    auto collection =
+        (*conn)[RedisMongoDB::MONGODB_DB_NAME][RedisMongoDB::MONGODB_CONFIG_COLLECTION_NAME];
+    collection.delete_many({});
+
+    collection.insert_one(
+        make_document(kvp("_id", protection_config_document_id()), kvp("protected", true)));
+    collection.insert_one(make_document(kvp("protected", false)));
+
+    TestRedisMongoDB loaded("test_", test_atomdb_json_config());
+    EXPECT_EQ(loaded.get_protection_mode(), atomdb_api_types::ProtectionMode::PROTECTED);
+
+    collection.delete_many({});
+}
+
 int main(int argc, char** argv) {
     ::testing::InitGoogleTest(&argc, argv);
     ::testing::AddGlobalTestEnvironment(new RedisMongoDBTestEnvironment());
diff --git a/src/tests/cpp/remote_atomdb_test.cc b/src/tests/cpp/remote_atomdb_test.cc
index eea028d3..759d8d59 100644
--- a/src/tests/cpp/remote_atomdb_test.cc
+++ b/src/tests/cpp/remote_atomdb_test.cc
@@ -20,6 +20,7 @@
 #include "Link.h"
 #include "LinkSchema.h"
 #include "Node.h"
+#include "ProtectedAtomDB.h"
 #include "RemoteAtomDB.h"
 #include "RemoteAtomDBPeer.h"
 
@@ -743,6 +744,21 @@ class CompositeTypeEnabledInMemoryDB : public InMemoryDB {
     bool composite_type_enabled() const override { return true; }
 };
 
+// Backend pointing at a protected database, like a RedisMongoDB whose Mongo config flags it.
+class ProtectedInMemoryDB : public InMemoryDB {
+   public:
+    explicit ProtectedInMemoryDB(const string& context) : InMemoryDB(context) {}
+
+    ProtectionMode get_protection_mode() const override { return ProtectionMode::PROTECTED; }
+};
+
+class ForwardInMemoryDB : public InMemoryDB {
+   public:
+    explicit ForwardInMemoryDB(const string& context) : InMemoryDB(context) {}
+
+    ProtectionMode get_protection_mode() const override { return ProtectionMode::FORWARD; }
+};
+
 TEST(RemoteAtomDBFederationTest, MetadataAggregationFromNestedPeer) {
     auto backend = make_shared<NestedInMemoryDB>("fed_nested_backend_");
     auto handles = populate_inheritance_mammal_links(backend);
@@ -845,6 +861,104 @@ TEST(RemoteAtomDBFederationTest, CompositeTypeEnabledAggregation) {
     }
 }
 
+TEST(RemoteAtomDBFederationTest, PeerIsProtectedFollowsRemoteBackend) {
+    // Neither the remote backend nor the local persistence is protected.
+    {
+        auto remote = make_shared<InMemoryDB>("prot_none_remote_");
+        auto local = make_shared<InMemoryDB>("prot_none_local_");
+        auto peer = make_shared<RemoteAtomDBPeer>(remote, local, "peer");
+        EXPECT_EQ(peer->get_protection_mode(), ProtectionMode::UNPROTECTED);
+    }
+
+    // Read-only peer (no local persistence) over an unprotected remote.
+    {
+        auto remote = make_shared<InMemoryDB>("prot_readonly_remote_");
+        auto peer = make_shared<RemoteAtomDBPeer>(remote, nullptr, "peer");
+        EXPECT_EQ(peer->get_protection_mode(), ProtectionMode::UNPROTECTED);
+    }
+
+    // Protected remote backend.
+    {
+        auto remote = make_shared<ProtectedInMemoryDB>("prot_remote_remote_");
+        auto local = make_shared<InMemoryDB>("prot_remote_local_");
+        auto peer = make_shared<RemoteAtomDBPeer>(remote, local, "peer");
+        EXPECT_EQ(peer->get_protection_mode(), ProtectionMode::PROTECTED);
+    }
+
+    // ProtectedAtomDB wrapper as remote backend propagates protection mode.
+    {
+        auto remote =
+            make_shared<ProtectedAtomDB>(make_shared<ProtectedInMemoryDB>("prot_wrapped_remote_"));
+        auto peer = make_shared<RemoteAtomDBPeer>(remote, nullptr, "peer");
+        EXPECT_EQ(peer->get_protection_mode(), ProtectionMode::PROTECTED);
+    }
+
+    // Protected local persistence is not supported.
+    {
+        auto remote = make_shared<InMemoryDB>("prot_local_remote_");
+        auto local = make_shared<ProtectedInMemoryDB>("prot_local_local_");
+        EXPECT_THROW(make_shared<RemoteAtomDBPeer>(remote, local, "peer"), runtime_error);
+    }
+
+    // FORWARD local persistence is not supported either.
+    {
+        auto remote = make_shared<InMemoryDB>("forward_local_remote_");
+        auto local = make_shared<ForwardInMemoryDB>("forward_local_local_");
+        EXPECT_THROW(make_shared<RemoteAtomDBPeer>(remote, local, "peer"), runtime_error);
+    }
+
+    // Factory-style ProtectedAtomDB wrapper over a protected backend is rejected for local persistence.
+    {
+        auto remote = make_shared<InMemoryDB>("wrapped_local_remote_");
+        auto local =
+            make_shared<ProtectedAtomDB>(make_shared<ProtectedInMemoryDB>("wrapped_local_local_"));
+        EXPECT_THROW(make_shared<RemoteAtomDBPeer>(remote, local, "peer"), runtime_error);
+    }
+}
+
+TEST(RemoteAtomDBFederationTest, IsProtectedWhenAnyPeerIsProtected) {
+    // No peers -> nothing to protect.
+    {
+        map<string, shared_ptr<RemoteAtomDBPeer>> peers;
+        auto db = make_shared<RemoteAtomDB>(peers);
+        EXPECT_EQ(db->get_protection_mode(), ProtectionMode::UNPROTECTED);
+    }
+
+    // All peers unprotected.
+    {
+        auto remote1 = make_shared<InMemoryDB>("fed_prot_off_remote1_");
+        auto remote2 = make_shared<InMemoryDB>("fed_prot_off_remote2_");
+        map<string, shared_ptr<RemoteAtomDBPeer>> peers;
+        peers["peer1"] = make_shared<RemoteAtomDBPeer>(remote1, nullptr, "peer1");
+        peers["peer2"] = make_shared<RemoteAtomDBPeer>(remote2, nullptr, "peer2");
+        auto db = make_shared<RemoteAtomDB>(peers);
+        EXPECT_EQ(db->get_protection_mode(), ProtectionMode::UNPROTECTED);
+    }
+
+    // A single protected peer makes the facade FORWARD (no local post-processing).
+    {
+        auto unprotected_remote = make_shared<InMemoryDB>("fed_prot_mixed_remote_");
+        auto protected_remote = make_shared<ProtectedInMemoryDB>("fed_prot_mixed_protected_");
+        map<string, shared_ptr<RemoteAtomDBPeer>> peers;
+        peers["unprotected"] = make_shared<RemoteAtomDBPeer>(unprotected_remote, nullptr, "unprotected");
+        peers["protected"] = make_shared<RemoteAtomDBPeer>(protected_remote, nullptr, "protected");
+        auto db = make_shared<RemoteAtomDB>(peers);
+        EXPECT_EQ(db->get_protection_mode(), ProtectionMode::FORWARD);
+    }
+
+    // ProtectedAtomDB-wrapped remote peer propagates to the federation facade.
+    {
+        auto unprotected_remote = make_shared<InMemoryDB>("fed_wrapped_remote_");
+        auto protected_remote =
+            make_shared<ProtectedAtomDB>(make_shared<ProtectedInMemoryDB>("fed_wrapped_protected_"));
+        map<string, shared_ptr<RemoteAtomDBPeer>> peers;
+        peers["unprotected"] = make_shared<RemoteAtomDBPeer>(unprotected_remote, nullptr, "unprotected");
+        peers["protected"] = make_shared<RemoteAtomDBPeer>(protected_remote, nullptr, "protected");
+        auto db = make_shared<RemoteAtomDB>(peers);
+        EXPECT_EQ(db->get_protection_mode(), ProtectionMode::FORWARD);
+    }
+}
+
 TEST(RemoteAtomDBFederationTest, CacheFirstProbingAcrossPeers) {
     // An atom that exists only in peer2's backend must still resolve via the facade,
     // and the resolving peer must cache it so subsequent probes are served from cache.
diff --git a/src/tests/cpp/test_commons/mocks/MockAtomDB.h b/src/tests/cpp/test_commons/mocks/MockAtomDB.h
index 95d01299..0b489c7c 100644
--- a/src/tests/cpp/test_commons/mocks/MockAtomDB.h
+++ b/src/tests/cpp/test_commons/mocks/MockAtomDB.h
@@ -5,6 +5,7 @@
 
 using namespace std;
 using namespace atomdb;
+using atomdb_api_types::ProtectionMode;
 
 class MockAtomDocument : public atomdb_api_types::AtomDocument {
    public:
@@ -29,6 +30,7 @@ class AtomDBMock : public AtomDB {
                 get_access_permissions,
                 (const atomdb_api_types::PublicKey& public_key),
                 (const, override));
+    MOCK_METHOD(ProtectionMode, get_protection_mode, (), (const, override));
     MOCK_METHOD(shared_ptr<Atom>, get_atom, (const string& handle), (override));
     MOCK_METHOD(shared_ptr<Node>, get_node, (const string& handle), (override));
     MOCK_METHOD(shared_ptr<Link>, get_link, (const string& handle), (override));
@@ -98,6 +100,8 @@ class AtomDBMock : public AtomDB {
 
     AtomDBMock() {
         ON_CALL(*this, composite_type_enabled()).WillByDefault(::testing::Return(true));
+        ON_CALL(*this, get_protection_mode())
+            .WillByDefault(::testing::Return(ProtectionMode::UNPROTECTED));
         ON_CALL(*this, get_atom(testing::_))
             .WillByDefault(::testing::Return(make_shared<Node>("Node", "TestNode")));
         ON_CALL(*this, get_node(testing::_))