feat(security): enforcement wired into every data handler; blocked on per-call identity

Still NOT shippable, and the guard now says so for a different reason than
before. Handler coverage is done; principal identity is broken.

Enforcement coverage (complete):
gate()/gateFile() is now called from every RPC that reads or writes user data.
Verified by auditing every DatabaseGrpcImpl:: handler rather than working from a
list, which is how the first pass's misses were caught: Upsert, all three
Batch*, all four Set*, and Subscribe - which streams document bodies and, with
an empty `collections` list, means every collection. Subscribe is gated per
event, because there is no single name to authorise up front.

Also closed several information leaks that are not obvious as "reads":
- ListCollections / ListViews / GetMemoryStats filter per entry; a name or a
  document count is information about data the caller may not read.
- ListFiles filters per entry AND recomputes total_count, which would otherwise
  report the number of files in file types the caller has no grant for.
- GetCollectionInfo gates on request->name(), not collection().
- Cross-project and service-wide operations (GetStats, SetReadOnly,
  Create/DropProject) go through requireAnyAdmin(); ListProjects filters.
  The per-project model cannot express these, so the real boundary for them is
  listener separation.

Masks and predicates are applied, not just returned: Get and Find strip masked
columns before serialising, Get honours the row predicate as not-found, Find
AND-merges it, filters naming a masked path are refused (otherwise the mask is
an inference channel), and writes touching a masked field are denied rather than
silently dropped.

Policy management needs no new RPC surface: writes to `_policies` through the
ordinary document API are intercepted, refresh the cache, and route
`__security__` through setSecurity so the lockout guards cannot be bypassed.
Wired on Insert, Upsert and Delete - Upsert was initially missed, and the
symptom was policies that appeared to store and had no effect.

BLOCKER found by the new e2e, and it is a security bug in my own design:
principal identity is per-CONNECTION, not per-call. BearerAuthProcessor stamps
the principal into the gRPC AuthContext, but AuthContext properties belong to
the connection, and gRPC pools channels by target. Three clients with three
distinct named keys all resolved as `ops`, so every check that should have
denied passed. Misattributing one caller's identity to another is worse than
having no identity, so kEnforcementCoverageComplete stays false and security
cannot be enabled.

Fix: resolve the principal per call from request metadata - stop consuming
`authorization` in the processor and map token -> name in a server interceptor
or in the gate itself. tests/load_test/test_policy_enforcement.sh is the
reproducer (currently 8/21).

Unrelated bug fixed on the way: auth.required=true with tls.enabled=false only
logged a WARN, then gRPC aborted the process on SetAuthMetadataProcessor over
insecure credentials ("assertion failed: 0"). Now refused at startup with a
message saying what to do instead.

ctest 17/17 (54 policy-engine assertions). e2e policy enforcement intentionally
red pending the identity fix.

service/src/database_grpc_impl.cpp

@@ -118,7 +118,6 @@ grpc::Status DatabaseGrpcImpl::gate(const grpc::ServerContext* context,
         out = smartbotic::database::Decision{};
         return grpc::Status::OK;
     }
