fix(security): per-call principal identity; project-scope subscribe; surface denials

Enforcement now works end to end. kEnforcementCoverageComplete is true and
security can be enabled.

1. Identity is resolved PER CALL (the blocker).
   BearerAuthProcessor stamped the principal into the gRPC AuthContext and
   handlers read it back. AuthContext properties belong to the CONNECTION, and
   gRPC pools channels by target, so three clients with three distinct named
   keys all resolved as `ops` and every check that should have denied passed.
   Misattributing one caller's identity to another is worse than having none.

   PrincipalResolver now maps token -> name from the request's own
   `authorization` metadata on every call, using the same constant-time compare
   as the processor. The processor no longer consumes that header - handlers need
   to see it - and no longer touches the AuthContext. The service hands the impl
   the union of every listener's keys; a token is only accepted if some
   listener's processor accepted it, so the union cannot widen authentication.

2. subscribe() was the one client call left unqualified.
   Two cross-project consequences: a bare name matched nothing, because events
   carry the qualified collection, so subscriptions silently never fired; and an
   EMPTY list means "every collection" server-side, which meant every collection
   in every project. Named collections are qualified now, and "everything" sends
   a `<project>:*` pattern so all means all of MY project - no proto change, the
   pattern field already existed.

   Audited every name-carrying field in the client to find others: only
   subscribe was wrong. createProject/dropProject take project names and
   uploadFile/listFiles take filenames, all correctly bare.

3. Read paths surface denials instead of swallowing them.
   get/exists/find/findWithMetrics/count logged the error and returned
   empty/nullopt, so PERMISSION_DENIED was indistinguishable from "no data" and
   an application would silently take the wrong branch. They throw now - the
   same reasoning as the v2.4.2 change that stopped Find returning empty for a
   missing collection. The masked-filter refusal became PERMISSION_DENIED rather
   than INVALID_ARGUMENT for the same reason; it confirms the field is masked,
   but the caller already knows that since the field is absent from every
   document it reads.

Tests: tests/load_test/test_policy_enforcement.sh 19/19,
test_client_namespacing.sh 31/31 (subscribe scoping added), ctest 17/17, views
e2e and TLS/auth 4/4.

client/src/client.cpp

@@ -270,6 +270,15 @@ public:
 
         auto status = stub_->Get(&context, request, &response);
         if (!status.ok()) {
+            // v2.8.0 — a permission denial must NOT look like "no data".
+            // Returning empty here would leave an application unable to tell
+            // "you may not see this" from "there is nothing to see", so it
+            // would silently take the wrong branch. Same reasoning as the
+            // v2.4.2 change that stopped Find returning empty for a missing
+            // collection.
+            if (status.error_code() == grpc::StatusCode::PERMISSION_DENIED) {
+                throw std::runtime_error("access denied: get");
+            }
             spdlog::error("Client::get failed: {}", status.error_message());
             return std::nullopt;
         }
@@ -500,6 +509,15 @@ public:
 
         auto status = stub_->Exists(&context, request, &response);
         if (!status.ok()) {
+            // v2.8.0 — a permission denial must NOT look like "no data".
+            // Returning empty here would leave an application unable to tell
+            // "you may not see this" from "there is nothing to see", so it
+            // would silently take the wrong branch. Same reasoning as the
+            // v2.4.2 change that stopped Find returning empty for a missing
+            // collection.
+            if (status.error_code() == grpc::StatusCode::PERMISSION_DENIED) {
+                throw std::runtime_error("access denied: exists");
+            }
             spdlog::error("Client::exists failed: {}", status.error_message());
             return false;
         }
@@ -546,6 +564,15 @@ public:
             throw std::runtime_error(status.error_message());
         }
         if (!status.ok()) {
+            // v2.8.0 — a permission denial must NOT look like "no data".
+            // Returning empty here would leave an application unable to tell
+            // "you may not see this" from "there is nothing to see", so it
+            // would silently take the wrong branch. Same reasoning as the
+            // v2.4.2 change that stopped Find returning empty for a missing
+            // collection.
+            if (status.error_code() == grpc::StatusCode::PERMISSION_DENIED) {
+                throw std::runtime_error("access denied: find");
+            }
             spdlog::error("Client::find failed: {}", status.error_message());
             return {};
         }
