fix(security): keep caller-named fields off internal columns and operators

src/Database/Adapter.php

@@ -994,6 +994,24 @@ public function filter(string $value): string
         return self::$filteredKeyCache[$value] = $filtered;
     }
 
+    /**
+     * The column an attribute a caller names is written to. filter() keeps underscores, so a name such as `_tenant`
+     * or `_ten.ant` would reach a column the storage keeps for the document itself and move the document to another
+     * tenant, or rewrite its id or timestamps.
+     *
+     * @throws NotFoundException When the attribute names a reserved column
+     */
+    protected function writableColumn(string $attribute): string
+    {
+        $column = $this->filter($attribute);
+
+        if (Storage::isReserved($column)) {
+            throw new NotFoundException('Attribute not found');
+        }
+
+        return $column;
+    }
+
     /**
      * The row as every write hook decorates a row written for the document.
      *

src/Database/Adapter/Mongo.php

@@ -1616,9 +1616,7 @@ public function createDocument(Document $collection, Document $document): Docume
 
         $document->removeAttribute(Document::SEQUENCE);
 
-        /** @var array<string, mixed> $documentArray */
-        $documentArray = (array) $document;
-        $record = $this->replaceCharacters('$', '_', $documentArray);
+        $record = $this->replaceCharacters('$', '_', (array) $this->withoutStorageAttributes($document));
         $record = $this->decorateRow($record, $document);
 
         if (! empty($sequence)) {
@@ -1663,9 +1661,7 @@ public function createDocuments(Document $collection, array $documents): array
                 throw new DatabaseException('All documents must have an sequence if one is set');
             }
 
