merge: v2.7.0 row- and column-level security

Brings in principal identity, the policy engine, enforcement across every
data-path handler, the policy CLI, and the fixes found along the way:
per-call identity (AuthContext is per-connection), subscribe project scoping,
client reads surfacing PERMISSION_DENIED, system-collection qualification, and
the auth-without-TLS startup abort.

VERSION

@@ -1 +1 @@
-2.6.0
+2.7.0

cli/main.cpp

@@ -70,6 +70,13 @@ void printUsage() {
               << "  " << C_CYAN << "lock" << C_RESET << "                                 Lock database (read-only)\n"
               << "  " << C_CYAN << "unlock" << C_RESET << "                               Unlock database (accept writes)\n"
               << "  " << C_CYAN << "status" << C_RESET << "                               Show read-only + recovery status\n"
+              << C_BOLD << "  Access policy (v2.7.0+)" << C_RESET << "\n"
+              << "  " << C_CYAN << "security" << C_RESET << " [project]                   Show whether a project enforces policy\n"
+              << "  " << C_CYAN << "security-set" << C_RESET << " <project> <on|off> [enforce|audit]\n"
+              << "  " << C_CYAN << "policies" << C_RESET << " [project]                   List principals with a policy\n"
+              << "  " << C_CYAN << "policy" << C_RESET << " <project> <principal>       Show one policy\n"
+              << "  " << C_CYAN << "policy-set" << C_RESET << " <project> <principal> '<json>'\n"
+              << "  " << C_CYAN << "policy-rm" << C_RESET << " <project> <principal>\n"
               << "  " << C_CYAN << "help" << C_RESET << "                                 Show this help\n\n"
               << C_BOLD << "Offline helpers" << C_RESET << " " << C_DIM << "(no server connection required)" << C_RESET << ":\n"
               << "  " << C_CYAN << "generate-auth-key" << C_RESET << "                    Emit a base64 32-byte API key\n"
@@ -543,6 +550,129 @@ bool execCommand(smartbotic::database::Client& client,
             return true;
         }
 
+
+        // ===== v2.7.0 access policy =====
+        //
+        // Policy lives in the `_policies` collection and is managed through the
+        // ordinary document API, which the server intercepts so edits refresh
+        // its cache and the `__security__` record goes through the lockout
+        // guards. These commands are ergonomics over that, not a second
+        // mechanism - which is why there is no policy RPC to keep in step.
+        if (cmd == "policies") {
+            const std::string project = params.empty() ? "default" : params[0];
+            auto rows = client.find("_policies", smartbotic::database::Client::QueryOptions{.limit = 1000});
+            std::cout << C_BOLD << "policies in project '" << project << "'" << C_RESET << "\n";
+            size_t shown = 0;
+            for (const auto& r : rows) {
+                const std::string id = r.value("_id", "");
+                if (id.rfind(project + ":", 0) != 0) continue;
+                const std::string tail = id.substr(project.size() + 1);
+                if (tail == "__security__") continue;
+                std::cout << "  " << C_CYAN << tail << C_RESET
+                          << (r.value("admin", false) ? "  (admin)" : "") << "\n";
+                ++shown;
+            }
+            if (shown == 0) std::cout << C_DIM << "  (none)" << C_RESET << "\n";
+            return true;
+        }
+
+        if (cmd == "policy") {
+            if (params.size() < 2) {
+                printError("usage: policy <project> <principal>");
+                return false;
+            }
+            auto doc = client.get("_policies", params[0] + ":" + params[1]);
+            if (!doc) { printError("no policy for " + params[0] + ":" + params[1]); return false; }
+            printJson(*doc);
+            return true;
+        }
+
+        if (cmd == "policy-set") {
+            if (params.size() < 3) {
+                printError("usage: policy-set <project> <principal> '<json>'\n"
+                           "  e.g. policy-set acme svc "
+                           "'{\"collections\":{\"users\":{\"read\":true,\"mask\":[\"ssn\"]}}}'\n"
+                           "  admin:  policy-set acme ops '{\"admin\":true}'");
+                return false;
+            }
+            try {
+                auto body = json::parse(params[2]);
+                client.upsert("_policies", body, params[0] + ":" + params[1]);
+                std::cout << C_GREEN << "ok" << C_RESET << " policy set for "
+                          << params[0] << ":" << params[1] << "\n";
+                return true;
+            } catch (const std::exception& e) {
+                printError(e.what());
+                return false;
+            }
+        }
+
+        if (cmd == "policy-rm") {
+            if (params.size() < 2) { printError("usage: policy-rm <project> <principal>"); return false; }
+            try {
+                bool ok = client.remove("_policies", params[0] + ":" + params[1]);
+                std::cout << (ok ? "removed\n" : "not found\n");
+                return ok;
+            } catch (const std::exception& e) {
+                // The server refuses to remove the last admin of a secured
+                // project - that refusal is the lockout guard, not an error to
+                // work around.
+                printError(e.what());
+                return false;
+            }
+        }
+
+        if (cmd == "security") {
+            const std::string project = params.empty() ? "default" : params[0];
+            auto doc = client.get("_policies", project + ":__security__");
+            if (!doc) {
+                std::cout << "project '" << project << "': security "
+                          << C_GREEN << "disabled" << C_RESET
+                          << C_DIM << " (default - all access allowed)" << C_RESET << "\n";
+                return true;
+            }
+            const bool on = doc->value("enabled", false);
+            const std::string mode = doc->value("mode", "enforce");
+            std::cout << "project '" << project << "': security "
+                      << (on ? (mode == "audit" ? C_YELLOW : C_RED) : C_GREEN)
+                      << (on ? (mode == "audit" ? "AUDIT" : "ENFORCED") : "disabled")
+                      << C_RESET << "\n";
+            if (on && mode == "audit") {
+                std::cout << C_DIM << "  audit mode logs what it would deny and "
+                             "allows the request - watch the service log, then "
+                             "switch to enforce." << C_RESET << "\n";
+            }
+            return true;
+        }
+
+        if (cmd == "security-set") {
+            if (params.size() < 2) {
+                printError("usage: security-set <project> <on|off> [enforce|audit]\n"
+                           "  Enabling is REFUSED unless some policy in the project has "
+                           "admin=true.\n"
+                           "  Switch a live project on with 'audit' first.");
+                return false;
+            }
+            const bool on = params[1] == "on" || params[1] == "true";
+            const std::string mode = params.size() > 2 ? params[2] : "enforce";
+            if (mode != "enforce" && mode != "audit") {
+                printError("mode must be 'enforce' or 'audit'");
+                return false;
+            }
+            try {
+                json body;
+                body["enabled"] = on;
+                body["mode"] = mode;
+                client.upsert("_policies", body, params[0] + ":__security__");
+                std::cout << C_GREEN << "ok" << C_RESET << " project '" << params[0]
+                          << "' security " << (on ? mode : "disabled") << "\n";
+                return true;
+            } catch (const std::exception& e) {
+                printError(e.what());
+                return false;
+            }
+        }
+
         if (cmd == "lock") {
             if (client.setReadOnly(true)) {
                 std::cout << C_GREEN << "ok" << C_RESET << " Database locked (read-only)\n";

client/src/client.cpp

@@ -72,6 +72,17 @@ public:
         if (collection.find(':') != std::string_view::npos) {
             return std::string(collection);
         }
+        // v2.7.0 — system collections (`_views`, `_collection_meta`,
+        // `_policies`, `_files`) are GLOBAL, not per-project: their rows are
+        // already keyed by a project-qualified id. Qualifying the collection
+        // name too produced `default:_policies`, which does not exist, so
+        // reads silently found nothing while writes succeeded (the server
+        // tolerates both spellings on the policy write path). That asymmetry
+        // is exactly how a management command can look like it worked and
+        // then show stale state.
+        if (!collection.empty() && collection.front() == '_') {
+            return std::string(collection);
+        }
         std::string out;
         out.reserve(config_.project.size() + 1 + collection.size());
         out.append(config_.project);
@@ -270,6 +281,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 +520,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 +575,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 +637,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 +692,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 +1278,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);
 

docs/integration-guide.md

@@ -15,6 +15,7 @@ How to install, configure, and integrate smartbotic-database into your C++ proje
 - [Upgrading from Legacy Packages](#upgrading-from-legacy-packages)
 - [Drop-in configuration (conf.d)](#drop-in-configuration-confd)
 - [Eviction & Memory Pressure](#eviction--memory-pressure)
+- [Access Policy — row and column level security](#access-policy--row-and-column-level-security-270)
 
 ---
 
@@ -394,6 +395,106 @@ this name is always a subset of what the view allows.
 > reader; the protection is against someone reading data files, snapshots, or
 > backups, not against a client.
 
+### Access Policy — row and column level security (2.7.0+)
+
+Per-project row- and column-level access control. **Off by default**: a project
+with no policy behaves exactly as it did before 2.7.0, so upgrading changes
+nothing until you turn it on.
+
+> Before 2.7.0 there was no access control beyond authentication. Every API-key
+> holder had identical, complete access. Views were never a boundary - see the
+> note under "Why AND-merge filters".
+
+#### The model
+
+- **Principal** = the *name* of the API key the request arrived with. Keys became
+  named in 2.7.0:
+  ```json
+  "auth": { "required": true, "keys": [
+      { "name": "shadowman", "key": "<base64>" },
+      { "name": "callerai",  "key": "<base64>" } ] }
+  ```
+  A bare string key still works and maps to the reserved principal `unnamed`. A
+  request with no usable token is the reserved principal `anonymous` - which is
+  grantable, so the local CLI keeps working by explicit policy rather than by an
+  implicit hole. `anonymous` and `unnamed` are refused as key names.
+- **Policy** = one record per (project, principal) granting `read`/`write` per
+  collection or file type, with an optional column `mask` and row predicate.
+- **admin** grants everything within a project plus the right to edit policy.
+
+#### Turning it on safely
+
+Enabling is **refused** unless some policy in the project has `admin: true` -
+lockout is prevented structurally, not by care. Removing the last admin of a
+secured project is refused for the same reason.
+
+Always arm a live project in **audit** mode first. Audit evaluates every policy,
+logs what it *would* deny, and allows the request:
+
+```bash
+smartbotic-db-cli policy-set acme ops '{"admin":true}'
+smartbotic-db-cli policy-set acme svc \
+  '{"collections":{"users":{"read":true,"mask":["ssn"],
+     "row":[{"field":"tenant","op":0,"value":"acme"}]}}}'
+
+smartbotic-db-cli security-set acme on audit     # watch the log
+smartbotic-db-cli security-set acme on enforce   # then commit
+smartbotic-db-cli security acme                  # show current state
+smartbotic-db-cli policies acme                  # list principals
+```
+
+Once enforcing, the project is **deny-by-default**: a principal with no matching
+rule gets nothing.
+
+#### What enforcement covers
+
+Every RPC that reads or writes user data. Notably including the ones that are
+easy to forget:
+
+- **`Subscribe`** is filtered per event, since an empty collection list means
+  "everything" and there is no single name to authorise up front. Masked columns
+  are stripped from event payloads.
+- **Version history** (`GetVersionHistory`, `GetDocumentVersion`,
+  `RestoreVersion`, `RestoreToDate`) - it returns previous document bodies, so
+  leaving it open would bypass a column mask entirely.
+- **Enumeration and counts** - `ListCollections`, `ListViews`, `ListFiles`,
+  `Count`, `GetCollectionInfo`, `GetMemoryStats` filter or refuse. A name or a
+  document count is information about data you may not read. `ListFiles` also
+  recomputes `total_count`, which would otherwise report how many files exist in
+  types you have no grant for.
+- **Batch and set operations**, which are ordinary reads and writes.
+
+Semantics worth knowing:
+
+- A **column mask** removes fields from every document returned, and **filtering
+  on a masked field is refused** - `salary > 100000` would otherwise leak a value
+  you cannot read.
+- A **write touching a masked field is denied**, not silently dropped. Silent
+  field loss is worse than an error.
+- A **row predicate** hides documents outright: `Get` reports not-found rather
+  than "forbidden", because the difference is itself information.
+- **System collections** (`_policies`, `_views`, …) require admin. `_policies`
+  describes the whole access model.
+
+#### Denials are visible to your code
+
+`get`, `exists`, `find`, `findWithMetrics` and `count` **throw** on a denial
+rather than returning empty. This is deliberate: a denial that looked like "no
+data" would leave your application unable to tell "you may not see this" from
+"there is nothing to see", and it would silently take the wrong branch.
+
+#### Limits
+
+- **Service-wide operations cannot be expressed per project.** `GetStats`,
+  `SetReadOnly`, `CreateProject` and `DropProject` require admin of *some*
+  secured project; `ListProjects` filters. The real boundary for operator
+  surfaces is **listener separation** - do not expose an admin listener publicly.
+- `HealthCheck` and `GetReadOnlyStatus` are intentionally open; they expose no
+  user data.
+- **Auth requires TLS.** gRPC aborts if an auth processor is attached to insecure
+  credentials, so `auth.required: true` with `tls.enabled: false` is refused at
+  startup.
+
 ### Collection Configuration
 
 Per-collection runtime settings. Currently supports timestamp precision; extensible for future knobs.

service/CMakeLists.txt

@@ -69,6 +69,8 @@ set(DATABASE_SERVICE_SOURCES
     src/storage/project_store.cpp
     src/tls/cert_generator.cpp
     src/auth/auth_interceptor.cpp
+    src/auth/principal.cpp
+    src/security/policy_manager.cpp
 )
 
 # Create executable

service/src/auth/auth_interceptor.cpp

@@ -2,6 +2,8 @@
 
 #include "auth/auth_interceptor.hpp"
 
+#include "auth/principal.hpp"
+
 #include <cstring>
 #include <string_view>
 
@@ -32,7 +34,7 @@ bool starts_with_bearer(std::string_view s) {
 
 grpc::Status BearerAuthProcessor::Process(
     const InputMetadata& auth_metadata,
-    grpc::AuthContext* /*context*/,
+    grpc::AuthContext* context,
     OutputMetadata* consumed_auth_metadata,
     OutputMetadata* /*response_metadata*/) {
 
@@ -51,12 +53,15 @@ grpc::Status BearerAuthProcessor::Process(
     std::string_view tok = value.substr(7);
 
     for (const auto& k : keys_) {
-        if (constant_time_eq(tok, k)) {
-            // 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)));
+        if (constant_time_eq(tok, k.key)) {
+            // 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/auth_interceptor.hpp

@@ -7,6 +7,12 @@
 // gRPC runtime handles rejection without each handler having to
 // check. Listeners with auth.required=false don't register a
 // processor at all.
+//
+// v2.7.0 — keys are named, and the matched key's name is stamped into the
+// AuthContext as the calling PRINCIPAL. That is the only channel gRPC offers
+// between an AuthMetadataProcessor and a handler, which is why the `context`
+// argument (ignored until now) is finally used. Handlers read it back via
+// auth::principalOf(). See auth/principal.hpp.
 
 #pragma once
 
@@ -21,7 +27,12 @@ namespace smartbotic::database::auth {
 
 class BearerAuthProcessor : public grpc::AuthMetadataProcessor {
 public:
-    explicit BearerAuthProcessor(std::vector<std::string> keys)
+    struct NamedKey {
+        std::string name;  // principal stamped on a match
+        std::string key;   // bearer token
+    };
+
+    explicit BearerAuthProcessor(std::vector<NamedKey> keys)
         : keys_(std::move(keys)) {}
 
     grpc::Status Process(const InputMetadata& auth_metadata,
@@ -34,7 +45,7 @@ public:
     bool IsBlocking() const override { return false; }
 
 private:
-    std::vector<std::string> keys_;
+    std::vector<NamedKey> keys_;
 };
 
 }  // namespace smartbotic::database::auth

service/src/auth/principal.cpp

@@ -0,0 +1,54 @@
+// 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>
+
+#include <cstring>
+
+namespace smartbotic::database::auth {
+
+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;
+}
+
+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

@@ -0,0 +1,83 @@
+// v2.7.0 — caller identity, resolved PER CALL.
+//
+// How a principal is decided, in order:
+//   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, 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.
+//
+// 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;
+}
+
+namespace smartbotic::database::auth {
+
+// Reserved principal names. Rejected as key names in config, because a key
+// called "anonymous" would be indistinguishable from an unauthenticated caller
+// in a policy.
+inline constexpr const char* kPrincipalAnonymous = "anonymous";
+inline constexpr const char* kPrincipalUnnamed   = "unnamed";
+
+inline bool isReservedPrincipal(std::string_view name) {
+    return name == kPrincipalAnonymous || name == kPrincipalUnnamed;
+}
+
+// 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.hpp

@@ -8,6 +8,8 @@
 #include "replication/replication_manager.hpp"
 #include "views/view_manager.hpp"
 #include "config/collection_config_manager.hpp"
+#include "security/policy_manager.hpp"
+#include "auth/principal.hpp"
 
 #include <database.grpc.pb.h>
 
@@ -39,11 +41,19 @@ public:
         FileManager& files,
         EncryptionManager& encryption,
         ViewManager& view_manager,
-        CollectionConfigManager& config_manager
+        CollectionConfigManager& config_manager,
+        PolicyManager& policy_manager
     );
 
     ~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(
@@ -370,6 +380,80 @@ private:
     EncryptionManager& encryption_;
     ViewManager& view_manager_;
     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.
+    // On allow, `out` carries the column mask and row predicate to apply. On
+    // deny, the returned Status is PERMISSION_DENIED and the handler must
+    // return it untouched.
+    //
+    // Deliberately ONE function rather than inline checks: with ~16 call sites,
+    // a per-handler check is a place to forget, and forgetting fails open.
+    grpc::Status gate(const grpc::ServerContext* context,
+                      const std::string& qualified,
+                      smartbotic::database::Access access,
+                      smartbotic::database::Decision& out) const;
+
+    // Same for the file RPCs, which key on file type rather than collection.
+    grpc::Status gateFile(const grpc::ServerContext* context,
+                          const std::string& project,
+                          const std::string& fileType,
+                          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.
+    static grpc::Status rejectMaskedFilters(
+        const std::vector<smartbotic::database::Filter>& filters,
+        const std::vector<std::string>& mask);
     std::chrono::steady_clock::time_point startTime_ = std::chrono::steady_clock::now();
 
     // v1.6.2 — per-RPC-type streaming concurrency limits (set by DatabaseService

service/src/database_service.cpp

@@ -5,11 +5,13 @@
 #include "storage/document_store_lmdb.hpp"
 #include "storage/dual_write_mirror.hpp"
 #include "auth/auth_interceptor.hpp"
+#include "auth/principal.hpp"
 #include "tls/cert_generator.hpp"
 
 #include <fstream>
 #include <sstream>
 #include <map>
+#include <set>
 #include "storage/lmdb_env.hpp"
 #include "storage/migrate_v1_to_v2.hpp"
 #include "storage/project_store.hpp"
@@ -192,6 +194,7 @@ bool DatabaseService::initialize() {
         // Must happen AFTER persistence recovery and BEFORE migrations so any
         // migration-created documents are stamped with the correct precision.
         config_manager_->loadFromStore();
+        policy_manager_->loadFromStore();
 
         // Run migrations if enabled
         if (config_.migrations.enabled && !config_.migrations.directory.empty()) {
@@ -632,7 +635,31 @@ DatabaseService::Config DatabaseService::parseConfig(const nlohmann::json& json)
                     lc.auth.required = a.value("required", false);
                     if (a.contains("keys") && a["keys"].is_array()) {
                         for (const auto& k : a["keys"]) {
-                            if (k.is_string()) lc.auth.keys.push_back(k.get<std::string>());
+                            // v2.7.0 — a key is either a bare token (v2.4-v2.6
+                            // shape) or {"name","key"}. The name becomes the
+                            // principal that policy is written against; a bare
+                            // token maps to the reserved `unnamed`, so old
+                            // configs keep authenticating but cannot be told
+                            // apart in a policy until they are named.
+                            if (k.is_string()) {
+                                lc.auth.keys.push_back(
+                                    {smartbotic::database::auth::kPrincipalUnnamed,
+                                     k.get<std::string>()});
+                            } else if (k.is_object()) {
+                                const std::string name = k.value("name", std::string{});
+                                const std::string key  = k.value("key", std::string{});
+                                if (key.empty()) {
+                                    spdlog::warn("auth: listener {}:{} has a key entry "
+                                                 "with no `key` value - skipped",
+                                                 lc.bind, lc.port);
+                                    continue;
+                                }
+                                lc.auth.keys.push_back(
+                                    {name.empty()
+                                         ? std::string(smartbotic::database::auth::kPrincipalUnnamed)
+                                         : name,
+                                     key});
+                            }
                         }
                     }
                 }
@@ -886,6 +913,9 @@ void DatabaseService::setupComponents() {
     // Create per-collection config manager and attach it to the store so the
     // write paths route document timestamp stamps through it.
     config_manager_ = std::make_unique<CollectionConfigManager>(*store_);
+    // v2.8.0 — access policy. Constructed here; its cache is loaded after
+    // recovery alongside the view and collection-config caches.
+    policy_manager_ = std::make_unique<PolicyManager>(*store_);
     store_->setConfigManager(config_manager_.get());
 
     // v1.9.0 — disk-resident version history. Replaces the in-heap
@@ -1062,8 +1092,21 @@ void DatabaseService::setupComponents() {
 
     // Create gRPC implementations
     storageImpl_ = std::make_unique<DatabaseGrpcImpl>(
-        *this, *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_
+        *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,
@@ -1266,16 +1309,51 @@ void DatabaseService::startGrpcServer() {
                 throw std::runtime_error(
                     "Listener " + addr + ": auth.required=true but auth.keys is empty");
             }
+            // v2.7.0 — a key name is a principal, so it must be unambiguous.
+            // Refuse at startup rather than resolving a policy against a name
+            // that means two different callers.
+            {
+                std::set<std::string> seen;
+                for (const auto& k : listener.auth.keys) {
+                    if (k.name != smartbotic::database::auth::kPrincipalUnnamed &&
+                        smartbotic::database::auth::isReservedPrincipal(k.name)) {
+                        throw std::runtime_error(
+                            "Listener " + addr + ": auth key name '" + k.name +
+                            "' is reserved. `anonymous` denotes an unauthenticated "
+                            "caller and `unnamed` denotes a legacy bare key; a real "
+                            "key must not be able to impersonate either in a policy.");
+                    }
+                    if (!seen.insert(k.name).second &&
+                        k.name != smartbotic::database::auth::kPrincipalUnnamed) {
+                        throw std::runtime_error(
+                            "Listener " + addr + ": duplicate auth key name '" +
+                            k.name + "'. A principal must identify one caller, "
+                            "otherwise a policy written against it is ambiguous.");
+                    }
+                }
+            }
             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());
+            for (const auto& k : listener.auth.keys) nk.push_back({k.name, k.key});
             creds->SetAuthMetadataProcessor(
                 std::make_shared<smartbotic::database::auth::BearerAuthProcessor>(
-                    listener.auth.keys));
+                    std::move(nk)));
         }
         builder.AddListeningPort(addr, creds);
 

service/src/database_service.hpp

@@ -12,6 +12,7 @@
 #include "migrations/migration_runner.hpp"
 #include "views/view_manager.hpp"
 #include "config/collection_config_manager.hpp"
+#include "security/policy_manager.hpp"
 
 // v2.0 storage substrate (Stage 4) — DocumentStore + LmdbEnv live alongside
 // the v1.x MemoryStore during the Phase C transition. Stage 4 opens both;
@@ -67,12 +68,28 @@ public:
 
         struct AuthConfig {
             bool required = false;
+
+            // v2.7.0 — a key now carries a name, which becomes the
+            // PRINCIPAL that access policy is written against.
+            //
+            // Config accepts either form:
+            //   "keys": [ {"name": "shadowman", "key": "<base64>"} ]
+            //   "keys": [ "<base64>" ]            // v2.4-v2.6 shape
+            //
+            // A bare string still authenticates and maps to the reserved
+            // principal `unnamed`, so existing deployments keep working; they
+            // simply cannot be told apart in a policy until they are named.
+            struct NamedKey {
+                std::string name;   // principal; "unnamed" when config gave a bare string
+                std::string key;    // the bearer token itself
+            };
+
             // Bearer tokens accepted by this listener. Constant-time
             // compared against the value of the `authorization` gRPC
             // metadata header. Rotation = add new key, distribute,
             // remove old key. Empty list with required=true is a config
             // error caught at startup.
-            std::vector<std::string> keys;
+            std::vector<NamedKey> keys;
         } auth;
     };
 
@@ -214,6 +231,7 @@ public:
      * Access the view manager (for RPC handlers, migrations, etc.).
      */
     ViewManager& viewManager() { return *view_manager_; }
+    PolicyManager& policyManager() { return *policy_manager_; }
 
     /** Current read-only state (atomic, lock-free read). */
     bool isReadOnly() const { return read_only_.load(std::memory_order_acquire); }
@@ -328,6 +346,9 @@ private:
     std::unique_ptr<ReplicationManager> replication_;
     std::unique_ptr<ViewManager> view_manager_;
     std::unique_ptr<CollectionConfigManager> config_manager_;
+    // v2.8.0 — per-project access policy. Owned here so its cache is loaded
+    // once, after recovery, alongside ViewManager and CollectionConfigManager.
+    std::unique_ptr<PolicyManager> policy_manager_;
 
     // gRPC
     std::unique_ptr<DatabaseGrpcImpl> storageImpl_;

service/src/security/policy_manager.cpp

@@ -0,0 +1,459 @@
+// v2.8.0 — per-project access policy. See policy_manager.hpp.
+
+#include "security/policy_manager.hpp"
+
+#include <nlohmann/json.hpp>
+#include <spdlog/spdlog.h>
+
+#include "memory_store.hpp"
+
+namespace smartbotic::database {
+
+namespace {
+
+std::string keyOf(const std::string& project, const std::string& principal) {
+    return project + ":" + principal;
+}
+
+nlohmann::json ruleToJson(const PolicyRule& r) {
+    nlohmann::json j;
+    j["read"] = r.read;
+    j["write"] = r.write;
+    j["mask"] = r.mask;
+    nlohmann::json rows = nlohmann::json::array();
+    for (const auto& f : r.row) {
+        rows.push_back({{"field", f.field},
+                        {"op", static_cast<int>(f.op)},
+                        {"value", f.value}});
+    }
+    j["row"] = rows;
+    return j;
+}
+
+PolicyRule ruleFromJson(const nlohmann::json& j) {
+    PolicyRule r;
+    r.read = j.value("read", false);
+    r.write = j.value("write", false);
+    if (j.contains("mask") && j["mask"].is_array()) {
+        for (const auto& m : j["mask"]) {
+            if (m.is_string()) r.mask.push_back(m.get<std::string>());
+        }
+    }
+    if (j.contains("row") && j["row"].is_array()) {
+        for (const auto& f : j["row"]) {
+            if (!f.is_object()) continue;
+            Filter flt;
+            flt.field = f.value("field", "");
+            flt.op = static_cast<FilterOp>(f.value("op", 0));
+            if (f.contains("value")) flt.value = f["value"];
+            if (!flt.field.empty()) r.row.push_back(std::move(flt));
+        }
+    }
+    return r;
+}
+
+nlohmann::json policyToJson(const Policy& p) {
+    nlohmann::json j;
+    j["principal"] = p.principal;
+    j["admin"] = p.admin;
+    nlohmann::json colls = nlohmann::json::object();
+    for (const auto& [k, v] : p.collections) colls[k] = ruleToJson(v);
+    j["collections"] = colls;
+    nlohmann::json files = nlohmann::json::object();
+    for (const auto& [k, v] : p.files) files[k] = ruleToJson(v);
+    j["files"] = files;
+    return j;
+}
+
+Policy policyFromJson(const nlohmann::json& j) {
+    Policy p;
+    p.principal = j.value("principal", "");
+    p.admin = j.value("admin", false);
+    if (j.contains("collections") && j["collections"].is_object()) {
+        for (const auto& [k, v] : j["collections"].items()) {
+            p.collections[k] = ruleFromJson(v);
+        }
+    }
+    if (j.contains("files") && j["files"].is_object()) {
+        for (const auto& [k, v] : j["files"].items()) {
+            p.files[k] = ruleFromJson(v);
+        }
+    }
+    return p;
+}
+
+}  // namespace
+
+PolicyManager::PolicyManager(MemoryStore& store) : store_(store) {}
+
+void PolicyManager::loadFromStore() {
+    std::unique_lock lock(mutex_);
+    policies_.clear();
+    security_.clear();
+
+    CollectionOptions opts;
+    opts.autoCreateId = false;
+    store_.createCollection(SYSTEM_COLLECTION, opts);
+
+    // Query::limit defaults to 100 and limit=0 returns NOTHING, not everything
+    // (`end = min(offset + limit, size)`), so paging is mandatory here. Loading
+    // a partial policy set would silently under-enforce, which is the worst
+    // possible failure for this component. Note ViewManager has the same latent
+    // truncation on installs with more than 100 views.
+    std::vector<Document> records;
+    {
+        constexpr uint32_t kPage = 500;
+        uint32_t offset = 0;
+        while (true) {
+            Query q;
+            q.limit = kPage;
+            q.offset = offset;
+            QueryResult page;
+            try {
+                page = store_.find(SYSTEM_COLLECTION, q);
+            } catch (const std::exception& e) {
+                spdlog::warn("PolicyManager: could not read {}: {}",
+                             SYSTEM_COLLECTION, e.what());
+                return;
+            }
+            for (auto& d : page.documents) records.push_back(std::move(d));
+            if (page.documents.size() < kPage) break;
+            offset += kPage;
+        }
+    }
+
+    uint32_t policies = 0, projects = 0;
+    for (const auto& doc : records) {
+        // id is "<project>:<principal>" or "<project>:__security__"
+        const auto sep = doc.id.find(':');
+        if (sep == std::string::npos) {
+            spdlog::warn("PolicyManager: skipping unqualified policy id '{}'", doc.id);
+            continue;
+        }
+        const std::string project = doc.id.substr(0, sep);
+        const std::string tail = doc.id.substr(sep + 1);
+
+        try {
+            const auto data = doc.data();
+            if (tail == SECURITY_DOC) {
+                ProjectSecurity sec;
+                sec.enabled = data.value("enabled", false);
+                sec.mode = data.value("mode", std::string("enforce")) == "audit"
+                               ? SecurityMode::Audit
+                               : SecurityMode::Enforce;
+                security_[project] = sec;
+                ++projects;
+            } else {
+                Policy p = policyFromJson(data);
+                if (p.principal.empty()) p.principal = tail;
+                policies_[doc.id] = std::move(p);
+                ++policies;
+            }
+        } catch (const std::exception& e) {
+            spdlog::warn("PolicyManager: skipping malformed record '{}': {}",
+                         doc.id, e.what());
+        }
+    }
+
+    recountSecured();
+    if (policies || projects) {
+        spdlog::info("PolicyManager: loaded {} policy record(s) across {} "
+                     "configured project(s); {} project(s) have security enabled",
+                     policies, projects,
+                     secured_count_.load(std::memory_order_relaxed));
+    }
+}
+
+void PolicyManager::recountSecured() {
+    uint32_t n = 0;
+    for (const auto& [_, sec] : security_) {
+        if (sec.enabled) ++n;
+    }
+    secured_count_.store(n, std::memory_order_relaxed);
+}
+
+ProjectSecurity PolicyManager::securityOf(const std::string& project) const {
+    std::shared_lock lock(mutex_);
+    auto it = security_.find(project);
+    if (it == security_.end()) return {};  // disabled
+    return it->second;
+}
+
+Decision PolicyManager::evaluate(const std::string& project,
+                                  const std::string& principal,
+                                  const std::string& target,
+                                  Access access,
+                                  bool is_file) const {
+    std::shared_lock lock(mutex_);
+
+    auto sec_it = security_.find(project);
+    const ProjectSecurity sec =
+        (sec_it == security_.end()) ? ProjectSecurity{} : sec_it->second;
+
+    // Security off: allow everything. This is the default and the entire
+    // pre-2.8.0 world, so it must cost almost nothing.
+    if (!sec.enabled) return Decision{};
+
+    Decision d;
+    auto deny = [&](const std::string& why) {
+        d.allowed = false;
+        d.reason = why;
+        // Audit mode: evaluate honestly, log, then allow anyway. The caller is
+        // told nothing; the operator sees exactly what enforce mode would do.
+        if (sec.mode == SecurityMode::Audit) {
+            d.allowed = true;
+            d.audited_denial = true;
+            spdlog::warn("policy AUDIT: would deny principal '{}' {} on {} '{}' "
+                         "in project '{}' ({}). Not denied - project is in audit "
+                         "mode.",
+                         principal,
+                         access == Access::Read ? "read" : "write",
+                         is_file ? "file type" : "collection",
+                         target, project, why);
+        }
+        return d;
+    };
+
+    // System collections are never reachable by a non-admin. `_policies` in
+    // particular: read access to it would expose the whole access model, and
+    // write access would be privilege escalation.
+    const bool is_system = !is_file && !target.empty() && target.front() == '_';
+
+    auto pol_it = policies_.find(keyOf(project, principal));
+    if (pol_it == policies_.end()) {
+        return deny("no policy for this principal (project is deny-by-default)");
+    }
+    const Policy& p = pol_it->second;
+
+    if (p.admin) return d;  // admin: unrestricted within the project
+    if (is_system) return deny("system collections are admin-only");
+
+    const auto& table = is_file ? p.files : p.collections;
+    auto rule_it = table.find(target);
+    if (rule_it == table.end()) {
+        rule_it = table.find("*");
+        if (rule_it == table.end()) {
+            return deny("no rule for this " +
+                        std::string(is_file ? "file type" : "collection"));
+        }
+    }
+    const PolicyRule& rule = rule_it->second;
+
+    const bool granted = (access == Access::Read) ? rule.read : rule.write;
+    if (!granted) {
+        return deny(std::string(access == Access::Read ? "read" : "write") +
+                    " not granted");
+    }
+
+    d.mask = rule.mask;
+    d.row = rule.row;
+    return d;
+}
+
+Decision PolicyManager::authorize(const std::string& project,
+                                   const std::string& principal,
+                                   const std::string& collection,
+                                   Access access) const {
+    return evaluate(project, principal, collection, access, /*is_file=*/false);
+}
+
+Decision PolicyManager::authorizeFile(const std::string& project,
+                                       const std::string& principal,
+                                       const std::string& fileType,
+                                       Access access) const {
+    return evaluate(project, principal, fileType, access, /*is_file=*/true);
+}
+
+bool PolicyManager::mayAdminister(const std::string& project,
+                                   const std::string& principal) const {
+    std::shared_lock lock(mutex_);
+    auto sec_it = security_.find(project);
+    // Security off: policy is freely editable, which is how the first policy
+    // gets created in the first place.
+    if (sec_it == security_.end() || !sec_it->second.enabled) return true;
+
+    auto it = policies_.find(keyOf(project, principal));
+    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) {
+    if (sec.enabled && !allow_enable_) {
+        // Refuse to arm a boundary that is not fully wired yet.
+        //
+        // The engine, its safety rules and its tests are done, but the gate is
+        // not yet called from every read/write handler. A project enabled in
+        // that state would enforce on some paths and silently allow on others,
+        // which is worse than no security at all: it reports protection that is
+        // not there. See kEnforcementCoverageComplete in policy_manager.hpp for
+        // the remaining call sites.
+        errorOut =
+            "refusing to enable security: enforcement is not yet wired into "
+            "every handler in this build, so a project would be protected on "
+            "some paths and open on others. Policies can be created and "
+            "inspected now; enabling is unlocked once coverage is complete.";
+        return false;
+    }
+    if (sec.enabled) {
+        // Lockout prevention. Refusing here is the whole safety mechanism:
+        // enabling deny-by-default with no admin would leave the project
+        // permanently unmanageable, including by the operator.
+        std::shared_lock lock(mutex_);
+        bool has_admin = false;
+        const std::string prefix = project + ":";
+        for (const auto& [key, pol] : policies_) {
+            if (key.rfind(prefix, 0) == 0 && pol.admin) {
+                has_admin = true;
+                break;
+            }
+        }
+        if (!has_admin) {
+            errorOut = "refusing to enable security for project '" + project +
+                       "': no policy in this project has admin=true, so enabling "
+                       "deny-by-default would lock everyone out permanently. "
+                       "Create an admin policy first.";
+            return false;
+        }
+    }
+
+    Document doc;
+    doc.id = keyOf(project, SECURITY_DOC);
+    doc.collection = SYSTEM_COLLECTION;
+    nlohmann::json data;
+    data["enabled"] = sec.enabled;
+    data["mode"] = sec.mode == SecurityMode::Audit ? "audit" : "enforce";
+    doc.set_data(data);
+
+    try {
+        store_.remove(SYSTEM_COLLECTION, doc.id);
+        store_.insert(SYSTEM_COLLECTION, doc);
+    } catch (const std::exception& e) {
+        errorOut = std::string("failed to persist security record: ") + e.what();
+        return false;
+    }
+
+    {
+        std::unique_lock lock(mutex_);
+        security_[project] = sec;
+        recountSecured();
+    }
+    spdlog::warn("policy: project '{}' security {} (mode={})", project,
+                 sec.enabled ? "ENABLED" : "disabled",
+                 sec.mode == SecurityMode::Audit ? "audit" : "enforce");
+    return true;
+}
+
+bool PolicyManager::setPolicy(const std::string& project, const Policy& policy,
+                               std::string& errorOut) {
+    if (policy.principal.empty()) {
+        errorOut = "policy principal must not be empty";
+        return false;
+    }
+    if (policy.principal == SECURITY_DOC) {
+        errorOut = "'" + std::string(SECURITY_DOC) + "' is reserved";
+        return false;
+    }
+
+    Document doc;
+    doc.id = keyOf(project, policy.principal);
+    doc.collection = SYSTEM_COLLECTION;
+    doc.set_data(policyToJson(policy));
+
+    try {
+        store_.remove(SYSTEM_COLLECTION, doc.id);
+        store_.insert(SYSTEM_COLLECTION, doc);
+    } catch (const std::exception& e) {
+        errorOut = std::string("failed to persist policy: ") + e.what();
+        return false;
+    }
+
+    {
+        std::unique_lock lock(mutex_);
+        policies_[doc.id] = policy;
+    }
+    spdlog::info("policy: set for principal '{}' in project '{}' (admin={})",
+                 policy.principal, project, policy.admin);
+    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) {
+    const std::string key = keyOf(project, principal);
+    {
+        // Removing the last admin of a secured project is the same lockout as
+        // enabling without one, so it is refused for the same reason.
+        std::shared_lock lock(mutex_);
+        auto sec_it = security_.find(project);
+        const bool secured = sec_it != security_.end() && sec_it->second.enabled;
+        auto target = policies_.find(key);
+        if (secured && target != policies_.end() && target->second.admin) {
+            const std::string prefix = project + ":";
+            uint32_t admins = 0;
+            for (const auto& [k, pol] : policies_) {
+                if (k.rfind(prefix, 0) == 0 && pol.admin) ++admins;
+            }
+            if (admins <= 1) {
+                errorOut = "refusing to remove the last admin policy of secured "
+                           "project '" + project + "': it would leave the project "
+                           "unmanageable. Disable security or add another admin "
+                           "first.";
+                return false;
+            }
+        }
+    }
+
+    try {
+        store_.remove(SYSTEM_COLLECTION, key);
+    } catch (const std::exception& e) {
+        errorOut = std::string("failed to remove policy: ") + e.what();
+        return false;
+    }
+    {
+        std::unique_lock lock(mutex_);
+        policies_.erase(key);
+    }
+    return true;
+}
+
+std::optional<Policy> PolicyManager::getPolicy(const std::string& project,
+                                                const std::string& principal) const {
+    std::shared_lock lock(mutex_);
+    auto it = policies_.find(keyOf(project, principal));
+    if (it == policies_.end()) return std::nullopt;
+    return it->second;
+}
+
+std::vector<Policy> PolicyManager::listPolicies(const std::string& project) const {
+    std::shared_lock lock(mutex_);
+    std::vector<Policy> out;
+    const std::string prefix = project + ":";
+    for (const auto& [key, pol] : policies_) {
+        if (key.rfind(prefix, 0) == 0) out.push_back(pol);
+    }
+    return out;
+}
+
+}  // namespace smartbotic::database

service/src/security/policy_manager.hpp

@@ -0,0 +1,224 @@
+// v2.8.0 — per-project row- and column-level access policy.
+//
+// Design and the approved decisions behind it:
+// docs/superpowers/specs/2026-08-08-rls-cls-design.md
+//
+// Shape
+// -----
+// One document per (project, principal) in the global `_policies` system
+// collection, id = "<project>:<principal>". A reserved id
+// "<project>:__security__" carries the per-project enable flag and mode. Using
+// one global collection keyed by qualified id matches how `_views` and
+// `_collection_meta` already work, so nothing new is needed to make policy
+// durable: `_policies` is an ordinary collection and is WAL'd and snapshotted
+// with everything else.
+//
+// Default posture
+// ---------------
+// Security is OFF per project. With it off, `authorize()` allows everything and
+// costs one cached map lookup. Once ON for a project, access inside that project
+// is DENY-BY-DEFAULT: a principal with no matching entry gets nothing. That is
+// what makes an enabled project an actual boundary rather than a suggestion.
+//
+// Two safety properties, both deliberate:
+//
+//   * Lockout is structurally impossible. Enabling security is REFUSED unless
+//     at least one policy in that project has `admin: true`. Care is not a
+//     safety mechanism; a refused enable is.
+//
+//   * `mode: "audit"` evaluates every policy and logs what it WOULD deny, then
+//     allows the request. This is how a live project gets switched on: enable in
+//     audit, watch until the log is quiet, then flip to enforce. Without it,
+//     deny-by-default would strand existing consumers the instant it was armed.
+
+#pragma once
+
+#include <atomic>
+#include <mutex>
+#include <optional>
+#include <shared_mutex>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "document.hpp"
+
+namespace smartbotic::database {
+
+class MemoryStore;
+
+// What a principal may do with one collection or file type.
+struct PolicyRule {
+    bool read = false;
+    bool write = false;
+    // Column mask: dot-notation paths removed from every document returned.
+    // Same path syntax as a view's `exclude`, and applied with the same
+    // applyProjection() helper.
+    std::vector<std::string> mask;
+    // Row predicate, AND-merged with the caller's filters exactly as a view's
+    // `where` already is.
+    std::vector<Filter> row;
+};
+
+struct Policy {
+    std::string principal;
+    // admin grants read+write on everything in the project plus the right to
+    // modify `_policies`. At least one admin is required to enable security.
+    bool admin = false;
+    // Keyed by collection name (bare, within the project) or file type.
+    // "*" is a fallback; an exact match always wins over it.
+    std::unordered_map<std::string, PolicyRule> collections;
+    std::unordered_map<std::string, PolicyRule> files;
+    uint64_t updatedAt = 0;
+};
+
+// 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.
+//
+// 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.
+//
+// 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 = true;
+
+enum class SecurityMode { Enforce, Audit };
+
+struct ProjectSecurity {
+    bool enabled = false;
+    SecurityMode mode = SecurityMode::Enforce;
+};
+
+// The answer handlers act on.
+struct Decision {
+    bool allowed = true;
+    // Populated only when allowed. Empty mask means no columns removed.
+    std::vector<std::string> mask;
+    std::vector<Filter> row;
+    // Set when the decision was a denial that audit mode converted to an
+    // allow. Handlers ignore it; it exists so the caller can be told nothing
+    // while the operator sees everything.
+    bool audited_denial = false;
+    std::string reason;
+};
+
+enum class Access { Read, Write };
+
+class PolicyManager {
+public:
+    static constexpr const char* SYSTEM_COLLECTION = "_policies";
+    static constexpr const char* SECURITY_DOC = "__security__";
+
+    explicit PolicyManager(MemoryStore& store);
+
+    // Load every policy into cache. Call once at startup, after recovery.
+    void loadFromStore();
+
+    // Hot path. Returns allow-everything when the project has security off,
+    // which is the default and the case for every pre-2.8.0 deployment.
+    [[nodiscard]] Decision authorize(const std::string& project,
+                                     const std::string& principal,
+                                     const std::string& collection,
+                                     Access access) const;
+
+    // Same evaluation against a file type rather than a collection.
+    [[nodiscard]] Decision authorizeFile(const std::string& project,
+                                         const std::string& principal,
+                                         const std::string& fileType,
+                                         Access access) const;
+
+    // True when `principal` is an admin of `project`, or when the project has
+    // security disabled (in which case everyone may manage policy).
+    [[nodiscard]] bool mayAdminister(const std::string& project,
+                                     const std::string& principal) const;
+
+    [[nodiscard]] ProjectSecurity securityOf(const std::string& project) const;
+
+    // Refused when enabling without at least one admin policy in the project.
+    bool setSecurity(const std::string& project, const ProjectSecurity& sec,
+                     std::string& errorOut);
+
+    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.
+    void allowEnableForTests() { allow_enable_ = true; }
+
+    // True if any project has security enabled. Lets handlers skip the
+    // per-request work entirely on installations that never turned it on.
+    [[nodiscard]] bool anyProjectSecured() const noexcept {
+        return secured_count_.load(std::memory_order_relaxed) > 0;
+    }
+
+private:
+    MemoryStore& store_;
+    mutable std::shared_mutex mutex_;
+    // key: "<project>:<principal>"
+    std::unordered_map<std::string, Policy> policies_;
+    std::unordered_map<std::string, ProjectSecurity> security_;
+    std::atomic<uint32_t> secured_count_{0};
+    bool allow_enable_ = kEnforcementCoverageComplete;
+
+    [[nodiscard]] Decision evaluate(
+        const std::string& project,
+        const std::string& principal,
+        const std::string& target,
+        Access access,
+        bool is_file) const;
+
+    void recountSecured();
+};
+
+}  // namespace smartbotic::database

tests/CMakeLists.txt

@@ -583,3 +583,43 @@ find_package(Threads REQUIRED)
 target_link_libraries(test_file_project_scope PRIVATE Threads::Threads)
 
 add_test(NAME test_file_project_scope COMMAND test_file_project_scope)
+
+# v2.8.0 — access policy engine. The critical assertions are the safety ones:
+# security off allows everything (so upgrades are inert), enabling is refused
+# without an admin policy, the last admin cannot be removed, and audit mode
+# evaluates honestly but allows.
+add_executable(test_policy_manager
+    test_policy_manager.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/security/policy_manager.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/memory_store.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/config/collection_config_manager.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/persistence/history_store.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/persistence/wal.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/json_parse.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/doc_binary.cpp
+)
+
+target_include_directories(test_policy_manager PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src
+    ${yyjson_INCLUDE_DIRS}
+)
+
+target_link_libraries(test_policy_manager PRIVATE ${yyjson_LIBRARIES})
+
+if(TARGET nlohmann_json::nlohmann_json)
+    target_link_libraries(test_policy_manager PRIVATE nlohmann_json::nlohmann_json)
+else()
+    target_include_directories(test_policy_manager PRIVATE ${NLOHMANN_JSON_INCLUDE_DIRS})
+endif()
+
+if(TARGET spdlog::spdlog)
+    target_link_libraries(test_policy_manager PRIVATE spdlog::spdlog)
+else()
+    target_link_libraries(test_policy_manager PRIVATE ${SPDLOG_LIBRARIES})
+    target_include_directories(test_policy_manager PRIVATE ${SPDLOG_INCLUDE_DIRS})
+endif()
+
+find_package(Threads REQUIRED)
+target_link_libraries(test_policy_manager PRIVATE Threads::Threads)
+
+add_test(NAME test_policy_manager COMMAND test_policy_manager)

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/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_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;
 }

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"

tests/test_policy_manager.cpp

@@ -0,0 +1,342 @@
+// v2.8.0 — access policy engine tests.
+//
+// The properties that matter most here are the safety ones, because they are
+// what make the feature usable on a live system rather than a footgun:
+//
+//   * security OFF is the default and allows everything, so upgrading changes
+//     nothing for any existing deployment;
+//   * enabling is REFUSED without an admin policy, which makes lockout
+//     structurally impossible rather than a matter of operator care;
+//   * removing the last admin of a secured project is refused for the same
+//     reason;
+//   * audit mode evaluates honestly and logs, but allows - the migration path
+//     for switching a live project on;
+//   * once enforcing, an unlisted principal gets nothing (deny-by-default), and
+//     system collections are admin-only, since read access to `_policies` would
+//     expose the whole access model and write access would be escalation.
+
+#include <atomic>
+#include <filesystem>
+#include <iostream>
+#include <string>
+#include <unistd.h>
+
+#include <nlohmann/json.hpp>
+
+#include "memory_store.hpp"
+#include "security/policy_manager.hpp"
+
+// setSecurity(enabled=true) is refused while kEnforcementCoverageComplete is
+// false - see policy_manager.hpp. The fixture calls allowEnableForTests() so
+// these tests can still exercise enforcing behaviour; a separate test below
+// asserts the guard itself.
+
+namespace fs = std::filesystem;
+using namespace smartbotic::database;
+
+namespace {
+
+int g_pass = 0;
+int g_fail = 0;
+
+void check(bool cond, const char* msg) {
+    if (cond) {
+        ++g_pass;
+    } else {
+        ++g_fail;
+        std::cerr << "FAIL: " << msg << "\n";
+    }
+}
+
+struct Fixture {
+    MemoryStore store;
+    PolicyManager pm;
+
+    Fixture() : store(MemoryStore::Config{}), pm(store) {
+        store.start();
+        pm.loadFromStore();
+        // These tests are the engine's own; they must be able to reach
+        // enforcing behaviour even though the deployment-level guard is armed.
+        pm.allowEnableForTests();
+    }
+    ~Fixture() { store.stop(); }
+};
+
+Policy mkPolicy(const std::string& principal, bool admin = false) {
+    Policy p;
+    p.principal = principal;
+    p.admin = admin;
+    return p;
+}
+
+PolicyRule rw(bool r, bool w) {
+    PolicyRule x;
+    x.read = r;
+    x.write = w;
+    return x;
+}
+
+// -------------------------------------------------------------------------
+
+void test_security_off_allows_everything() {
+    Fixture f;
+    // No policies, no security record: the pre-2.8.0 world.
+    auto d = f.pm.authorize("acme", "nobody", "users", Access::Read);
+    check(d.allowed, "security off allows an unknown principal to read");
+    check(f.pm.authorize("acme", "nobody", "users", Access::Write).allowed,
+          "security off allows writes too");
+    check(f.pm.authorize("acme", "nobody", "_policies", Access::Read).allowed,
+          "security off does not even guard system collections");
+    check(!f.pm.anyProjectSecured(), "no project is secured by default");
+    check(f.pm.mayAdminister("acme", "anyone"),
+          "with security off anyone may create the first policy");
+}
+
+void test_enabling_without_admin_is_refused() {
+    Fixture f;
+    std::string err;
+    ProjectSecurity sec;
+    sec.enabled = true;
+
+    check(!f.pm.setSecurity("acme", sec, err),
+          "enabling security with no admin policy must be refused");
+    check(err.find("admin") != std::string::npos,
+          "and the error must say why - no admin policy");
+    check(!f.pm.securityOf("acme").enabled, "security stays off after refusal");
+
+    // Non-admin policies are not enough.
+    auto p = mkPolicy("shadowman");
+    p.collections["users"] = rw(true, false);
+    check(f.pm.setPolicy("acme", p, err), "a non-admin policy can be created");
+    check(!f.pm.setSecurity("acme", sec, err),
+          "a non-admin policy does not satisfy the admin requirement");
+
+    check(f.pm.setPolicy("acme", mkPolicy("ops", /*admin=*/true), err),
+          "an admin policy can be created");
+    check(f.pm.setSecurity("acme", sec, err),
+          "with an admin present, enabling succeeds");
+    check(f.pm.securityOf("acme").enabled, "and security is now on");
+    check(f.pm.anyProjectSecured(), "anyProjectSecured reflects it");
+}
+
+void test_deny_by_default_once_enforcing() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto reader = mkPolicy("reader");
+    reader.collections["users"] = rw(true, false);
+    f.pm.setPolicy("acme", reader, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(!f.pm.authorize("acme", "stranger", "users", Access::Read).allowed,
+          "a principal with no policy gets nothing");
+    check(f.pm.authorize("acme", "reader", "users", Access::Read).allowed,
+          "a granted read is allowed");
+    check(!f.pm.authorize("acme", "reader", "users", Access::Write).allowed,
+          "read does not imply write");
+    check(!f.pm.authorize("acme", "reader", "sessions", Access::Read).allowed,
+          "a collection with no rule is denied even for a known principal");
+    check(f.pm.authorize("acme", "ops", "anything", Access::Write).allowed,
+          "admin may write anything in the project");
+
+    // Another project is untouched.
+    check(f.pm.authorize("other", "stranger", "users", Access::Read).allowed,
+          "enabling one project must not affect another");
+}
+
+void test_wildcard_rule_and_exact_match_precedence() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto p = mkPolicy("svc");
+    p.collections["*"] = rw(true, false);
+    p.collections["secrets"] = rw(false, false);
+    f.pm.setPolicy("acme", p, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(f.pm.authorize("acme", "svc", "anything", Access::Read).allowed,
+          "the * fallback grants read on an unlisted collection");
+    check(!f.pm.authorize("acme", "svc", "secrets", Access::Read).allowed,
+          "an exact rule overrides the * fallback, even to deny");
+}
+
+void test_system_collections_are_admin_only() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto p = mkPolicy("svc");
+    p.collections["*"] = rw(true, true);
+    f.pm.setPolicy("acme", p, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(!f.pm.authorize("acme", "svc", "_policies", Access::Read).allowed,
+          "a * grant must NOT reach _policies - that would expose the access model");
+    check(!f.pm.authorize("acme", "svc", "_policies", Access::Write).allowed,
+          "nor allow writing it - that would be privilege escalation");
+    check(!f.pm.authorize("acme", "svc", "_views", Access::Read).allowed,
+          "no system collection is reachable via *");
+    check(f.pm.authorize("acme", "ops", "_policies", Access::Write).allowed,
+          "an admin may still manage system collections");
+    check(!f.pm.mayAdminister("acme", "svc"), "a non-admin may not administer policy");
+    check(f.pm.mayAdminister("acme", "ops"), "an admin may");
+}
+
+void test_mask_and_row_predicate_are_returned() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto p = mkPolicy("svc");
+    PolicyRule r = rw(true, false);
+    r.mask = {"ssn", "profile.dob"};
+    Filter tenant;
+    tenant.field = "tenant";
+    tenant.op = FilterOp::EQ;
+    tenant.value = "acme";
+    r.row = {tenant};
+    p.collections["users"] = r;
+    f.pm.setPolicy("acme", p, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    auto d = f.pm.authorize("acme", "svc", "users", Access::Read);
+    check(d.allowed, "granted");
+    check(d.mask.size() == 2, "the column mask comes back with the decision");
+    check(d.mask[0] == "ssn", "mask paths preserved");
+    check(d.row.size() == 1 && d.row[0].field == "tenant",
+          "the row predicate comes back so the handler can AND-merge it");
+}
+
+void test_audit_mode_logs_but_allows() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    ProjectSecurity sec;
+    sec.enabled = true;
+    sec.mode = SecurityMode::Audit;
+    check(f.pm.setSecurity("acme", sec, err), "audit mode can be enabled");
+
+    auto d = f.pm.authorize("acme", "stranger", "users", Access::Read);
+    check(d.allowed, "audit mode ALLOWS what enforce mode would deny");
+    check(d.audited_denial, "but records that it was really a denial");
+    check(!d.reason.empty(), "and keeps the reason for the operator log");
+
+    // Flip to enforce and the same request is refused.
+    sec.mode = SecurityMode::Enforce;
+    f.pm.setSecurity("acme", sec, err);
+    auto d2 = f.pm.authorize("acme", "stranger", "users", Access::Read);
+    check(!d2.allowed, "enforce mode denies the same request");
+    check(!d2.audited_denial, "and does not mark it as merely audited");
+}
+
+void test_cannot_remove_last_admin_of_secured_project() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(!f.pm.removePolicy("acme", "ops", err),
+          "removing the last admin of a secured project must be refused");
+    check(err.find("last admin") != std::string::npos, "with a clear reason");
+
+    check(f.pm.setPolicy("acme", mkPolicy("ops2", true), err), "add a second admin");
+    check(f.pm.removePolicy("acme", "ops", err),
+          "now the first admin can be removed");
+    check(!f.pm.removePolicy("acme", "ops2", err),
+          "but not the remaining last one");
+}
+
+void test_policies_survive_reload() {
+    Fixture f;
+    std::string err;
+    auto p = mkPolicy("svc");
+    PolicyRule r = rw(true, false);
+    r.mask = {"ssn"};
+    p.collections["users"] = r;
+    f.pm.setPolicy("acme", p, err);
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    ProjectSecurity sec; sec.enabled = true; sec.mode = SecurityMode::Audit;
+    f.pm.setSecurity("acme", sec, err);
+
+    // Reload from the same store, as a restart would.
+    f.pm.loadFromStore();
+
+    check(f.pm.securityOf("acme").enabled, "enable flag survives reload");
+    check(f.pm.securityOf("acme").mode == SecurityMode::Audit,
+          "mode survives reload");
+    auto got = f.pm.getPolicy("acme", "svc");
+    check(got.has_value(), "policy survives reload");
+    check(got && got->collections.count("users") == 1, "rules survive reload");
+    check(got && got->collections["users"].mask.size() == 1,
+          "the column mask survives reload");
+    check(f.pm.listPolicies("acme").size() == 2, "both policies listed");
+    check(f.pm.listPolicies("other").empty(),
+          "listing is scoped to the project");
+}
+
+void test_file_rules_are_separate_from_collections() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto p = mkPolicy("svc");
+    p.collections["*"] = rw(true, true);
+    p.files["plugin"] = rw(true, false);
+    f.pm.setPolicy("acme", p, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(f.pm.authorizeFile("acme", "svc", "plugin", Access::Read).allowed,
+          "a granted file type reads");
+    check(!f.pm.authorizeFile("acme", "svc", "plugin", Access::Write).allowed,
+          "file write is separately gated");
+    check(!f.pm.authorizeFile("acme", "svc", "document", Access::Read).allowed,
+          "an unlisted file type is denied - a collection * does not cover files");
+}
+
+// The deployment guard itself: without the test seam, arming security must be
+// refused while enforcement is not wired into every handler. A project
+// protected on some paths and open on others reports safety it does not have.
+void test_enable_is_refused_while_coverage_incomplete() {
+    MemoryStore store(MemoryStore::Config{});
+    store.start();
+    PolicyManager pm(store);          // note: NO allowEnableForTests()
+    pm.loadFromStore();
+    std::string err;
+    pm.setPolicy("acme", mkPolicy("ops", true), err);
+    ProjectSecurity sec; sec.enabled = true;
+
+    if (kEnforcementCoverageComplete) {
+        check(pm.setSecurity("acme", sec, err),
+              "coverage complete: enabling is permitted");
+    } else {
+        check(!pm.setSecurity("acme", sec, err),
+              "coverage incomplete: enabling must be refused even with an admin");
+        check(err.find("not yet wired") != std::string::npos,
+              "and the refusal must say enforcement is incomplete");
+        check(!pm.securityOf("acme").enabled, "security stays off");
+    }
+    store.stop();
+}
+
+}  // namespace
+
+int main() {
+    std::cout << "=== test_policy_manager ===\n";
+    test_security_off_allows_everything();
+    test_enabling_without_admin_is_refused();
+    test_deny_by_default_once_enforcing();
+    test_wildcard_rule_and_exact_match_precedence();
+    test_system_collections_are_admin_only();
+    test_mask_and_row_predicate_are_returned();
+    test_audit_mode_logs_but_allows();
+    test_cannot_remove_last_admin_of_secured_project();
+    test_policies_survive_reload();
+    test_file_rules_are_separate_from_collections();
+    test_enable_is_refused_while_coverage_incomplete();
+
+    std::cout << "passed: " << g_pass << ", failed: " << g_fail << "\n";
+    return g_fail == 0 ? 0 : 1;
+}