@@ -599,6 +626,15 @@ public:
             throw std::runtime_error(status.error_message());  // see find()
         }
         if (!status.ok()) {
+            // v2.8.0 — a permission denial must NOT look like "no data".
+            // Returning empty here would leave an application unable to tell
+            // "you may not see this" from "there is nothing to see", so it
+            // would silently take the wrong branch. Same reasoning as the
+            // v2.4.2 change that stopped Find returning empty for a missing
+            // collection.
+            if (status.error_code() == grpc::StatusCode::PERMISSION_DENIED) {
+                throw std::runtime_error("access denied: findWithMetrics");
+            }
             spdlog::error("Client::findWithMetrics failed: {}", status.error_message());
             return {};
         }
@@ -645,6 +681,15 @@ public:
 
         auto status = stub_->Count(&context, request, &response);
         if (!status.ok()) {
+            // v2.8.0 — a permission denial must NOT look like "no data".
+            // Returning empty here would leave an application unable to tell
+            // "you may not see this" from "there is nothing to see", so it
+            // would silently take the wrong branch. Same reasoning as the
+            // v2.4.2 change that stopped Find returning empty for a missing
+            // collection.
+            if (status.error_code() == grpc::StatusCode::PERMISSION_DENIED) {
+                throw std::runtime_error("access denied: count");
+            }
             spdlog::error("Client::count failed: {}", status.error_message());
             return 0;
         }
@@ -1222,8 +1267,23 @@ public:
         attachAuth(*context);
 
         smartbotic::databasepb::SubscribeRequest request;
+        // v2.8.0 — subscribe was the one call left unqualified while every
+        // document call qualified. Two consequences, both cross-project leaks:
+        //
+        //   * a bare name matched nothing, because events carry the qualified
+        //     collection - so subscriptions silently never fired;
+        //   * an EMPTY list means "every collection" server-side, which meant
+        //     every collection in every PROJECT.
+        //
+        // Named collections are qualified like any other. For the "everything"
+        // case we send a `<project>:*` pattern instead of an empty request, so
+        // "all" means all of MY project. That needs no proto change - the
+        // pattern field already existed.
         for (const auto& coll : collections) {
-            request.add_collections(coll);
+            request.add_collections(qualify(coll));
+        }
+        if (collections.empty()) {
+            request.add_patterns(config_.project + ":*");
         }
         request.set_include_data(true);
 

service/src/auth/auth_interceptor.cpp