-            /** @var array<string, mixed> $documentArr */
-            $documentArr = (array) $document;
-            $record = $this->replaceCharacters('$', '_', $documentArr);
+            $record = $this->replaceCharacters('$', '_', (array) $this->withoutStorageAttributes($document));
             $record = $this->decorateRow($record, $document);
 
             if (! empty($sequence)) {
@@ -1757,8 +1753,7 @@ public function updateDocument(Document $collection, string $id, Document $docum
     {
         $name = $this->getNamespace().'_'.$this->filter($collection->getId());
 
-        $record = $document->getArrayCopy();
-        $record = $this->replaceCharacters('$', '_', $record);
+        $record = $this->replaceCharacters('$', '_', $this->withoutStorageKeys($document->getArrayCopy()));
 
         $filters = [Storage::UID => $id];
         $filters = $this->applyTenantFilter($filters, $collection->getId());
@@ -1804,8 +1799,7 @@ public function updateDocuments(Document $collection, Document $updates, array $
         $filters = $this->buildFilters($queries);
         $filters = $this->applyTenantFilter($filters, $collection->getId());
 
-        $record = $updates->getArrayCopy();
-        $record = $this->replaceCharacters('$', '_', $record);
+        $record = $this->replaceCharacters('$', '_', $this->withoutStorageKeys($updates->getArrayCopy()));
 
         try {
             $pipeline = $this->buildOperatorPipeline($record);
@@ -2167,15 +2161,14 @@ public function upsertDocuments(Document $collection, array $changes, ?string $i
 
         try {
             $name = $this->getNamespace().'_'.$this->filter($collection->getId());
-            $attribute = $this->filter($increase ?? '');
+            $attribute = $this->writableColumn($increase ?? '');
 
             $operations = [];
             $hasPipeline = false;
             foreach ($changes as $change) {
                 $document = $change->new;
                 $oldDocument = $change->old;
-                /** @var array<string, mixed> $attributes */
-                $attributes = $document->getAttributes();
+                $attributes = $this->withoutStorageKeys($document->getAttributes());
                 $attributes[Storage::UID] = $document->getId();
                 $attributes[Storage::CREATED_AT] = $document[Document::CREATED_AT];
                 $attributes[Storage::UPDATED_AT] = $document[Document::UPDATED_AT];
@@ -2343,7 +2336,7 @@ public function increaseDocumentAttribute(Document $collection, string $id, stri
         $min = $min === null ? null : $this->normalizeAtomicNumber($min, 'minimum');
         $max = $max === null ? null : $this->normalizeAtomicNumber($max, 'maximum');
 
-        $attribute = $this->filter($attribute);
+        $attribute = $this->writableColumn($attribute);
         $current = ['$ifNull' => ['$'.$attribute, 0]];
         $filters = [Storage::UID => $id];
         $filters = $this->applyTenantFilter($filters, $collectionId);
@@ -2438,8 +2431,7 @@ public function find(Document $collection, array $queries = [], ?int $limit = 25
         $sortOptions = [];
 
         foreach ($orderAttributes as $i => $originalAttribute) {
-            $attribute = $this->getInternalKeyForAttribute($originalAttribute);
-            $attribute = $this->filter($attribute);
+            $attribute = $this->orderField($originalAttribute);
 
             $orderType = $orderTypes[$i] ?? OrderDirection::Asc;
             $direction = $orderType;
@@ -2464,10 +2456,11 @@ public function find(Document $collection, array $queries = [], ?int $limit = 25
                 $andConditions = [];
                 for ($j = 0; $j < $i; $j++) {
                     $originalPrev = $orderAttributes[$j];
-                    $prevAttr = $this->filter($this->getInternalKeyForAttribute($originalPrev));
+                    $prevAttr = $this->orderField($originalPrev);
                     $tmp = $cursor[$originalPrev];
+                    // $eq, so a stored value shaped like {"$gt": ""} is compared, not run as an operator
                     $andConditions[] = [
-                        $prevAttr => $tmp,
+                        $prevAttr => ['$eq' => $tmp],
                     ];
                 }
 
@@ -3277,6 +3270,42 @@ public function castDatetime(string $value): mixed
         return new UTCDateTime(new NativeDateTime($value));
     }
 
+    /**
+     * A document's fields without those stored over a storage field. A schemaless document holds any key, and one
+     * such as `_id`, `_tenant` or `_permissions`, or `$uid`, which the `$` to `_` swap stores as `_uid`, would
+     * otherwise be written over the field the storage keeps for the document's own $id, $sequence, $tenant or
+     * $permissions.
+     *
+     * @param  array<mixed>  $fields
+     * @return array<mixed>
+     */
+    private function withoutStorageKeys(array $fields): array
+    {
+        return \array_filter($fields, $this->isWritableField(...), \ARRAY_FILTER_USE_KEY);
+    }
+
+    private function isWritableField(int|string $key): bool
+    {
+        if (isset(Storage::attributeMap()[$key])) {
+            return true;
+        }
+
+        $stored = \array_key_first($this->replaceCharacters('$', '_', [$key => null]));
+
+        return ! Storage::isReserved((string) $stored);
+    }
+
+    private function withoutStorageAttributes(Document $document): Document
+    {
+        foreach (\array_keys($document->getArrayCopy()) as $key) {
+            if (! $this->isWritableField($key)) {
+                $document->removeAttribute((string) $key);
+            }
+        }
+
+        return $document;
+    }
+
     /**
      * MongoDB field names cannot start with $ or contain dots.
      */
@@ -3500,7 +3529,7 @@ protected function buildFilters(array $queries, string $separator = '$and'): arr
                     /** @var array<Query> $elemMatchValues */
                     $elemMatchValues = $query->getValues();
                     $filters[$separator][] = [
-                        $query->getAttribute() => [
+                        $this->filterField($query->getAttribute()) => [
                             '$elemMatch' => $this->buildFilters($elemMatchValues, $separator),
                         ],
                     ];
@@ -3544,22 +3573,7 @@ protected function buildFilter(Query $query): array
             $query->setValues($values);
         }
 
-        if ($query->getAttribute() === Document::ID) {
-            $query->setAttribute(Storage::UID);
-        } elseif ($query->getAttribute() === Document::SEQUENCE) {
-            $query->setAttribute(Storage::SEQUENCE);
-            $values = $query->getValues();
-            foreach ($values as $k => $v) {
-                $values[$k] = $v;
-            }
-            $query->setValues($values);
-        } elseif ($query->getAttribute() === Document::CREATED_AT) {
-            $query->setAttribute(Storage::CREATED_AT);
-        } elseif ($query->getAttribute() === Document::UPDATED_AT) {
-            $query->setAttribute(Storage::UPDATED_AT);
-        } elseif (\str_starts_with($query->getAttribute(), '$')) {
-            $query->setAttribute($this->escapeMongoFieldName($query->getAttribute()));
-        }
+        $query->setAttribute($this->filterField($query->getAttribute()));
 
         $attribute = $query->getAttribute();
         $operator = $this->getQueryOperator($query->getMethod());
@@ -3678,7 +3692,7 @@ protected function buildFilter(Query $query): array
             $existsOr = \is_array($filter['$or'] ?? null) ? $filter['$or'] : [];
             foreach ($query->getValues() as $existsAttribute) {
                 $existsAttrStr = \is_string($existsAttribute) ? $existsAttribute : (\is_scalar($existsAttribute) ? (string) $existsAttribute : '');
-                $existsOr[] = [$existsAttrStr => [$operator => $value]];
+                $existsOr[] = [$this->filterField($existsAttrStr) => [$operator => $value]];
             }
             $filter['$or'] = $existsOr;
         } else {
@@ -3691,6 +3705,33 @@ protected function buildFilter(Query $query): array
         return $filter;
     }
 
+    /**
+     * The stored field a filter names: $id, $sequence and the timestamps by their storage fields, and any other name
+     * starting with $ escaped, so it is matched as a field and never read as a query operator such as $where or
+     * $comment.
+     */
+    private function filterField(string $attribute): string
+    {
+        return match ($attribute) {
+            Document::ID => Storage::UID,
+            Document::SEQUENCE => Storage::SEQUENCE,
+            Document::CREATED_AT => Storage::CREATED_AT,
+            Document::UPDATED_AT => Storage::UPDATED_AT,
+            default => \str_starts_with($attribute, '$') ? $this->escapeMongoFieldName($attribute) : $attribute,
+        };
+    }
+
+    private function orderField(string $attribute): string
+    {
+        return $this->filter($this->filterField($attribute));
+    }
+
+    #[\Override]
+    protected function writableColumn(string $attribute): string
+    {
+        return parent::writableColumn($this->filterField($attribute));
+    }
+
     /**
      * @throws Exception
      */
@@ -3800,7 +3841,7 @@ private function getAttributeProjection(array $selections): array
                 continue;
             }
 
-            $projection[$selection] = 1;
+            $projection[$this->filterField($selection)] = 1;
         }
 
         $projection[Storage::UID] = 1;

src/Database/Adapter/SQL.php

@@ -1492,7 +1492,7 @@ public function increaseDocumentAttribute(
         int|float|string|null $max = null
     ): bool {
         $name = $this->filter($collection->getId());
-        $attribute = $this->filter($attribute);
+        $attribute = $this->writableColumn($attribute);
 
         $builder = $this->newBuilder($name);
         $builder->setRaw($attribute, 'COALESCE('.$this->quote($attribute).', 0) + ?', [$value]);
@@ -4007,7 +4007,7 @@ protected function executeUpsertBatch(
         $skipColumns = [Storage::UID, Storage::SEQUENCE, Storage::CREATED_AT, Storage::TENANT];
 
         if (! empty($attribute)) {
-            $updateColumns = [$this->filter($attribute), Storage::UPDATED_AT];
+            $updateColumns = [$this->writableColumn($attribute), Storage::UPDATED_AT];
         } else {
             $updateColumns = \array_values(\array_filter(
                 $allColumnNames,
@@ -4019,7 +4019,7 @@ protected function executeUpsertBatch(
 
         // conflictSetRaw() takes the column names given to onConflict(); the expression methods quote their own.
         if (! empty($attribute)) {
-            $incrementColumn = $this->filter($attribute);
+            $incrementColumn = $this->writableColumn($attribute);
             if ($this->sharedTables) {
                 $builder->conflictSetRaw($incrementColumn, $this->getConflictTenantIncrementExpression($incrementColumn));
                 $builder->conflictSetRaw(Storage::UPDATED_AT, $this->getConflictTenantExpression(Storage::UPDATED_AT));

src/Database/Adapter/SQLite.php

@@ -73,6 +73,9 @@ class SQLite extends SQL
     /** Suffix appended to every FTS5 virtual table name created by this adapter. */
     private const string FTS_TABLE_SUFFIX = '_fts';
 
+    /** Length of the hashed attribute set between the FTS5 table prefix and suffix. */
+    private const int FTS_TABLE_KEY_LENGTH = 16;
+
     /** AFTER INSERT trigger suffix on the parent collection. */
     private const string FTS_TRIGGER_INSERT = 'ai';
 
@@ -669,7 +672,7 @@ public function deleteCollection(string $collection): bool
 
         // FTS5 shadow tables don't drop with the parent.
         foreach ($this->findFulltextTables($id) as $ftsTable) {
-            $sql = "DROP TABLE IF EXISTS `{$ftsTable}`";
+            $sql = 'DROP TABLE IF EXISTS '.$this->quote($ftsTable);
             $this->execute($this->prepare($sql, event: Event::CollectionDelete));
         }
 
@@ -935,7 +938,7 @@ protected function getFulltextTableName(string $collection, array|string $attrib
     {
         $names = \array_map($this->filter(...), \is_array($attributes) ? $attributes : [$attributes]);
         \sort($names);
-        $key = \substr(\hash('sha1', \implode("\0", $names)), 0, 16);
+        $key = \substr(\hash('sha1', \implode("\0", $names)), 0, self::FTS_TABLE_KEY_LENGTH);
 
         return $this->getFulltextTablePrefix($collection) . $key . self::FTS_TABLE_SUFFIX;
     }
@@ -1105,9 +1108,9 @@ protected function dropFulltextIndexById(string $collection, string $id, Event $
         $this->startTransaction();
         try {
             foreach ($triggerSuffixes as $suffix) {
-                $this->execute($this->prepare("DROP TRIGGER IF EXISTS `{$ftsTable}_{$suffix}`", event: $event));
+                $this->execute($this->prepare('DROP TRIGGER IF EXISTS '.$this->quote($ftsTable.'_'.$suffix), event: $event));
             }
-            $sql = "DROP TABLE IF EXISTS `{$ftsTable}`";
+            $sql = 'DROP TABLE IF EXISTS '.$this->quote($ftsTable);
             $this->execute($this->prepare($sql, event: $event));
             $this->commitTransaction();
         } catch (\Throwable $e) {
@@ -1168,14 +1171,19 @@ protected function findFulltextTables(string $collection): array
     {
         // ESCAPE '\\' so the literal `_` separators in the prefix don't
         // act as LIKE wildcards (e.g. `db_users_` matching `db_usersA_`).
+        // One unescaped `_` per character of the hashed key, never `%`: the
+        // prefix of collection `a` also starts the tables of collection
+        // `a_b`, which a search on `a` would otherwise read and a delete of
+        // `a` would otherwise drop.
         $statement = $this->prepare("
             SELECT name FROM sqlite_master
             WHERE type='table'
-              AND name LIKE :_prefix ESCAPE '\\'
-              AND name LIKE :_suffix ESCAPE '\\'
+              AND name LIKE :_name ESCAPE '\\'
+              AND sql LIKE 'CREATE VIRTUAL TABLE % USING fts5(%'
         ");
-        $statement->bindValue(':_prefix', $this->escapeLikePattern($this->getFulltextTablePrefix($collection)) . '%');
-        $statement->bindValue(':_suffix', '%' . $this->escapeLikePattern(self::FTS_TABLE_SUFFIX));
+        $statement->bindValue(':_name', $this->escapeLikePattern($this->getFulltextTablePrefix($collection))
+            . \str_repeat('_', self::FTS_TABLE_KEY_LENGTH)
+            . $this->escapeLikePattern(self::FTS_TABLE_SUFFIX));
         $statement->execute();
         $tables = $statement->fetchAll(PDO::FETCH_COLUMN);
         $statement->closeCursor();
@@ -1697,7 +1705,8 @@ protected function compileAdapterFilter(Query $query, string $collection, string
             );
         }
 
-        $subquery = "{$quotedAlias}.{$this->quote(Storage::SEQUENCE)} IN (SELECT rowid FROM `{$ftsTable}` WHERE `{$ftsTable}` MATCH ?)";
+        $quotedTable = $this->quote($ftsTable);
+        $subquery = "{$quotedAlias}.{$this->quote(Storage::SEQUENCE)} IN (SELECT rowid FROM {$quotedTable} WHERE {$quotedTable} MATCH ?)";
 
         return new Expression(
             $method === Method::Search ? $subquery : "NOT ({$subquery})",
@@ -2376,7 +2385,7 @@ protected function executeUpsertBatch(
 
         if (! empty($attribute)) {
             $updateColumns = [
-                $getUpdateClause($attribute, increment: true),
+                $getUpdateClause($this->writableColumn($attribute), increment: true),
                 $getUpdateClause(Storage::UPDATED_AT),
             ];
         } else {
@@ -2700,7 +2709,7 @@ public function getSchemaIndexes(string $collection): array
      */
     private function getIndexColumns(string $index): array
     {
-        $statement = $this->prepare("PRAGMA index_info(`{$index}`)", event: Event::CollectionRead);
+        $statement = $this->prepare('PRAGMA index_info('.$this->quote($index).')', event: Event::CollectionRead);
         $this->execute($statement);
         $rows = $statement->fetchAll();
         $statement->closeCursor();
@@ -2735,7 +2744,7 @@ protected function getFulltextSchemaIndexes(string $collection): array
 
         $indexes = [];
         foreach ($tables as $fulltextTable) {
-            $statement = $this->prepare("PRAGMA table_info(`{$fulltextTable}`)");
+            $statement = $this->prepare('PRAGMA table_info('.$this->quote($fulltextTable).')');
             $statement->execute();
             $rows = $statement->fetchAll(PDO::FETCH_ASSOC);
             $statement->closeCursor();
@@ -2830,7 +2839,7 @@ private function buildFulltextAttributeMap(string $collection): array
     {
         $map = [];
         foreach ($this->findFulltextTables($collection) as $table) {
-            $info = $this->prepare("PRAGMA table_info(`{$table}`)");
+            $info = $this->prepare('PRAGMA table_info('.$this->quote($table).')');
             $info->execute();
             $cols = $info->fetchAll(PDO::FETCH_ASSOC);
             $info->closeCursor();

src/Database/Builder/Postgres.php

@@ -64,7 +64,7 @@ public function compileFilter(Query $query): string
             }
         }
 
-        return parent::compileFilter($query);
+        return $this->enclose($query, parent::compileFilter($query));
     }
 
     /**