-
     smartbotic::database::ProjectCollection pc;
     try {
         pc = smartbotic::database::parseProjectCollection(qualified);
@@ -163,6 +162,85 @@ grpc::Status DatabaseGrpcImpl::gateFile(const grpc::ServerContext* context,
     return grpc::Status::OK;
 }
 
+bool DatabaseGrpcImpl::handlePolicyWrite(const grpc::ServerContext* context,
+                                          const std::string& collection,
+                                          const std::string& id,
+                                          const nlohmann::json* data,
+                                          bool is_delete,
+                                          grpc::Status& statusOut) {
+    // `collection` arrives qualified; the policy collection is global.
+    const auto sep = collection.find(':');
+    const std::string bare = sep == std::string::npos ? collection
+                                                       : collection.substr(sep + 1);
+    if (bare != PolicyManager::SYSTEM_COLLECTION &&
+        collection != PolicyManager::SYSTEM_COLLECTION) {
+        return false;  // not a policy write
+    }
+
+    // Policy ids are "<project>:<principal>" or "<project>:__security__".
+    const auto idsep = id.find(':');
+    if (idsep == std::string::npos) {
+        statusOut = grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
+                                 "policy id must be '<project>:<principal>'");
+        return true;
+    }
+    const std::string project = id.substr(0, idsep);
+    const std::string tail = id.substr(idsep + 1);
+
+    const std::string principal = smartbotic::database::auth::principalOf(context);
+    if (!policy_manager_.mayAdministerNow(project, principal)) {
+        spdlog::warn("policy: DENIED principal '{}' management of project '{}'",
+                     principal, project);
+        statusOut = grpc::Status(grpc::StatusCode::PERMISSION_DENIED, "access denied");
+        return true;
+    }
+
+    std::string err;
+    bool ok = false;
+    if (tail == PolicyManager::SECURITY_DOC) {
+        if (is_delete) {
+            smartbotic::database::ProjectSecurity off;
+            ok = policy_manager_.setSecurity(project, off, err);
+        } else {
+            smartbotic::database::ProjectSecurity sec;
+            sec.enabled = data && data->value("enabled", false);
+            sec.mode = (data && data->value("mode", std::string("enforce")) == "audit")
+                           ? smartbotic::database::SecurityMode::Audit
+                           : smartbotic::database::SecurityMode::Enforce;
+            // Routed through setSecurity so the no-admin and last-admin guards
+            // apply. A raw insert here would defeat them.
+            ok = policy_manager_.setSecurity(project, sec, err);
+        }
+    } else if (is_delete) {
+        ok = policy_manager_.removePolicy(project, tail, err);
+    } else {
+        if (!data) {
+            statusOut = grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
+                                     "policy body required");
+            return true;
+        }
+        nlohmann::json body = *data;
+        body["principal"] = tail;
+        ok = policy_manager_.setPolicyFromJson(project, body, err);
+    }
+
+    statusOut = ok ? grpc::Status::OK
+                   : grpc::Status(grpc::StatusCode::FAILED_PRECONDITION, err);
+    return true;
+}
+
+grpc::Status DatabaseGrpcImpl::requireAnyAdmin(const grpc::ServerContext* context,
+                                                const char* operation) const {
+    if (!policy_manager_.anyProjectSecured()) return grpc::Status::OK;
+    const std::string principal = smartbotic::database::auth::principalOf(context);
+    // mayAdminister returns true for an unsecured project, so ask about the
+    // secured ones only: being admin of ANY secured project is the bar.
+    if (policy_manager_.isAdminSomewhere(principal)) return grpc::Status::OK;
+    spdlog::warn("policy: DENIED principal '{}' service-wide operation '{}' "
+                 "(requires admin of some secured project)", principal, operation);
+    return grpc::Status(grpc::StatusCode::PERMISSION_DENIED, "access denied");
+}
+
 void DatabaseGrpcImpl::applyMask(smartbotic::database::Document& doc,
                                   const std::vector<std::string>& mask) {
     if (mask.empty()) return;
@@ -173,6 +251,33 @@ void DatabaseGrpcImpl::applyMask(smartbotic::database::Document& doc,
     doc.set_data(data);
 }
 
+grpc::Status DatabaseGrpcImpl::rejectMaskedWrite(
+    const nlohmann::json& data,
+    const std::vector<std::string>& mask) {
+    if (mask.empty() || !data.is_object()) return grpc::Status::OK;
+    for (const auto& m : mask) {
+        // Walk the dot path; if it resolves to anything present in the body,
+        // the write touches a field the caller cannot read.
+        const nlohmann::json* cur = &data;
+        size_t pos = 0;
+        bool present = true;
+        while (pos <= m.size()) {
+            const size_t dot = m.find('.', pos);
+            const std::string seg =
+                m.substr(pos, dot == std::string::npos ? std::string::npos : dot - pos);
+            if (!cur->is_object() || !cur->contains(seg)) { present = false; break; }
+            cur = &(*cur)[seg];
+            if (dot == std::string::npos) break;
+            pos = dot + 1;
+        }
+        if (present) {
+            return grpc::Status(grpc::StatusCode::PERMISSION_DENIED,
+                                "writing field '" + m + "' is not permitted");
+        }
+    }
+    return grpc::Status::OK;
+}
+
 grpc::Status DatabaseGrpcImpl::rejectMaskedFilters(
     const std::vector<smartbotic::database::Filter>& filters,
     const std::vector<std::string>& mask) {
@@ -195,6 +300,12 @@ grpc::Status DatabaseGrpcImpl::Insert(
     const pb::InsertRequest* request,
     pb::InsertResponse* response
 ) {
+    // v2.8.0 — access gate. Write access is gated separately from read: read never implies write.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("Insert", service_);
     }
@@ -207,6 +318,19 @@ grpc::Status DatabaseGrpcImpl::Insert(
     }
     try {
         nlohmann::json data = smartbotic::db::parse_to_nlohmann(request->data());
+        if (auto st = rejectMaskedWrite(data, dec.mask); !st.ok()) return st;
+
+        // v2.8.0 — a write to `_policies` is policy management, not a document
+        // insert: it must refresh the cache and honour the lockout guards.
+        {
+            grpc::Status pst;
+            if (handlePolicyWrite(context, request->collection(), request->id(),
+                                  &data, /*is_delete=*/false, pst)) {
+                if (!pst.ok()) return pst;
+                response->set_id(request->id());
+                return grpc::Status::OK;
+            }
+        }
 
         Document doc;
         doc.id = request->id();
@@ -311,6 +435,16 @@ grpc::Status DatabaseGrpcImpl::Get(
     // Decrypt sensitive fields
     encryption_.decryptSensitiveFields(*doc);
 
+    // v2.8.0 — the policy row predicate can hide the document outright. Same
+    // treatment as a view's where below: not-found, never "exists but
+    // forbidden", since that distinction is itself information.
+    if (!dec.row.empty() && !store_.matchesFilters(*doc, dec.row)) {
+        response->set_found(false);
+        return grpc::Status::OK;
+    }
+    // Remove masked columns before anything is serialised out.
+    applyMask(*doc, dec.mask);
+
     // Apply view's where filters — if doc doesn't match, treat as not found
     if (view && !view->where.empty()) {
         if (!store_.matchesFilters(*doc, view->where)) {
@@ -341,6 +475,12 @@ grpc::Status DatabaseGrpcImpl::Update(
     const pb::UpdateRequest* request,
     pb::UpdateResponse* response
 ) {
+    // v2.8.0 — access gate. Write access is gated separately from read: read never implies write.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         response->set_success(false);
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
@@ -424,6 +564,12 @@ grpc::Status DatabaseGrpcImpl::PatchDocument(
     const pb::PatchDocumentRequest* request,
     pb::PatchDocumentResponse* response
 ) {
+    // v2.8.0 — access gate. Write access is gated separately from read: read never implies write.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         response->set_success(false);
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
@@ -479,10 +625,16 @@ grpc::Status DatabaseGrpcImpl::PatchDocument(
 }
 
 grpc::Status DatabaseGrpcImpl::Upsert(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::UpsertRequest* request,
     pb::UpsertResponse* response
 ) {
+    // v2.8.0 — access gate. Upsert writes; gated exactly like Insert.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("Upsert", service_);
     }
@@ -495,6 +647,21 @@ grpc::Status DatabaseGrpcImpl::Upsert(
     }
     try {
         nlohmann::json data = smartbotic::db::parse_to_nlohmann(request->data());
+        if (auto st = rejectMaskedWrite(data, dec.mask); !st.ok()) return st;
+
+        // v2.8.0 — Upsert is a write path too. Policy management must be
+        // intercepted on EVERY write path or an edit silently lands as a plain
+        // document: the guards are skipped and the cache is never refreshed, so
+        // the policy appears to be stored and has no effect.
+        {
+            grpc::Status pst;
+            if (handlePolicyWrite(context, request->collection(), request->id(),
+                                  &data, /*is_delete=*/false, pst)) {
+                if (!pst.ok()) return pst;
+                response->set_id(request->id());
+                return grpc::Status::OK;
+            }
+        }
 
         bool existed = store_.exists(request->collection(), request->id());
 
@@ -558,6 +725,24 @@ grpc::Status DatabaseGrpcImpl::Delete(
     const pb::DeleteRequest* request,
     pb::DeleteResponse* response
 ) {
+    // v2.8.0 — a delete on `_policies` is policy management. Handled before the
+    // ordinary gate so the last-admin guard applies.
+    {
+        grpc::Status pst;
+        if (handlePolicyWrite(context, request->collection(), request->id(),
+                              nullptr, /*is_delete=*/true, pst)) {
+            if (!pst.ok()) return pst;
+            response->set_deleted(true);
+            return grpc::Status::OK;
+        }
+    }
+
+    // v2.8.0 — access gate. Write access is gated separately from read: read never implies write.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("Delete", service_);
     }
@@ -590,6 +775,12 @@ grpc::Status DatabaseGrpcImpl::Exists(
     const pb::ExistsRequest* request,
     pb::ExistsResponse* response
 ) {
+    // v2.8.0 — access gate. Checked before anything else, so an unauthorised caller learns nothing about server state.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     // v2.0 Stage 4 — LMDB-first read. Gated on mirror_healthy_ + zero
     // drift + non-system collection + doc_store_ available. The
     // defensive MemoryStore-on-LMDB-miss fallback was removed once the
@@ -632,6 +823,12 @@ grpc::Status DatabaseGrpcImpl::GetVersionHistory(
     const pb::GetVersionHistoryRequest* request,
     pb::GetVersionHistoryResponse* response
 ) {
+    // v2.8.0 — access gate. History returns previous document bodies; ungated it would bypass column masks.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     auto result = store_.getVersionHistory(
         request->collection(), request->id(),
         request->limit(), request->offset());
@@ -648,10 +845,16 @@ grpc::Status DatabaseGrpcImpl::GetVersionHistory(
 }
 
 grpc::Status DatabaseGrpcImpl::GetDocumentVersion(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetDocumentVersionRequest* request,
     pb::GetDocumentVersionResponse* response
 ) {
+    // v2.8.0 — access gate. Checked before anything else, so an unauthorised caller learns nothing about server state.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     auto ver = store_.getDocumentAtVersion(
         request->collection(), request->id(), request->version());
 
@@ -667,10 +870,16 @@ grpc::Status DatabaseGrpcImpl::GetDocumentVersion(
 }
 
 grpc::Status DatabaseGrpcImpl::RestoreVersion(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::RestoreVersionRequest* request,
     pb::RestoreVersionResponse* response
 ) {
+    // v2.8.0 — access gate. Write access is gated separately from read: read never implies write.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         response->set_success(false);
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
@@ -698,10 +907,16 @@ grpc::Status DatabaseGrpcImpl::RestoreVersion(
 }
 
 grpc::Status DatabaseGrpcImpl::RestoreToDate(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::RestoreToDateRequest* request,
     pb::RestoreToDateResponse* response
 ) {
+    // v2.8.0 — access gate. Write access is gated separately from read: read never implies write.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         response->set_success(false);
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
@@ -732,10 +947,16 @@ grpc::Status DatabaseGrpcImpl::RestoreToDate(
 // ===== Batch Operations =====
 
 grpc::Status DatabaseGrpcImpl::BatchInsert(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::BatchInsertRequest* request,
     pb::BatchInsertResponse* response
 ) {
+    // v2.8.0 — access gate. A batch is still a write - gated once for the whole batch, which shares one collection.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("BatchInsert", service_);
     }
@@ -785,10 +1006,16 @@ grpc::Status DatabaseGrpcImpl::BatchInsert(
 }
 
 grpc::Status DatabaseGrpcImpl::BatchGet(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::BatchGetRequest* request,
     pb::BatchGetResponse* response
 ) {
+    // v2.8.0 — access gate. Batch reads must mask and filter like Get.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     for (const auto& id : request->ids()) {
         auto doc = store_.get(request->collection(), id);
         if (doc) {
@@ -801,10 +1028,16 @@ grpc::Status DatabaseGrpcImpl::BatchGet(
 }
 
 grpc::Status DatabaseGrpcImpl::BatchDelete(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::BatchDeleteRequest* request,
     pb::BatchDeleteResponse* response
 ) {
+    // v2.8.0 — access gate. Batch delete is a write.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("BatchDelete", service_);
     }
@@ -864,6 +1097,20 @@ grpc::Status DatabaseGrpcImpl::Find(
     // construction: Insert ACK is gated on the dual-write mirror commit,
     // so any ACK'd write is visible to subsequent LMDB scans.
     // v2.3 — parse for project routing (same shape as Get / Exists).
+    // v2.8.0 — access gate. After view resolution so policy applies to the
+    // UNDERLYING collection: a view must not become a way around a rule.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, targetCollection, smartbotic::database::Access::Read, dec);
+        !st.ok()) {
+        return st;
+    }
+    // A filter naming a masked path is refused: `salary > 100000` would leak a
+    // value the caller is not permitted to read.
+    if (auto st = rejectMaskedFilters(query.filters, dec.mask); !st.ok()) return st;
+    // Row predicate AND-merges with the caller's filters, exactly as a view's
+    // `where` already does.
+    for (const auto& f : dec.row) query.filters.push_back(f);
+
     QueryResult result;
     bool used_lmdb = false;
     if (!targetCollection.empty() && targetCollection[0] != '_'
@@ -901,6 +1148,9 @@ grpc::Status DatabaseGrpcImpl::Find(
 
     for (auto& doc : result.documents) {
         encryption_.decryptSensitiveFields(doc);
+        // v2.8.0 — mask before the view projection runs, so a view cannot
+        // re-expose a column that policy removed.
+        applyMask(doc, dec.mask);
         pb::Document protoDoc = toProto(doc);
         if (view) {
             nlohmann::json docJson = smartbotic::db::parse_to_nlohmann(protoDoc.data());
@@ -998,6 +1248,12 @@ grpc::Status DatabaseGrpcImpl::SimilaritySearch(
     const pb::SimilaritySearchRequest* request,
     pb::SimilaritySearchResponse* response
 ) {
+    // v2.8.0 — access gate. Checked before anything else, so an unauthorised caller learns nothing about server state.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     try {
         if (request->collection().empty()) {
             return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, "Collection name is required");
@@ -1174,10 +1430,16 @@ grpc::Status DatabaseGrpcImpl::SimilaritySearch(
 // ===== Set Operations =====
 
 grpc::Status DatabaseGrpcImpl::SetAdd(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::SetAddRequest* request,
     pb::SetAddResponse* response
 ) {
+    // v2.8.0 — access gate. Set operations mutate collection data.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("SetAdd", service_);
     }
@@ -1190,10 +1452,16 @@ grpc::Status DatabaseGrpcImpl::SetAdd(
 }
 
 grpc::Status DatabaseGrpcImpl::SetRemove(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::SetRemoveRequest* request,
     pb::SetRemoveResponse* response
 ) {
+    // v2.8.0 — access gate. Set operations mutate collection data.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("SetRemove", service_);
     }
@@ -1206,10 +1474,16 @@ grpc::Status DatabaseGrpcImpl::SetRemove(
 }
 
 grpc::Status DatabaseGrpcImpl::SetMembers(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::SetMembersRequest* request,
     pb::SetMembersResponse* response
 ) {
+    // v2.8.0 — access gate. Set membership is collection data.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     auto members = store_.setMembers(request->collection(), request->set_id());
     for (const auto& member : members) {
         response->add_members(member);
@@ -1218,10 +1492,16 @@ grpc::Status DatabaseGrpcImpl::SetMembers(
 }
 
 grpc::Status DatabaseGrpcImpl::SetIsMember(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::SetIsMemberRequest* request,
     pb::SetIsMemberResponse* response
 ) {
+    // v2.8.0 — access gate. Set membership is collection data.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     bool isMember = store_.setIsMember(request->collection(), request->set_id(), request->member());
     response->set_is_member(isMember);
     return grpc::Status::OK;
@@ -1230,10 +1510,16 @@ grpc::Status DatabaseGrpcImpl::SetIsMember(
 // ===== Collection Management =====
 
 grpc::Status DatabaseGrpcImpl::CreateCollection(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::CreateCollectionRequest* request,
     pb::CreateCollectionResponse* response
 ) {
+    // v2.8.0 — access gate. Creating a collection in a project is a write to that project.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->name(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("CreateCollection", service_);
     }
@@ -1264,10 +1550,16 @@ grpc::Status DatabaseGrpcImpl::CreateCollection(
 }
 
 grpc::Status DatabaseGrpcImpl::DropCollection(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::DropCollectionRequest* request,
     pb::DropCollectionResponse* response
 ) {
+    // v2.8.0 — access gate. Destructive; must never be reachable without a write grant.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->name(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("DropCollection", service_);
     }
@@ -1285,12 +1577,20 @@ grpc::Status DatabaseGrpcImpl::DropCollection(
 }
 
 grpc::Status DatabaseGrpcImpl::ListCollections(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::ListCollectionsRequest* /*request*/,
     pb::ListCollectionsResponse* response
 ) {
     auto names = store_.listCollections();
+    // v2.8.0 — filter per entry rather than refusing the call. The mere
+    // existence of a collection is information: an enumeration that shows names
+    // a principal cannot read leaks the shape of the dataset.
     for (const auto& name : names) {
+        smartbotic::database::Decision ldec;
+        if (auto st = gate(context, name, smartbotic::database::Access::Read, ldec);
+            !st.ok()) {
+            continue;
+        }
         response->add_names(name);
     }
     return grpc::Status::OK;
@@ -1301,6 +1601,14 @@ grpc::Status DatabaseGrpcImpl::GetCollectionInfo(
     const pb::GetCollectionInfoRequest* request,
     pb::GetCollectionInfoResponse* response
 ) {
+    // v2.8.0 — access gate. GetCollectionInfo names the collection in `name`,
+    // not `collection`. Gated because documentCount and sizeBytes are
+    // information about data the caller may not be permitted to read.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->name(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     auto info = store_.getCollectionInfo(request->name());
     if (!info) {
         response->set_found(false);
@@ -1346,22 +1654,37 @@ grpc::Status DatabaseGrpcImpl::GetCollectionInfo(
 // without parsing status strings.
 
 grpc::Status DatabaseGrpcImpl::ListProjects(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::ListProjectsRequest* /*request*/,
     pb::ListProjectsResponse* response
 ) {
     auto names = service_.listProjects();
+    // v2.8.0 — a project name is information. Once any project is secured, show
+    // a caller only the projects it has some grant in; unsecured projects stay
+    // visible to everyone, matching pre-2.8.0 behaviour.
+    const bool secured = policy_manager_.anyProjectSecured();
+    const std::string principal =
+        secured ? smartbotic::database::auth::principalOf(context) : std::string();
     for (auto& name : names) {
+        if (secured) {
+            const auto sec = policy_manager_.securityOf(name);
+            if (sec.enabled && !policy_manager_.getPolicy(name, principal).has_value()) {
+                continue;
+            }
+        }
         response->add_projects(std::move(name));
     }
     return grpc::Status::OK;
 }
 
 grpc::Status DatabaseGrpcImpl::CreateProject(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::CreateProjectRequest* request,
     pb::CreateProjectResponse* response
 ) {
+    // v2.8.0 — service-wide/cross-project: no single project to gate on.
+    if (auto st = requireAnyAdmin(context, "CreateProject"); !st.ok()) return st;
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("CreateProject", service_);
     }
@@ -1375,10 +1698,13 @@ grpc::Status DatabaseGrpcImpl::CreateProject(
 }
 
 grpc::Status DatabaseGrpcImpl::DropProject(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::DropProjectRequest* request,
     pb::DropProjectResponse* response
 ) {
+    // v2.8.0 — service-wide/cross-project: no single project to gate on.
+    if (auto st = requireAnyAdmin(context, "DropProject"); !st.ok()) return st;
+
     if (service_.isReadOnly()) {
         return readOnlyStatus("DropProject", service_);
     }
@@ -1433,6 +1759,15 @@ grpc::Status DatabaseGrpcImpl::UploadFile(
         info.isPublic = metadata.is_public();
         // v2.6.0 — empty project means the default namespace.
         info.project = metadata.project().empty() ? "default" : metadata.project();
+        {
+            // v2.8.0 — gate the upload on the declared file type.
+            smartbotic::database::Decision fdec;
+            if (auto st = gateFile(context, info.project, info.fileType,
+                                   smartbotic::database::Access::Write, fdec);
+                !st.ok()) {
+                return st;
+            }
+        }
         for (const auto& [key, value] : metadata.metadata()) {
             info.metadata[key] = value;
         }
@@ -1471,6 +1806,17 @@ grpc::Status DatabaseGrpcImpl::DownloadFile(
         if (!info) {
             return grpc::Status(grpc::StatusCode::NOT_FOUND, "File not found");
         }
+        // v2.8.0 — gate on the record's file type, which is only known after
+        // the lookup. A denial is reported as NOT_FOUND, matching the
+        // cross-project rule: "exists but forbidden" is itself information.
+        {
+            smartbotic::database::Decision fdec;
+            if (auto st = gateFile(context, info->project, info->fileType,
+                                   smartbotic::database::Access::Read, fdec);
+                !st.ok()) {
+                return grpc::Status(grpc::StatusCode::NOT_FOUND, "File not found");
+            }
+        }
 
         // Send metadata first
         pb::FileChunk metadataChunk;
@@ -1515,6 +1861,18 @@ grpc::Status DatabaseGrpcImpl::DeleteFile(
     if (store_.pressure() == MemoryPressure::Emergency) {
         return memoryEmergencyStatus("DeleteFile", store_);
     }
+    // v2.8.0 — gate the delete on the record's file type. Resolve first so we
+    // know the type; report a denial as "not deleted" rather than distinguishing
+    // it from a missing file.
+    if (auto info = files_.getFileInfoIn(request->project(), request->id())) {
+        smartbotic::database::Decision fdec;
+        if (auto st = gateFile(context, info->project, info->fileType,
+                               smartbotic::database::Access::Write, fdec);
+            !st.ok()) {
+            response->set_deleted(false);
+            return grpc::Status::OK;
+        }
+    }
     // v2.6.0 — scoped by project; a cross-project id deletes nothing.
     bool deleted = files_.deleteFileIn(request->project(), request->id());
     response->set_deleted(deleted);
@@ -1530,6 +1888,16 @@ grpc::Status DatabaseGrpcImpl::GetFileInfo(
     if (!info) {
         return grpc::Status(grpc::StatusCode::NOT_FOUND, "File not found");
     }
+    {
+        // v2.8.0 — same shape as DownloadFile: gate on the resolved file type,
+        // report a denial as NOT_FOUND.
+        smartbotic::database::Decision fdec;
+        if (auto st = gateFile(context, info->project, info->fileType,
+                               smartbotic::database::Access::Read, fdec);
+            !st.ok()) {
+            return grpc::Status(grpc::StatusCode::NOT_FOUND, "File not found");
+        }
+    }
 
     response->set_project(info->project);
     response->set_id(info->id);
@@ -1562,10 +1930,20 @@ grpc::Status DatabaseGrpcImpl::ListFiles(
         request->checksum(), request->name()
     );
 
-    response->set_total_count(result.totalCount);
     response->set_has_more(result.hasMore);
 
     for (const auto& info : result.files) {
+        // v2.8.0 — filter per entry rather than refusing the whole listing, so a
+        // principal granted one file type still gets a usable result. total_count
+        // is corrected below.
+        {
+            smartbotic::database::Decision fdec;
+            if (auto st = gateFile(context, info.project, info.fileType,
+                                   smartbotic::database::Access::Read, fdec);
+                !st.ok()) {
+                continue;
+            }
+        }
         auto* file = response->add_files();
         file->set_id(info.id);
         file->set_name(info.name);
@@ -1579,6 +1957,10 @@ grpc::Status DatabaseGrpcImpl::ListFiles(
         file->set_project(info.project);
     }
 
+    // v2.8.0 — total_count must reflect what the caller may actually see, not
+    // what the store holds. Reporting the unfiltered total would leak the number
+    // of files in file types this principal has no grant for.
+    response->set_total_count(static_cast<uint64_t>(response->files_size()));
     return grpc::Status::OK;
 }
 
@@ -1600,9 +1982,24 @@ grpc::Status DatabaseGrpcImpl::Subscribe(
     std::vector<std::string> patterns(request->patterns().begin(), request->patterns().end());
     bool includeData = request->include_data();
 
-    uint64_t subId = events_.subscribe(collections, patterns, [writer, includeData, context](const DatabaseEvent& event) {
+    // v2.8.0 — Subscribe is gated PER EVENT, not once at entry. An empty
+    // `collections` list means "every collection" and `patterns` accepts globs,
+    // so there is no single name to authorise up front. Filtering in the
+    // callback is also what makes a partial grant work: a principal subscribed
+    // to everything receives only the collections it may read.
+    uint64_t subId = events_.subscribe(collections, patterns,
+                                        [this, writer, includeData, context](const DatabaseEvent& event) {
         if (context->IsCancelled()) return;
 
+        smartbotic::database::Decision edec;
+        if (auto st = gate(context, event.collection,
+                           smartbotic::database::Access::Read, edec);
+            !st.ok()) {
+            // Silently skip. Emitting an error here would confirm that a
+            // collection the caller cannot read just changed.
+            return;
+        }
+
         pb::DatabaseEvent protoEvent;
         protoEvent.set_type(static_cast<pb::EventType>(static_cast<int>(event.type) + 1));
         protoEvent.set_collection(event.collection);
@@ -1611,8 +2008,15 @@ grpc::Status DatabaseGrpcImpl::Subscribe(
         protoEvent.set_node_id(event.nodeId);
 
         if (includeData && event.data) {
-            std::string dataStr = event.data->dump();
-            protoEvent.set_data(dataStr);
+            // Masked columns must not ride out through the event stream - that
+            // would be the simplest possible way around a column mask.
+            if (edec.mask.empty()) {
+                protoEvent.set_data(event.data->dump());
+            } else {
+                nlohmann::json masked =
+                    applyProjection(*event.data, /*include=*/{}, /*exclude=*/edec.mask);
+                protoEvent.set_data(masked.dump());
+            }
         }
 
         writer->Write(protoEvent);
@@ -1652,10 +2056,13 @@ grpc::Status DatabaseGrpcImpl::HealthCheck(
 }
 
 grpc::Status DatabaseGrpcImpl::GetStats(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetStatsRequest* /*request*/,
     pb::GetStatsResponse* response
 ) {
+    // v2.8.0 — service-wide/cross-project: no single project to gate on.
+    if (auto st = requireAnyAdmin(context, "GetStats"); !st.ok()) return st;
+
     auto stats = store_.getStats();
     auto persistStats = persistence_.getStats();
     const auto& config = store_.getConfig();
@@ -1697,7 +2104,7 @@ grpc::Status DatabaseGrpcImpl::GetStats(
 }
 
 grpc::Status DatabaseGrpcImpl::GetMemoryStats(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetMemoryStatsRequest* /*request*/,
     pb::GetMemoryStatsResponse* response
 ) {
@@ -1723,6 +2130,17 @@ grpc::Status DatabaseGrpcImpl::GetMemoryStats(
     };
 
     for (const auto& c : snap.collections) {
+        // v2.8.0 — per-collection document counts and byte sizes are
+        // information about data the caller may not read, so the same filter
+        // applies here as to ListCollections.
+        {
+            smartbotic::database::Decision sdec;
+            if (auto st = gate(context, c.collection,
+                               smartbotic::database::Access::Read, sdec);
+                !st.ok()) {
+                continue;
+            }
+        }
         auto* pbc = response->add_collections();
         pbc->set_collection(c.collection);
         pbc->set_document_count(c.documentCount);
@@ -1838,10 +2256,16 @@ Query DatabaseGrpcImpl::fromProtoQuery(const pb::FindRequest& request) {
 // ===== View Operations =====
 
 grpc::Status DatabaseGrpcImpl::CreateView(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::CreateViewRequest* request,
     pb::CreateViewResponse* response
 ) {
+    // v2.8.0 — access gate. A view over a collection you cannot write is a schema write to that project.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         response->set_success(false);
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
@@ -1894,10 +2318,16 @@ grpc::Status DatabaseGrpcImpl::CreateView(
 }
 
 grpc::Status DatabaseGrpcImpl::DropView(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::DropViewRequest* request,
     pb::DropViewResponse* response
 ) {
+    // v2.8.0 — access gate. Dropping a view is a schema write.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->name(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         response->set_success(false);
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
@@ -1921,7 +2351,7 @@ grpc::Status DatabaseGrpcImpl::DropView(
 }
 
 grpc::Status DatabaseGrpcImpl::ListViews(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::ListViewsRequest* request,
     pb::ListViewsResponse* response
 ) {
@@ -1934,6 +2364,16 @@ grpc::Status DatabaseGrpcImpl::ListViews(
             && smartbotic::database::resolveCollection(v.name).project != wantProject) {
             continue;
         }
+        // v2.8.0 — a view is a window onto a collection, so listing it leaks
+        // that collection's existence and shape. Filter per entry.
+        {
+            smartbotic::database::Decision vdec;
+            if (auto st = gate(context, v.collection,
+                               smartbotic::database::Access::Read, vdec);
+                !st.ok()) {
+                continue;
+            }
+        }
         auto* out = response->add_views();
         out->set_name(v.name);
         out->set_collection(v.collection);
@@ -1957,10 +2397,16 @@ grpc::Status DatabaseGrpcImpl::ListViews(
 }
 
 grpc::Status DatabaseGrpcImpl::GetViewInfo(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetViewInfoRequest* request,
     pb::GetViewInfoResponse* response
 ) {
+    // v2.8.0 — access gate. A view definition describes a collection's shape.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->name(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     auto v = view_manager_.getView(request->name());
     if (!v) {
         response->set_found(false);
@@ -2114,10 +2560,16 @@ grpc::Status DatabaseReplicationGrpcImpl::GetNodeState(
 // ===== Collection Config Operations =====
 
 grpc::Status DatabaseGrpcImpl::ConfigureCollection(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::ConfigureCollectionRequest* request,
     pb::ConfigureCollectionResponse* response
 ) {
+    // v2.8.0 — access gate. Config changes are writes - versioning and precision affect stored data.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         response->set_success(false);
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
@@ -2162,10 +2614,16 @@ grpc::Status DatabaseGrpcImpl::ConfigureCollection(
 }
 
 grpc::Status DatabaseGrpcImpl::GetCollectionConfig(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetCollectionConfigRequest* request,
     pb::GetCollectionConfigResponse* response
 ) {
+    // v2.8.0 — access gate. Collection config describes stored data.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+
     auto cfg = config_manager_.configFor(request->collection());
     response->mutable_config()->set_timestamp_precision(cfg.timestampPrecision);
     response->mutable_config()->set_versioning_enabled(cfg.versioningEnabled);
@@ -2174,10 +2632,16 @@ grpc::Status DatabaseGrpcImpl::GetCollectionConfig(
 }
 
 grpc::Status DatabaseGrpcImpl::MigrateCollectionTimestamps(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::MigrateCollectionTimestampsRequest* request,
     pb::MigrateCollectionTimestampsResponse* response
 ) {
+    // v2.8.0 — access gate. Rewrites every document's timestamps.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(), smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+
     if (service_.isReadOnly()) {
         response->set_success(false);
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
@@ -2278,10 +2742,13 @@ grpc::Status DatabaseGrpcImpl::MigrateCollectionTimestamps(
 // ===== Read-Only Control =====
 
 grpc::Status DatabaseGrpcImpl::SetReadOnly(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::SetReadOnlyRequest* request,
     pb::SetReadOnlyResponse* response
 ) {
+    // v2.8.0 — service-wide/cross-project: no single project to gate on.
+    if (auto st = requireAnyAdmin(context, "SetReadOnly"); !st.ok()) return st;
+
     bool was_readonly = service_.isReadOnly();
     std::string reason = request->read_only()
         ? "manually locked via SetReadOnly RPC"

service/src/database_grpc_impl.hpp

@@ -396,10 +396,47 @@ private:
                           smartbotic::database::Access access,
                           smartbotic::database::Decision& out) const;
 
+    // v2.8.0 — policy management goes through the ordinary document API on the
+    // `_policies` collection, so there is no separate RPC surface to keep in
+    // step. Two things have to happen that a plain insert would not do:
+    //
+    //   1. the in-memory policy cache must be refreshed, or an edit would not
+    //      take effect until the next restart;
+    //   2. a write to a `__security__` record must go through
+    //      PolicyManager::setSecurity(), or a direct write would bypass the
+    //      lockout guards (no-admin enable, removing the last admin) - which is
+    //      exactly the failure those guards exist to prevent.
+    //
+    // Returns true when the write was a `_policies` write and has been handled;
+    // `statusOut` then carries the result.
+    bool handlePolicyWrite(const grpc::ServerContext* context,
+                           const std::string& collection,
+                           const std::string& id,
+                           const nlohmann::json* data,
+                           bool is_delete,
+                           grpc::Status& statusOut);
+
+    // Service-wide and cross-project operations have no single project to
+    // authorise against, so the per-project model cannot express them. When any
+    // project is secured they require the caller to be an admin of some project;
+    // otherwise they stay open, preserving pre-2.8.0 behaviour.
+    //
+    // This is a coarse control on purpose. The real boundary for operator
+    // surfaces is listener separation - do not expose an admin listener
+    // publicly. Documented in the integration guide.
+    grpc::Status requireAnyAdmin(const grpc::ServerContext* context,
+                                 const char* operation) const;
+
     // Strip masked paths from a document's data in place. No-op on empty mask.
     static void applyMask(smartbotic::database::Document& doc,
                           const std::vector<std::string>& mask);
 
+    // Reject a write whose body sets a masked path. Silently dropping the field
+    // would be worse: the caller believes it wrote a value that never landed,
+    // and on a masked audit column that is a way to forge history.
+    static grpc::Status rejectMaskedWrite(const nlohmann::json& data,
+                                          const std::vector<std::string>& mask);
+
     // Reject a filter that names a masked path. Allowing it would turn the mask
     // into an inference channel: `salary > 100000` leaks a value the caller is
     // not permitted to read.

service/src/database_service.cpp

@@ -1321,11 +1321,20 @@ void DatabaseService::startGrpcServer() {
                 }
             }
             if (!listener.tls.enabled) {
-                spdlog::warn(
-                    "Listener {}: auth.required=true but tls.enabled=false — "
-                    "the bearer token is sent in cleartext over the wire. "
-                    "Consider enabling TLS for any non-loopback listener.",
-                    addr);
+                // gRPC ABORTS the process if an auth metadata processor is
+                // attached to insecure credentials
+                // (insecure_server_credentials.cc: "assertion failed: 0").
+                // This used to be a WARN, which meant a plaintext+auth listener
+                // looked merely inadvisable in the config and then killed the
+                // server on start with an unexplained assert. Refuse it here
+                // with a message that says what to do instead.
+                throw std::runtime_error(
+                    "Listener " + addr + ": auth.required=true requires "
+                    "tls.enabled=true. gRPC does not support an auth metadata "
+                    "processor on insecure credentials and will abort the "
+                    "process. Enable TLS for this listener (set "
+                    "tls.auto_self_signed_if_missing for a local dev cert), or "
+                    "drop auth and rely on binding to loopback.");
             }
             std::vector<smartbotic::database::auth::BearerAuthProcessor::NamedKey> nk;
             nk.reserve(listener.auth.keys.size());

service/src/security/policy_manager.cpp

@@ -276,6 +276,16 @@ bool PolicyManager::mayAdminister(const std::string& project,
     return it != policies_.end() && it->second.admin;
 }
 
+bool PolicyManager::isAdminSomewhere(const std::string& principal) const {
+    std::shared_lock lock(mutex_);
+    for (const auto& [project, sec] : security_) {
+        if (!sec.enabled) continue;
+        auto it = policies_.find(project + ":" + principal);
+        if (it != policies_.end() && it->second.admin) return true;
+    }
+    return false;
+}
+
 bool PolicyManager::setSecurity(const std::string& project,
                                  const ProjectSecurity& sec,
                                  std::string& errorOut) {
@@ -377,6 +387,17 @@ bool PolicyManager::setPolicy(const std::string& project, const Policy& policy,
     return true;
 }
 
+bool PolicyManager::setPolicyFromJson(const std::string& project,
+                                       const nlohmann::json& body,
+                                       std::string& errorOut) {
+    try {
+        return setPolicy(project, policyFromJson(body), errorOut);
+    } catch (const std::exception& e) {
+        errorOut = std::string("malformed policy body: ") + e.what();
+        return false;
+    }
+}
+
 bool PolicyManager::removePolicy(const std::string& project,
                                   const std::string& principal,
                                   std::string& errorOut) {

service/src/security/policy_manager.hpp

@@ -41,6 +41,8 @@
 #include <unordered_map>
 #include <vector>
 
+#include <nlohmann/json.hpp>
+
 #include "document.hpp"
 
 namespace smartbotic::database {
@@ -72,21 +74,41 @@ struct Policy {
     uint64_t updatedAt = 0;
 };
 
-// Set to true ONLY when the access gate is called from every handler that can
-// read or write user data. Until then `setSecurity(enabled=true)` is refused,
-// because a project enforcing on some paths and silently allowing on others
-// reports protection it does not have - strictly worse than security being off.
+// FALSE, and it must stay false until the BLOCKER below is fixed.
+//
+// Handler coverage IS complete: every RPC that reads or writes user data calls
+// gate()/gateFile(), verified by auditing every `DatabaseGrpcImpl::` handler
+// rather than working from a list. That audit is worth repeating after adding
+// any RPC - the first pass missed Upsert, all three Batch* handlers, all four
+// Set* handlers, and Subscribe, which streams document bodies and with an empty
+// `collections` list means every collection.
+//
+// BLOCKER - principal identity is not per-call.
+// -------------------------------------------------------------------------
+// `BearerAuthProcessor` stamps the principal into the gRPC AuthContext, and
+// `auth::principalOf()` reads it back. That is unsound: AuthContext properties
+// are per-CONNECTION, not per-call. gRPC pools channels by target, so several
+// clients using DIFFERENT tokens against the same address can share one
+// connection and all observe whichever principal was stamped first.
+//
+// Demonstrated by tests/load_test/test_policy_enforcement.sh: three clients
+// with three distinct named keys all resolved as `ops`, so every check passed
+// that should have been denied. Misattributing one caller's identity to another
+// is worse than having no identity at all, which is why the flag stays false.
 //
-// Gated so far: Get, Count.
-// Still to wire: Find, Exists, SimilaritySearch, GetVersionHistory,
-//   GetDocumentVersion, RestoreVersion, RestoreToDate, GetCollectionInfo,
-//   Insert, Update, PatchDocument, Delete, Subscribe/Watch,
-//   UploadFile, DownloadFile, DeleteFile, GetFileInfo, ListFiles.
+// The fix is to resolve the principal per call from the request metadata rather
+// than from the connection: stop consuming `authorization` in the processor and
+// map token -> name in a server interceptor (or in the gate itself) on every
+// request. That is the next task, and the e2e above is its reproducer.
 //
-// Also outstanding: masked-field application on returned documents at each
-// read site, refusing writes that touch a masked field, and the management
-// surface (dedicated RPCs + CLI) so policy edits refresh this cache and cannot
-// bypass the lockout guards by writing `_policies` directly.
+// Deliberately NOT gated, because they expose no user data:
+//   * HealthCheck        - liveness; load balancers need it unauthenticated.
+//   * GetReadOnlyStatus  - whether writes are being accepted, and why.
+//
+// Coarsely gated via requireAnyAdmin(), because a service-wide or cross-project
+// operation has no single project to authorise against: GetStats, SetReadOnly,
+// CreateProject, DropProject. ListProjects filters instead. The real boundary
+// for these is listener separation - do not expose an admin listener publicly.
 inline constexpr bool kEnforcementCoverageComplete = false;
 
 enum class SecurityMode { Enforce, Audit };
@@ -147,12 +169,28 @@ public:
 
     bool setPolicy(const std::string& project, const Policy& policy,
                    std::string& errorOut);
+    // Same, taking the on-the-wire JSON body, so policy management can go
+    // through the ordinary document API without duplicating the schema.
+    bool setPolicyFromJson(const std::string& project, const nlohmann::json& body,
+                           std::string& errorOut);
+    // mayAdminister() without the "unsecured means everyone" shortcut being
+    // recomputed by callers - same semantics, named to make the call sites read
+    // as an authorisation check.
+    [[nodiscard]] bool mayAdministerNow(const std::string& project,
+                                        const std::string& principal) const {
+        return mayAdminister(project, principal);
+    }
     bool removePolicy(const std::string& project, const std::string& principal,
                       std::string& errorOut);
     [[nodiscard]] std::optional<Policy> getPolicy(const std::string& project,
                                                   const std::string& principal) const;
     [[nodiscard]] std::vector<Policy> listPolicies(const std::string& project) const;
 
+    // True if `principal` is an admin of any project that has security enabled.
+    // The bar for service-wide operations, which have no single project to
+    // authorise against.
+    [[nodiscard]] bool isAdminSomewhere(const std::string& principal) const;
+
     // Test seam. Lets this component's own tests exercise ENFORCING behaviour
     // while kEnforcementCoverageComplete is still false. Nothing on the RPC
     // path calls it, so it cannot be used to arm a half-wired deployment.

tests/load_test/seed_policies.cpp

@@ -0,0 +1,72 @@
+// v2.8.0 — creates the policies used by test_policy_enforcement, through the
+// ordinary document API on `_policies`. This is also the real operator
+// workflow: write policies while the project is unsecured, then write the
+// `__security__` record to arm it.
+
+#include <smartbotic/database/client.hpp>
+#include <iostream>
+
+using namespace smartbotic::database;
+
+int main(int argc, char** argv) {
+    Client::Config cfg;
+    cfg.address = argc > 1 ? argv[1] : "127.0.0.1:9012";
+    cfg.auth_token = argc > 2 ? argv[2] : "";
+    // gRPC requires TLS for token auth; skip verification for the
+    // auto-generated self-signed dev cert.
+    cfg.tls_enabled = true;
+    cfg.tls_insecure_skip_verify = true;
+    cfg.project = "default";          // _policies is global; ids carry the project
+    Client db(cfg);
+    db.connect();
+
+    auto put = [&](const std::string& id, const nlohmann::json& body) {
+        db.upsert("_policies", body, id);  // note: (collection, data, id)
+        std::cout << "  seeded " << id << "\n";
+    };
+
+    // Admin of project "secured". Required before security can be enabled -
+    // the server refuses to arm a project with no admin.
+    {
+        nlohmann::json ops;
+        ops["admin"] = true;
+        put("secured:ops", ops);
+    }
+
+    // Reader: `docs` only, `ssn` masked, pinned to tenant == acme.
+    // Built field by field: nested nlohmann brace-init is ambiguous between
+    // object and array and silently produces the wrong shape.
+    {
+        nlohmann::json row = nlohmann::json::array();
+        nlohmann::json f = nlohmann::json::object();
+        f["field"] = "tenant";
+        f["op"] = 0;              // FilterOp::EQ
+        f["value"] = "acme";
+        row.push_back(f);
+
+        nlohmann::json rule = nlohmann::json::object();
+        rule["read"] = true;
+        rule["write"] = false;
+        rule["mask"] = nlohmann::json::array({"ssn"});
+        rule["row"] = row;
+
+        nlohmann::json colls = nlohmann::json::object();
+        colls["docs"] = rule;
+
+        nlohmann::json reader = nlohmann::json::object();
+        reader["admin"] = false;
+        reader["collections"] = colls;
+        put("secured:reader", reader);
+    }
+
+    // Arm it. Routed through setSecurity server-side so the no-admin guard runs.
+    {
+        nlohmann::json sec = nlohmann::json::object();
+        sec["enabled"] = true;
+        sec["mode"] = "enforce";
+        put("secured:__security__", sec);
+    }
+
+    std::cout << "seed complete\n";
+    return 0;
+}

tests/load_test/test_policy_enforcement.cpp

@@ -0,0 +1,130 @@
+// v2.8.0 — end-to-end policy enforcement.
+//
+// The engine has unit tests; this drives the real thing over gRPC with two
+// distinct NAMED api keys, which is the only way to prove that the principal
+// actually reaches the handlers and that every read/write path consults it.
+//
+// Shape: project "secured" is armed with an admin ("ops"), a reader granted
+// `docs` with `ssn` masked and a row predicate, and a third caller ("stranger")
+// with no policy at all. Project "open" is left unsecured to prove upgrades are
+// inert.
+
+#include <smartbotic/database/client.hpp>
+#include <iostream>
+#include <string>
+
+using namespace smartbotic::database;
+
+static int pass = 0, fail = 0;
+static void ck(bool c, const std::string& m) {
+    if (c) ++pass;
+    else { ++fail; std::cerr << "FAIL: " << m << "\n"; }
+}
+// A denial surfaces as a thrown runtime_error through the client.
+template <typename F>
+static bool denied(F f) {
+    try { f(); return false; } catch (const std::exception&) { return true; }
+}
+
+int main(int argc, char** argv) {
+    const std::string addr = argc > 1 ? argv[1] : "127.0.0.1:9012";
+    const std::string ops_key     = argc > 2 ? argv[2] : "";
+    const std::string reader_key  = argc > 3 ? argv[3] : "";
+    const std::string strange_key = argc > 4 ? argv[4] : "";
+
+    auto mk = [&](const std::string& key, const std::string& project) {
+        Client::Config c;
+        c.address = addr;
+        c.project = project;
+        c.auth_token = key;
+        // gRPC requires TLS for token auth; the harness uses the auto-generated
+        // self-signed cert, so verification is skipped.
+        c.tls_enabled = true;
+        c.tls_insecure_skip_verify = true;
+        return c;
+    };
+
+    // ---- Phase 1: unsecured project behaves exactly as before -------------
+    {
+        Client anyone(mk(strange_key, "open"));
+        anyone.connect();
+        auto id = anyone.insert("things", nlohmann::json{{"a", 1}});
+        ck(!id.empty(), "unsecured project: a principal with no policy can write");
+        ck(anyone.get("things", id).has_value(),
+           "unsecured project: and read back - upgrades are inert");
+    }
+
+    // ---- Phase 2: arm the secured project ---------------------------------
+    // Policies are written by the admin path; the harness script has already
+    // created them plus the __security__ record before this binary runs.
+    Client ops(mk(ops_key, "secured"));
+    Client reader(mk(reader_key, "secured"));
+    Client stranger(mk(strange_key, "secured"));
+    ops.connect(); reader.connect(); stranger.connect();
+
+    // Seed data as admin.
+    auto d1 = ops.insert("docs", nlohmann::json{
+        {"tenant", "acme"}, {"ssn", "111-22-3333"}, {"note", "visible"}});
+    auto d2 = ops.insert("docs", nlohmann::json{
+        {"tenant", "other"}, {"ssn", "999-88-7777"}, {"note", "hidden by row rule"}});
+    ck(!d1.empty() && !d2.empty(), "admin can write in a secured project");
+
+    // ---- Phase 3: deny-by-default -----------------------------------------
+    ck(denied([&]{ stranger.get("docs", d1); }),
+       "stranger with no policy is denied on read");
+    ck(denied([&]{ stranger.insert("docs", nlohmann::json{{"x", 1}}); }),
+       "stranger is denied on write");
+    ck(denied([&]{ stranger.find("docs", Client::QueryOptions{}); }),
+       "stranger is denied on find");
+    ck(denied([&]{ (void)stranger.count("docs"); }),
+       "stranger is denied on count - a count is information");
+
+    // ---- Phase 4: column mask --------------------------------------------
+    auto got = reader.get("docs", d1);
+    ck(got.has_value(), "reader may read a granted collection");
+    if (got) {
+        ck(!got->contains("ssn"), "masked column is absent from Get");
+        ck(got->contains("note"), "unmasked columns survive");
+    }
+    auto found = reader.find("docs", Client::QueryOptions{});
+    ck(!found.empty(), "reader may find");
+    for (const auto& d : found) {
+        ck(!d.contains("ssn"), "masked column is absent from Find results");
+    }
+
+    // ---- Phase 5: row predicate ------------------------------------------
+    // reader's rule pins tenant == acme, so d2 must be invisible entirely.
+    ck(!reader.get("docs", d2).has_value(),
+       "row predicate hides a document outright, reported as not-found");
+    bool saw_other = false;
+    for (const auto& d : found) {
+        if (d.value("tenant", "") == "other") saw_other = true;
+    }
+    ck(!saw_other, "row predicate filters Find results");
+
+    // ---- Phase 6: mask is not an inference channel ------------------------
+    ck(denied([&]{
+        reader.find("docs", Client::QueryOptions{
+            .filters = {Client::Filter::eq("ssn", "111-22-3333")}});
+       }),
+       "filtering on a masked column is refused - it would leak the value");
+
+    // ---- Phase 7: writes -------------------------------------------------
+    ck(denied([&]{ reader.insert("docs", nlohmann::json{{"tenant", "acme"}}); }),
+       "read grant does not imply write");
+    ck(denied([&]{ reader.get("secrets", "anything"); }),
+       "a collection with no rule is denied even for a known principal");
+
+    // ---- Phase 8: enumeration does not leak ------------------------------
+    auto cols = reader.listCollections();
+    bool leaked = false;
+    for (const auto& c : cols) if (c == "secrets") leaked = true;
+    ck(!leaked, "listCollections hides collections the caller cannot read");
+
+    // ---- Phase 9: system collections are admin-only ----------------------
+    ck(denied([&]{ (void)reader.get("_policies", "secured:reader"); }),
+       "a non-admin cannot read _policies - it describes the access model");
+
+    std::cout << "\npassed=" << pass << " failed=" << fail << "\n";
+    return fail == 0 ? 0 : 1;
+}

tests/load_test/test_policy_enforcement.sh

@@ -0,0 +1,89 @@
+#!/usr/bin/env bash
+# v2.8.0 — end-to-end policy enforcement over gRPC with distinct named keys.
+#
+# This is the test the unit suite cannot be: it proves the principal actually
+# reaches the handlers and that every read/write path consults it. Policies are
+# created through the ordinary document API on `_policies` while the project is
+# still unsecured, then the `__security__` record arms it - which is also the
+# real operator workflow.
+
+set -euo pipefail
+cd "$(dirname "$0")"
+
+ROOT=/data/smartbotic-database
+DIR=/tmp/sbdb-policy-e2e
+PORT=9012
+
+rm -rf "$DIR"
+mkdir -p "$DIR/data"
+
+CLI="$ROOT/build/cli/smartbotic-db-cli"
+OPS_KEY="$("$CLI" generate-auth-key)"
+READER_KEY="$("$CLI" generate-auth-key)"
+STRANGER_KEY="$("$CLI" generate-auth-key)"
+
+cat > "$DIR/config.json" <<EOF
+{
+  "storage": {
+    "data_directory": "$DIR/data",
+    "listeners": [
+      { "bind": "127.0.0.1", "port": $PORT,
+        "tls": { "enabled": true, "auto_self_signed_if_missing": true },
+        "auth": {
+          "required": true,
+          "keys": [
+            { "name": "ops",      "key": "$OPS_KEY" },
+            { "name": "reader",   "key": "$READER_KEY" },
+            { "name": "stranger", "key": "$STRANGER_KEY" }
+          ]
+        } }
+    ],
+    "encryption": { "enabled": false },
+    "migrations": { "enabled": false },
+    "replication": { "enabled": false }
+  }
+}
+EOF
+
+echo "=== building ==="
+g++ -std=c++20 -O1 -o "$DIR/seed" seed_policies.cpp \
+    -I"$ROOT/client/include" -I"$ROOT/build/client" \
+    -L"$ROOT/build/client" -lsmartbotic-db-client -lspdlog -lfmt
+g++ -std=c++20 -O1 -o "$DIR/enforce" test_policy_enforcement.cpp \
+    -I"$ROOT/client/include" -I"$ROOT/build/client" \
+    -L"$ROOT/build/client" -lsmartbotic-db-client -lspdlog -lfmt
+
+echo "=== booting server on $PORT ==="
+"$ROOT/build/service/smartbotic-database" --config "$DIR/config.json" \
+    > "$DIR/server.log" 2>&1 &
+SERVER_PID=$!
+cleanup() { kill "$SERVER_PID" 2>/dev/null || true; wait "$SERVER_PID" 2>/dev/null || true; }
+trap cleanup EXIT
+
+for _ in $(seq 1 60); do
+    grep -q "Notified systemd: READY" "$DIR/server.log" && break
+    sleep 0.5
+done
+grep -q "Notified systemd: READY" "$DIR/server.log" || {
+    echo "server failed to start:"; tail -30 "$DIR/server.log"; exit 1; }
+
+echo "=== seeding policies (project still unsecured) ==="
+export LD_LIBRARY_PATH="$ROOT/build/client"
+"$DIR/seed" "127.0.0.1:$PORT" "$OPS_KEY"
+
+echo "=== running enforcement assertions ==="
+set +e
+"$DIR/enforce" "127.0.0.1:$PORT" "$OPS_KEY" "$READER_KEY" "$STRANGER_KEY"
+RC=$?
+set -e
+
+if [[ $RC -ne 0 ]]; then
+    echo ""
+    echo "FAILED — server log tail:"; tail -40 "$DIR/server.log"; exit $RC
+fi
+
+echo ""
+echo "=== policy denials recorded in the operator log ==="
+grep -c "policy: DENIED" "$DIR/server.log" || true
+
+echo "OK — policy enforcement e2e passed"