@@ -54,23 +54,14 @@ grpc::Status BearerAuthProcessor::Process(
 
     for (const auto& k : keys_) {
         if (constant_time_eq(tok, k.key)) {
-            // Mark the metadata as consumed so gRPC doesn't surface it
-            // to handler code.
-            consumed_auth_metadata->insert(
-                std::make_pair(std::string("authorization"),
-                                std::string(value)));
-
-            // v2.7.0 — stamp the principal. This is the ONLY channel from an
-            // AuthMetadataProcessor to a handler, so policy enforcement
-            // downstream depends on it. Empty name should be impossible
-            // (config normalises bare keys to `unnamed`) but fall back rather
-            // than stamping an empty identity.
-            if (context) {
-                const std::string principal =
-                    k.name.empty() ? std::string(kPrincipalUnnamed) : k.name;
-                context->AddProperty(kPrincipalProperty, principal);
-                context->SetPeerIdentityPropertyName(kPrincipalProperty);
-            }
+            // v2.7.0 — deliberately NOT consumed. Handlers must be able to read
+            // `authorization` so identity can be resolved PER CALL; see
+            // auth/principal.hpp. Stamping the name into the AuthContext here
+            // instead was unsound: AuthContext properties belong to the
+            // connection, and pooled channels made several distinct callers
+            // resolve as whichever principal was stamped first.
+            (void)consumed_auth_metadata;
+            (void)context;
             return grpc::Status::OK;
         }
     }

service/src/auth/principal.cpp

@@ -1,25 +1,54 @@
-// v2.7.0 — caller identity. See principal.hpp.
+// v2.7.0 — per-call caller identity. See principal.hpp for why this does not
+// use the gRPC AuthContext.
 
 #include "auth/principal.hpp"
 
 #include <grpcpp/grpcpp.h>
 
-namespace smartbotic::database::auth {
-
-std::string principalOf(const grpc::ServerContext* context) {
-    if (!context) return kPrincipalAnonymous;
+#include <cstring>
 
-    auto auth_context = context->auth_context();
-    if (!auth_context) return kPrincipalAnonymous;
+namespace smartbotic::database::auth {
 
-    // Listeners without auth register no processor, so the property is absent.
-    // That is the `anonymous` case, not an error.
-    auto values = auth_context->FindPropertyValues(kPrincipalProperty);
-    if (values.empty()) return kPrincipalAnonymous;
+namespace {
+
+// Length-independent compare, mirroring BearerAuthProcessor. The token is a
+// secret; a short-circuiting compare leaks how much of it matched.
+bool constant_time_eq(std::string_view a, std::string_view b) {
+    if (a.size() != b.size()) return false;
+    unsigned char diff = 0;
+    for (size_t i = 0; i < a.size(); ++i) {
+        diff |= static_cast<unsigned char>(a[i]) ^ static_cast<unsigned char>(b[i]);
+    }
+    return diff == 0;
+}
 
-    std::string name(values[0].data(), values[0].size());
-    if (name.empty()) return kPrincipalAnonymous;
-    return name;
+constexpr std::string_view kPrefix = "Bearer ";
+
+}  // namespace
+
+std::string PrincipalResolver::resolve(const grpc::ServerContext* context) const {
+    if (!context || keys_.empty()) return kPrincipalAnonymous;
+
+    const auto& md = context->client_metadata();
+    auto it = md.find("authorization");
+    if (it == md.end()) return kPrincipalAnonymous;
+
+    std::string_view value(it->second.data(), it->second.size());
+    if (value.size() < kPrefix.size() ||
+        std::memcmp(value.data(), kPrefix.data(), kPrefix.size()) != 0) {
+        return kPrincipalAnonymous;
+    }
+    const std::string_view token = value.substr(kPrefix.size());
+
+    for (const auto& k : keys_) {
+        if (constant_time_eq(token, k.key)) {
+            return k.name.empty() ? std::string(kPrincipalUnnamed) : k.name;
+        }
+    }
+    // The token authenticated at the listener but is not in our table. Treat as
+    // anonymous rather than trusting it: an unrecognised token must not become a
+    // privileged identity.
+    return kPrincipalAnonymous;
 }
 
 }  // namespace smartbotic::database::auth

service/src/auth/principal.hpp

@@ -1,30 +1,40 @@
-// v2.7.0 — caller identity.
-//
-// Until now the server knew only "the token was valid". It attached nothing to
-// the call, so every key holder was indistinguishable and access control was
-// impossible to express. This is the identity layer that per-project row- and
-// column-level policy is written against.
+// v2.7.0 — caller identity, resolved PER CALL.
 //
 // How a principal is decided, in order:
-//   1. the listener has auth and the token matched a NAMED key  -> that name
-//   2. the listener has auth and the token matched a BARE key    -> `unnamed`
-//   3. the listener has no auth processor at all                 -> `anonymous`
+//   1. the request carries a bearer token matching a NAMED key -> that name
+//   2. the request carries a token matching a BARE key          -> `unnamed`
+//   3. the request carries no usable token                     -> `anonymous`
 //
-// Case 3 is not an error. The default 127.0.0.1 plaintext listener has no auth,
-// and `smartbotic-db-cli` plus operator tooling ride on it. `anonymous` is a
-// real, grantable principal so local access is allowed by an explicit policy
+// Case 3 is not an error. The default 127.0.0.1 plaintext listener has no auth
+// and `smartbotic-db-cli` plus operator tooling ride on it, so `anonymous` is a
+// real, grantable principal: local access is allowed by an explicit policy
 // rather than by an implicit hole. See
 // docs/superpowers/specs/2026-08-08-rls-cls-design.md.
 //
-// Transport: BearerAuthProcessor stores the name as an auth-context property.
-// gRPC gives no way to pass state from an AuthMetadataProcessor to a handler
-// other than the AuthContext, which is why the processor's `context` argument
-// (ignored since v2.4) is now used.
+// WHY NOT AuthContext
+// -------------------
+// The obvious implementation - have BearerAuthProcessor stamp the name into the
+// gRPC AuthContext and read it back in the handler - is UNSOUND, and was
+// actually shipped and caught by
+// tests/load_test/test_policy_enforcement.sh before it could be enabled.
+//
+// AuthContext properties belong to the CONNECTION, not the call. gRPC pools
+// channels by target address, so several clients using DIFFERENT tokens against
+// the same address can share one connection and every one of them observes
+// whichever principal was stamped first. In the reproducer, three clients with
+// three distinct named keys all resolved as `ops`, so every check that should
+// have denied passed instead.
+//
+// Misattributing one caller's identity to another is worse than having no
+// identity at all. So identity is resolved from the request's own
+// `authorization` metadata on every call, and the auth processor no longer
+// consumes that header - handlers need to see it.
 
 #pragma once
 
 #include <string>
 #include <string_view>
+#include <vector>
 
 namespace grpc {
 class ServerContext;
@@ -32,12 +42,9 @@ class ServerContext;
 
 namespace smartbotic::database::auth {
 
-// Auth-context property carrying the principal name.
-inline constexpr const char* kPrincipalProperty = "sbdb_principal";
-
 // Reserved principal names. Rejected as key names in config, because a key
-// called "anonymous" would otherwise be indistinguishable from an unauthed
-// caller in a policy.
+// called "anonymous" would be indistinguishable from an unauthenticated caller
+// in a policy.
 inline constexpr const char* kPrincipalAnonymous = "anonymous";
 inline constexpr const char* kPrincipalUnnamed   = "unnamed";
 
@@ -45,8 +52,32 @@ inline bool isReservedPrincipal(std::string_view name) {
     return name == kPrincipalAnonymous || name == kPrincipalUnnamed;
 }
 
-// Resolve the calling principal. Never throws and never returns empty: a
-// request with no identity is `anonymous`, which is a decision, not a failure.
-std::string principalOf(const grpc::ServerContext* context);
+// Maps a bearer token to the principal that owns it.
+//
+// Holds the union of every listener's keys, because one DatabaseGrpcImpl is
+// registered on all listeners and a request does not carry which listener it
+// arrived on. A token is only accepted at all if some listener's auth processor
+// accepted it, so the union cannot widen authentication - it only names what
+// was already authenticated.
+class PrincipalResolver {
+public:
+    struct NamedKey {
+        std::string name;
+        std::string key;
+    };
+
+    PrincipalResolver() = default;
+    explicit PrincipalResolver(std::vector<NamedKey> keys) : keys_(std::move(keys)) {}
+
+    void setKeys(std::vector<NamedKey> keys) { keys_ = std::move(keys); }
+    [[nodiscard]] bool empty() const noexcept { return keys_.empty(); }
+
+    // Never throws, never returns empty: no identity means `anonymous`, which is
+    // a decision rather than a failure.
+    [[nodiscard]] std::string resolve(const grpc::ServerContext* context) const;
+
+private:
+    std::vector<NamedKey> keys_;
+};
 
 }  // namespace smartbotic::database::auth

service/src/database_grpc_impl.cpp

@@ -127,7 +127,7 @@ grpc::Status DatabaseGrpcImpl::gate(const grpc::ServerContext* context,
                             "access denied");
     }
 
-    const std::string principal = smartbotic::database::auth::principalOf(context);
+    const std::string principal = principals_.resolve(context);
     out = policy_manager_.authorize(pc.project, principal, pc.collection, access);
     if (!out.allowed) {
         // Deliberately terse and identical for every denial reason. A response
@@ -151,7 +151,7 @@ grpc::Status DatabaseGrpcImpl::gateFile(const grpc::ServerContext* context,
         return grpc::Status::OK;
     }
     const std::string proj = project.empty() ? "default" : project;
-    const std::string principal = smartbotic::database::auth::principalOf(context);
+    const std::string principal = principals_.resolve(context);
     out = policy_manager_.authorizeFile(proj, principal, fileType, access);
     if (!out.allowed) {
         spdlog::warn("policy: DENIED principal '{}' {} on file type '{}' in '{}' ({})",
@@ -187,7 +187,7 @@ bool DatabaseGrpcImpl::handlePolicyWrite(const grpc::ServerContext* context,
     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);
+    const std::string principal = principals_.resolve(context);
     if (!policy_manager_.mayAdministerNow(project, principal)) {
         spdlog::warn("policy: DENIED principal '{}' management of project '{}'",
                      principal, project);
@@ -232,7 +232,7 @@ bool DatabaseGrpcImpl::handlePolicyWrite(const grpc::ServerContext* context,
 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);
+    const std::string principal = principals_.resolve(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;
@@ -286,8 +286,13 @@ grpc::Status DatabaseGrpcImpl::rejectMaskedFilters(
         for (const auto& m : mask) {
             // Exact path, or the filter reaching into a masked subtree.
             if (f.field == m || f.field.rfind(m + ".", 0) == 0) {
+                // PERMISSION_DENIED, not INVALID_ARGUMENT: this is an
+                // authorisation outcome, and clients surface denials rather
+                // than swallowing them into an empty result. It confirms the
+                // field is masked, but the caller already knows that - the
+                // field is absent from every document it reads.
                 return grpc::Status(
-                    grpc::StatusCode::INVALID_ARGUMENT,
+                    grpc::StatusCode::PERMISSION_DENIED,
                     "filtering on field '" + f.field + "' is not permitted");
             }
         }
@@ -1664,7 +1669,7 @@ grpc::Status DatabaseGrpcImpl::ListProjects(
     // 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();
+        secured ? principals_.resolve(context) : std::string();
     for (auto& name : names) {
         if (secured) {
             const auto sec = policy_manager_.securityOf(name);

service/src/database_grpc_impl.hpp

@@ -47,6 +47,13 @@ public:
 
     ~DatabaseGrpcImpl() override = default;
 
+    // Set the token -> principal table. Called once at startup, before the
+    // server accepts requests.
+    void setPrincipalKeys(
+        std::vector<smartbotic::database::auth::PrincipalResolver::NamedKey> keys) {
+        principals_.setKeys(std::move(keys));
+    }
+
     // ===== Document Operations =====
 
     grpc::Status Insert(
@@ -375,6 +382,10 @@ private:
     CollectionConfigManager& config_manager_;
     PolicyManager& policy_manager_;
 
+    // v2.7.0 — token -> principal, resolved per call. Populated by
+    // DatabaseService from the union of all listeners' keys.
+    smartbotic::database::auth::PrincipalResolver principals_;
+
     // v2.8.0 — the single access gate every handler goes through.
     //
     // `qualified` is the "<project>:<collection>" form the handler already has.

service/src/database_service.cpp

@@ -1095,6 +1095,18 @@ void DatabaseService::setupComponents() {
         *this, *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_,
         *policy_manager_
     );
+    // v2.7.0 — hand the impl the union of every listener's keys so it can
+    // resolve a principal per call. A token is only accepted if some listener's
+    // auth processor accepted it, so the union cannot widen authentication; it
+    // only names what was already authenticated.
+    {
+        std::vector<smartbotic::database::auth::PrincipalResolver::NamedKey> all;
+        for (const auto& l : config_.listeners) {
+            for (const auto& k : l.auth.keys) all.push_back({k.name, k.key});
+        }
+        storageImpl_->setPrincipalKeys(std::move(all));
+    }
+
     // v1.6.2 — wire the streaming-RPC concurrency limits from GrpcConfig.
     storageImpl_->setStreamLimits(
         config_.grpc.maxConcurrentSubscribeStreams,

service/src/security/policy_manager.hpp

@@ -109,7 +109,7 @@ struct Policy {
 // 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;
+inline constexpr bool kEnforcementCoverageComplete = true;
 
 enum class SecurityMode { Enforce, Audit };
 

tests/load_test/README.md

@@ -485,3 +485,32 @@ Covers:
 - `test_v24_tls_auth.sh` — boots plaintext + TLS/auth listeners and drives 4
   scenarios: plaintext-local OK, TLS+token OK, TLS+wrong-token
   `UNAUTHENTICATED`, TLS+no-token `UNAUTHENTICATED`.
+
+---
+
+## Test 10 — Policy enforcement (`test_policy_enforcement.sh`)
+
+    ./test_policy_enforcement.sh
+
+19 assertions over gRPC with **three distinct named API keys**, on a TLS
+listener (gRPC aborts if an auth processor is attached to insecure credentials).
+Boots its own server on port 9012 and tears it down.
+
+This is the test that caught the per-connection identity bug. The engine's unit
+tests (`test_policy_manager`, 54 assertions) cannot: they call `authorize()`
+directly, so they never exercise how the principal reaches a handler. Driving
+three real clients does, and the first run had all three resolve as the same
+principal because `AuthContext` properties belong to the connection, not the
+call.
+
+Covers: an unsecured project stays fully open (upgrades are inert);
+deny-by-default for a principal with no policy on read, write, find and count; a
+column mask absent from `Get` and `Find`; a row predicate hiding a document
+outright and filtering `Find`; a filter on a masked column refused (otherwise
+the mask is an inference channel); read not implying write; a collection with no
+rule denied; `listCollections` hiding what the caller cannot read; `_policies`
+unreadable by a non-admin.
+
+Policies are seeded 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.

tests/load_test/test_client_namespacing.cpp

@@ -16,7 +16,10 @@
 // is reversible, and a precision-only configure must not clobber it.
 
 #include <smartbotic/database/client.hpp>
+#include <atomic>
+#include <chrono>
 #include <iostream>
+#include <thread>
 using namespace smartbotic::database;
 static int pass=0, fail=0;
 static void ck(bool c, const char* m){ if(c){++pass;} else {++fail; std::cerr<<"FAIL: "<<m<<"\n";} }
@@ -118,6 +121,34 @@ int main() {
        "identical bytes from another project must NOT report deduplicated");
     ck(c.deleteFile(up.id), "owner can delete its own file");
 
+    // ---- Part 4 (v2.8.0): subscribe is project-scoped like everything else.
+    // Before this, a bare collection name never matched (events carry the
+    // qualified name) and an empty list meant every collection in every project.
+    {
+        Client peer({.address="127.0.0.1:9011", .project="proj9"});
+        peer.connect();
+
+        std::atomic<int> mine{0}, theirs{0};
+        auto sub_mine = c.subscribe({"subs"}, [&](const std::string&, const std::string&,
+                                                  const std::string&,
+                                                  const std::optional<nlohmann::json>&) {
+            ++mine;
+        });
+        auto sub_all = peer.subscribe({}, [&](const std::string&, const std::string&,
+                                              const std::string&,
+                                              const std::optional<nlohmann::json>&) {
+            ++theirs;   // "everything" must mean everything in proj9, not proj1
+        });
+        std::this_thread::sleep_for(std::chrono::milliseconds(300));
+
+        c.insert("subs", nlohmann::json{{"n", 1}});
+        std::this_thread::sleep_for(std::chrono::milliseconds(600));
+
+        ck(mine.load() > 0, "a named subscription fires for the caller's own project");
+        ck(theirs.load() == 0,
+           "an empty subscription does NOT receive another project's events");
+    }
+
     std::cout << "\npassed=" << pass << " failed=" << fail << "\n";
     return fail==0?0:1;
 }