Column level permissions immutable

README.md

@@ -416,7 +416,8 @@ $database->updateCollection(
         Permission::update(Role::any()),
         Permission::delete(Role::any())
     ],
-    documentSecurity: true
+    documentSecurity: true,
+    columnSecurity: false
 );
 
 // Get Collection

docker-compose.yml

@@ -19,6 +19,15 @@ services:
       - ./docker-compose.yml:/usr/src/code/docker-compose.yml
     environment:
       PHP_IDE_CONFIG: serverName=tests
+      # Xdebug stays installed but idle. dev/xdebug.ini sets mode=develop,debug,profile
+      # with start_with_request=yes, so it engages on every request -- and it is a trap in
+      # two directions. It costs enough time to push the timeout-sensitive tests past
+      # their thresholds, and it inflates memory per allocation, so tests that assert a
+      # memory ceiling measure xdebug rather than the code. Either way the run fails
+      # locally and passes in CI, where the image is built without xdebug.so at all, and
+      # the failure looks like a real regression. Override per command to step-debug:
+      # docker compose exec -e XDEBUG_MODE=debug tests ...
+      XDEBUG_MODE: off
     depends_on:
       postgres:
         condition: service_healthy

src/Database/Adapter.php

@@ -867,9 +867,10 @@ abstract public function deleteDocuments(string $collection, array $sequences, a
      * @param array<string, mixed> $cursor
      * @param string $cursorDirection
      * @param string $forPermission
+     * @param array<string> $columnPermissions columns that must be readable on the row
      * @return array<Document>
      */
-    abstract public function find(Document $collection, array $queries = [], ?int $limit = 25, ?int $offset = null, array $orderAttributes = [], array $orderTypes = [], array $cursor = [], string $cursorDirection = Database::CURSOR_AFTER, string $forPermission = Database::PERMISSION_READ): array;
+    abstract public function find(Document $collection, array $queries = [], ?int $limit = 25, ?int $offset = null, array $orderAttributes = [], array $orderTypes = [], array $cursor = [], string $cursorDirection = Database::CURSOR_AFTER, string $forPermission = Database::PERMISSION_READ, array $columnPermissions = []): array;
 
     /**
      * Sum an attribute
@@ -879,9 +880,10 @@ abstract public function find(Document $collection, array $queries = [], ?int $l
      * @param array<Query> $queries
      * @param int|null $max
      *
+     * @param array<string> $columnPermissions columns that must be readable on the row
      * @return int|float
      */
-    abstract public function sum(Document $collection, string $attribute, array $queries = [], ?int $max = null): float|int;
+    abstract public function sum(Document $collection, string $attribute, array $queries = [], ?int $max = null, array $columnPermissions = []): float|int;
 
     /**
      * Count Documents
@@ -890,9 +892,10 @@ abstract public function sum(Document $collection, string $attribute, array $que
      * @param array<Query> $queries
      * @param int|null $max
      *
+     * @param array<string> $columnPermissions columns that must be readable on the row
      * @return int
      */
-    abstract public function count(Document $collection, array $queries = [], ?int $max = null): int;
+    abstract public function count(Document $collection, array $queries = [], ?int $max = null, array $columnPermissions = []): int;
 
     /**
      * Get Collection Size of the raw data
@@ -1011,6 +1014,23 @@ abstract public function getSupportForAttributes(): bool;
      */
     abstract public function getSupportForSchemaAttributes(): bool;
 
+    /**
+     * Can a permission be scoped to a single column?
+     *
+     * @return bool
+     */
+    abstract public function getSupportForColumnPermissions(): bool;
+
+
+    /**
+     * Drop every permission scoped to a column that no longer exists.
+     *
+     * @param Document $collection
+     * @param string $column
+     * @return int documents whose $permissions changed
+     */
+    abstract public function deleteColumnPermissions(Document $collection, string $column): int;
+
     /**
      * Are schema indexes supported?
      *

src/Database/Adapter/MariaDB.php

@@ -189,24 +189,41 @@ public function createCollection(string $name, array $attributes = [], array $in
         $collection .= ")";
         $collection = $this->trigger(Database::EVENT_COLLECTION_CREATE, $collection);
 
+        // _column scopes a permission to a single column. An empty string means
+        // every column, which is how every permission written before column-level
+        // permissions reads. It is NOT NULL on purpose: MySQL and MariaDB treat
+        // NULLs as distinct in a UNIQUE index, so a nullable _column would let
+        // duplicate permission rows slip past the unique index.
+        //
+        // Sized to MAX_UID_DEFAULT_LENGTH rather than the 255 the other string members
+        // use. The unique index holds four of those, and in utf8mb4 a fifth
+        // VARCHAR(255) member
+        // takes the key past InnoDB's 3072-byte limit -- MySQL refuses the CREATE with
+        // "Specified key was too long", though MariaDB allows it, so testing on one
+        // says nothing about the other. The Permissions validator already caps a
+        // scoped column at this same constant, so nothing storable is lost.
         $permissions = "
             CREATE TABLE {$this->getSQLTable($id . '_perms')} (
                 _id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
                 _type VARCHAR(12) NOT NULL,
                 _permission VARCHAR(255) NOT NULL,
+                _column VARCHAR(" . Database::MAX_PERMISSION_COLUMN_LENGTH . ") NOT NULL DEFAULT '',
                 _document VARCHAR(255) NOT NULL,
+                _documentInternalId BIGINT UNSIGNED NOT NULL DEFAULT 0,
                 PRIMARY KEY (_id),
         ";
 
         if ($this->sharedTables) {
             $permissions .= "
                 _tenant INT(11) UNSIGNED DEFAULT NULL,
-                UNIQUE INDEX _index1 (_document, _tenant, _type, _permission),
+                UNIQUE INDEX " . static::PERMISSIONS_INDEX . " (_document, _tenant, _type, _permission, _column),
+                INDEX " . static::PERMISSIONS_INDEX_DOCUMENT . " (_documentInternalId, _tenant, _type, _permission, _column),
                 INDEX _permission (_tenant, _permission, _type)
             ";
         } else {
             $permissions .= "
-                UNIQUE INDEX _index1 (_document, _type, _permission),
+                UNIQUE INDEX " . static::PERMISSIONS_INDEX . " (_document, _type, _permission, _column),
+                INDEX " . static::PERMISSIONS_INDEX_DOCUMENT . " (_documentInternalId, _type, _permission, _column),
                 INDEX _permission (_permission, _type)
             ";
         }
@@ -894,13 +911,22 @@ public function createDocument(Document $collection, Document $document): Docume
                 $attributeIndex++;
             }
 
+            // _column is always named. Every permissions table carries it -- new ones
+            // from CREATE TABLE, older ones from the column-permissions migration -- so
+            // there is nothing to make it conditional on. Writing it unconditionally also
+            // keeps _perms in step with the _permissions JSON on the row: a grant stored
+            // as read("role", "salary") lands as _column = 'salary' whatever the flag
+            // says, so the query gate and masking can never disagree about it.
             $permissions = [];
+            $permissionBinds = [];
             foreach (Database::PERMISSIONS as $type) {
-                foreach ($document->getPermissionsByType($type) as $permission) {
+                foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
                     $tenantBind = $this->sharedTables ? ", :_tenant" : '';
-                    $permission = \str_replace('"', '', $permission);
-                    $permission = "('{$type}', '{$permission}', :_uid {$tenantBind})";
-                    $permissions[] = $permission;
+                    $role = \str_replace('"', '', $permission['role']);
+
+                    $columnBind = ":_column_{$type}_{$i}";
+                    $permissionBinds[$columnBind] = $permission['column'];
+                    $permissions[] = "('{$type}', '{$role}', {$columnBind}, :_uid {$tenantBind})";
                 }
             }
 
@@ -909,7 +935,7 @@ public function createDocument(Document $collection, Document $document): Docume
                 $permissions = \implode(', ', $permissions);
 
                 $sqlPermissions = "
-                    INSERT INTO {$this->getSQLTable($name . '_perms')} (_type, _permission, _document {$tenantColumn})
+                    INSERT INTO {$this->getSQLTable($name . '_perms')} (_type, _permission, _column, _document {$tenantColumn})
                     VALUES {$permissions};
                 ";
 
@@ -918,6 +944,9 @@ public function createDocument(Document $collection, Document $document): Docume
                 if ($this->sharedTables) {
                     $stmtPermissions->bindValue(':_tenant', $document->getTenant());
                 }
+                foreach ($permissionBinds as $key => $value) {
+                    $stmtPermissions->bindValue($key, $value);
+                }
             }
 
             $stmt->execute();
@@ -932,10 +961,14 @@ public function createDocument(Document $collection, Document $document): Docume
                 try {
                     $stmtPermissions->execute();
                 } catch (PDOException $e) {
+                    // Compare the violated key exactly rather than searching the
+                    // message for a substring: the index names are contained in
+                    // plenty of other index names, and misreading one would run the
+                    // cleanup below against permissions that were never orphaned.
                     $isOrphanedPermission = $e->getCode() === '23000'
                         && isset($e->errorInfo[1])
                         && $e->errorInfo[1] === 1062
-                        && \str_contains($e->getMessage(), '_index1');
+                        && $this->isPermissionsIndex($this->getViolatedKey($e->getMessage()));
 
                     if (!$isOrphanedPermission) {
                         throw $e;
@@ -1007,18 +1040,21 @@ public function updateDocument(Document $collection, string $id, Document $docum
                 $values = [];
                 $binds = [];
                 foreach (Database::PERMISSIONS as $type) {
-                    foreach ($document->getPermissionsByType($type) as $i => $permission) {
+                    foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
                         $tenantPlaceholder = $this->sharedTables ? ', :_tenant' : '';
-                        $values[] = "( :_uid, '{$type}', :_add_{$type}_{$i} {$tenantPlaceholder})";
-                        $binds[":_add_{$type}_{$i}"] = $permission;
+
+                        $values[] = "( :_uid, '{$type}', :_add_{$type}_{$i}, :_addcol_{$type}_{$i} {$tenantPlaceholder})";
+                        $binds[":_addcol_{$type}_{$i}"] = $permission['column'];
+
+                        $binds[":_add_{$type}_{$i}"] = $permission['role'];
                     }
                 }
 
                 if (!empty($values)) {
                     $tenantColumn = $this->sharedTables ? ', _tenant' : '';
 
                     $sql = "
-				    INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission {$tenantColumn})
+				    INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission, _column {$tenantColumn})
 				    VALUES " . \implode(', ', $values);
 
                     $sql = $this->trigger(Database::EVENT_PERMISSIONS_CREATE, $sql);
@@ -1771,6 +1807,37 @@ public function getSupportForUpsertOnUniqueIndex(): bool
         return true;
     }
 
+    /**
+     * _permissions is MEDIUMTEXT under the table's utf8mb4 collation, which is case
+     * insensitive, so a plain comparison would treat read("user:hr") and
+     * read("user:HR") as the same permissions. Casting the bound value forces a
+     * byte-for-byte comparison, which is what the compare-and-set needs.
+     */
+    protected function getJsonBind(string $placeholder): string
+    {
+        return "CAST({$placeholder} AS BINARY)";
+    }
+
+    public function getSupportForColumnPermissions(): bool
+    {
+        return true;
+    }
+
+    /**
+     * Is this the unique index on a permissions table, under either name?
+     *
+     * A duplicate-key error has to be recognised on tables the column-permissions
+     * migration has reached and on ones it has not, so both spellings count.
+     *
+     * @param string|null $key
+     * @return bool
+     */
+    protected function isPermissionsIndex(?string $key): bool
+    {
+        return $key === static::PERMISSIONS_INDEX || $key === static::PERMISSIONS_INDEX_LEGACY;
+    }
+
+
     public function getSupportForSchemaAttributes(): bool
     {
         return true;
@@ -1896,7 +1963,7 @@ protected function processException(PDOException $e): \Exception
         // Duplicate row
         if ($e->getCode() === '23000' && isset($e->errorInfo[1]) && $e->errorInfo[1] === 1062) {
             $key = $this->getViolatedKey($e->getMessage());
-            if ($key === '_index1') {
+            if ($this->isPermissionsIndex($key)) {
                 return new DuplicateException('Duplicate permissions for document', $e->getCode(), $e);
             }
             if ($key !== null && $key !== '_uid' && $key !== 'PRIMARY') {