Phase 0: fixes from review (torn files, pending blobs, exact sequences, UTF-8, storage checks)
- detect files changing during reading or between resumable steps and re-read them (size/mtime/ctime/inode), spike F overwrites mid-run - drop pending blob ids when a pack or index upload fails - dump and restore PostgreSQL sequence state exactly, bound parameters, validated catalog names; spike C checks all sequences exactly - keep invalid UTF-8 byte-exact (tagged base64) - refuse large tables without a primary key - StorageBackend: check write/flush/close results, validate base path - user_migration import/export: root paths, intermediate folders, partial writes, read errors - spikes: never touch an existing account, own fixtures, 512M note
This commit is contained in:
@@ -18,30 +18,40 @@ final class LocalBackend implements IBackend {
|
||||
}
|
||||
|
||||
public function put(string $path, $stream): void {
|
||||
$target = $this->abs($path);
|
||||
$dir = dirname($target);
|
||||
if (!is_dir($dir) && !mkdir($dir, 0700, true) && !is_dir($dir)) {
|
||||
throw new BackendException('Cannot create ' . $dir);
|
||||
}
|
||||
// Write to a temporary name and rename, so a crash never leaves a partial object.
|
||||
$tmp = $target . '.part-' . bin2hex(random_bytes(4));
|
||||
$out = fopen($tmp, 'wb');
|
||||
if ($out === false) {
|
||||
throw new BackendException('Cannot write ' . $path);
|
||||
}
|
||||
$tmp = null;
|
||||
$out = false;
|
||||
$done = false;
|
||||
try {
|
||||
$target = $this->abs($path);
|
||||
$dir = dirname($target);
|
||||
if (!is_dir($dir) && !mkdir($dir, 0700, true) && !is_dir($dir)) {
|
||||
throw new BackendException('Cannot create ' . $dir);
|
||||
}
|
||||
// Write to a temporary name and rename, so a crash never leaves a partial object.
|
||||
$tmp = $target . '.part-' . bin2hex(random_bytes(4));
|
||||
$out = fopen($tmp, 'wb');
|
||||
if ($out === false) {
|
||||
throw new BackendException('Cannot write ' . $path);
|
||||
}
|
||||
if (stream_copy_to_stream($stream, $out) === false || !fflush($out)) {
|
||||
throw new BackendException('Write failed for ' . $path);
|
||||
}
|
||||
} finally {
|
||||
fclose($out);
|
||||
$out = false;
|
||||
if (!rename($tmp, $target)) {
|
||||
throw new BackendException('Cannot finalise ' . $path);
|
||||
}
|
||||
$done = true;
|
||||
} finally {
|
||||
if ($out !== false) {
|
||||
fclose($out);
|
||||
}
|
||||
if (is_resource($stream)) {
|
||||
fclose($stream);
|
||||
}
|
||||
}
|
||||
if (!rename($tmp, $target)) {
|
||||
@unlink($tmp);
|
||||
throw new BackendException('Cannot finalise ' . $path);
|
||||
if (!$done && $tmp !== null) {
|
||||
@unlink($tmp);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -23,6 +23,7 @@ final class StorageBackend implements IBackend {
|
||||
private string $base = '',
|
||||
) {
|
||||
$this->base = trim($base, '/');
|
||||
self::validate($this->base, true);
|
||||
$this->atomicPut = $storage->instanceOfStorage(\OCA\Files_External\Lib\Storage\AmazonS3::class);
|
||||
if ($this->base !== '') {
|
||||
$this->mkdirs($this->base);
|
||||
@@ -37,13 +38,18 @@ final class StorageBackend implements IBackend {
|
||||
if ($this->storage->instanceOfStorage(IWriteStreamStorage::class)) {
|
||||
/** @var IWriteStreamStorage $s */
|
||||
$s = $this->storage;
|
||||
$s->writeStream($writeTo, $stream);
|
||||
$written = $s->writeStream($writeTo, $stream);
|
||||
if ($written <= 0 && $this->storage->filesize($writeTo) !== 0) {
|
||||
throw new BackendException('Write reported no data for ' . $path);
|
||||
}
|
||||
} else {
|
||||
$out = $this->storage->fopen($writeTo, 'w');
|
||||
if ($out === false || stream_copy_to_stream($stream, $out) === false) {
|
||||
if ($out === false || stream_copy_to_stream($stream, $out) === false || !fflush($out)) {
|
||||
throw new BackendException('Write failed for ' . $path);
|
||||
}
|
||||
fclose($out);
|
||||
if (!fclose($out)) {
|
||||
throw new BackendException('Close failed for ' . $path);
|
||||
}
|
||||
}
|
||||
if (is_resource($stream)) {
|
||||
fclose($stream);
|
||||
@@ -139,10 +145,23 @@ final class StorageBackend implements IBackend {
|
||||
$this->storage->mkdir($dir);
|
||||
}
|
||||
|
||||
private function abs(string $path): string {
|
||||
if (str_contains($path, '..')) {
|
||||
/** Reject empty segments, '.', '..', NUL, backslashes and absolute paths. */
|
||||
private static function validate(string $path, bool $allowEmpty = false): void {
|
||||
if ($path === '' && $allowEmpty) {
|
||||
return;
|
||||
}
|
||||
if ($path === '' || str_starts_with($path, '/') || str_contains($path, "\\") || str_contains($path, "\0")) {
|
||||
throw new BackendException('Invalid path: ' . $path);
|
||||
}
|
||||
foreach (explode('/', $path) as $segment) {
|
||||
if ($segment === '' || $segment === '.' || $segment === '..') {
|
||||
throw new BackendException('Invalid path: ' . $path);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private function abs(string $path): string {
|
||||
self::validate(ltrim($path, '/'));
|
||||
return ltrim(($this->base === '' ? '' : $this->base . '/') . ltrim($path, '/'), '/');
|
||||
}
|
||||
}
|
||||
|
||||
+69
-11
@@ -31,6 +31,8 @@ use OCA\NgBackup\Repository\Repository;
|
||||
*/
|
||||
final class DbDumper {
|
||||
private const BATCH = 2000;
|
||||
/** Tables without a primary key are read in one query; refuse that above this size. */
|
||||
private const MAX_ROWS_WITHOUT_PK = 50000;
|
||||
|
||||
public function __construct(
|
||||
private Connection $db,
|
||||
@@ -66,11 +68,12 @@ final class DbDumper {
|
||||
}
|
||||
$result[$table] = ['blobs' => $writer->finish(), 'rows' => $rows, 'bytes' => $writer->bytes(), 'sha256' => hash_final($hash)];
|
||||
}
|
||||
$sequences = $this->sequenceStates($tables);
|
||||
} finally {
|
||||
$this->db->rollBack();
|
||||
}
|
||||
$repo->flushPacks();
|
||||
return ['tables' => $result, 'provider' => get_class($this->db->getDatabasePlatform())];
|
||||
return ['tables' => $result, 'sequences' => $sequences, 'provider' => get_class($this->db->getDatabasePlatform())];
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -106,6 +109,9 @@ final class DbDumper {
|
||||
$row = [];
|
||||
foreach ($columns as $i => $col) {
|
||||
$value = $data[$i];
|
||||
if (is_array($value) && isset($value['$b'])) {
|
||||
$value = base64_decode($value['$b'], true); // text that was not valid UTF-8
|
||||
}
|
||||
$row[$col] = ($value !== null && isset($binary[$col])) ? base64_decode($value, true) : $value;
|
||||
}
|
||||
$batch[] = $row;
|
||||
@@ -117,8 +123,8 @@ final class DbDumper {
|
||||
if ($batch !== []) {
|
||||
$total += $this->insertBatch($table, $batch, $binary);
|
||||
}
|
||||
$this->resetSequences($table, $sm->introspectTable($table));
|
||||
}
|
||||
$this->restoreSequences($manifest['sequences'] ?? []);
|
||||
$this->db->commit();
|
||||
} catch (\Throwable $e) {
|
||||
$this->db->rollBack();
|
||||
@@ -158,7 +164,11 @@ final class DbDumper {
|
||||
$select = 'SELECT ' . implode(', ', array_map($q, $columns)) . ' FROM ' . $q($table->getName());
|
||||
|
||||
if ($pk === []) {
|
||||
// Nextcloud requires primary keys; fall back to one query for odd tables.
|
||||
// Nextcloud requires primary keys; small odd tables are read in one query.
|
||||
$count = (int)$this->db->fetchOne('SELECT COUNT(*) FROM ' . $q($table->getName()));
|
||||
if ($count > self::MAX_ROWS_WITHOUT_PK) {
|
||||
throw new \RuntimeException("Table {$table->getName()} has no primary key and $count rows; cannot dump it with bounded memory");
|
||||
}
|
||||
$batches = [$this->db->fetchAllNumeric($select)];
|
||||
} else {
|
||||
$batches = $this->keysetBatches($select, $pk, $columns);
|
||||
@@ -171,12 +181,15 @@ final class DbDumper {
|
||||
}
|
||||
if ($row[$i] !== null && isset($binaryFlip[$col])) {
|
||||
$row[$i] = base64_encode((string)$row[$i]);
|
||||
} elseif (is_string($row[$i]) && !mb_check_encoding($row[$i], 'UTF-8')) {
|
||||
// Keep invalid UTF-8 byte-exact instead of substituting characters.
|
||||
$row[$i] = ['$b' => base64_encode($row[$i])];
|
||||
} elseif (is_int($row[$i]) || is_float($row[$i]) || is_bool($row[$i])) {
|
||||
// Normalise: drivers differ in returning numbers as int or string.
|
||||
$row[$i] = (string)(is_bool($row[$i]) ? (int)$row[$i] : $row[$i]);
|
||||
}
|
||||
}
|
||||
yield 'row' => json_encode($row, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_INVALID_UTF8_SUBSTITUTE | JSON_THROW_ON_ERROR);
|
||||
yield 'row' => json_encode($row, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -219,15 +232,60 @@ final class DbDumper {
|
||||
return count($rows);
|
||||
}
|
||||
|
||||
private function resetSequences(string $table, \Doctrine\DBAL\Schema\Table $schema): void {
|
||||
/**
|
||||
* Exact state of every sequence owned by a dumped column (PostgreSQL), so a restore continues
|
||||
* where the original left off, not at MAX(id)+1.
|
||||
*
|
||||
* @return array<string, array{table:string, column:string, last:int|null, called:bool}>
|
||||
*/
|
||||
private function sequenceStates(array $tables): array {
|
||||
if (!$this->db->getDatabasePlatform() instanceof \Doctrine\DBAL\Platforms\PostgreSQLPlatform) {
|
||||
return; // MySQL/SQLite derive AUTO_INCREMENT from the data
|
||||
return []; // MySQL/SQLite derive AUTO_INCREMENT from the data
|
||||
}
|
||||
foreach ($schema->getColumns() as $column) {
|
||||
if ($column->getAutoincrement()) {
|
||||
$this->db->executeStatement(sprintf(
|
||||
"SELECT setval(pg_get_serial_sequence('%s', '%s'), COALESCE((SELECT MAX(%s) FROM %s), 0) + 1, false)",
|
||||
$table, $column->getName(), $this->db->quoteIdentifier($column->getName()), $this->db->quoteIdentifier($table)));
|
||||
$rows = $this->db->fetchAllAssociative(
|
||||
"SELECT c.table_name, c.column_name, pg_get_serial_sequence(quote_ident(c.table_name), c.column_name) AS seq
|
||||
FROM information_schema.columns c
|
||||
WHERE c.table_schema = current_schema() AND pg_get_serial_sequence(quote_ident(c.table_name), c.column_name) IS NOT NULL");
|
||||
$wanted = array_flip($tables);
|
||||
$result = [];
|
||||
foreach ($rows as $r) {
|
||||
if (!isset($wanted[$r['table_name']])) {
|
||||
continue;
|
||||
}
|
||||
$state = $this->db->fetchAssociative('SELECT last_value, is_called FROM ' . self::sequenceIdentifier($r['seq']));
|
||||
$result[$r['seq']] = ['table' => $r['table_name'], 'column' => $r['column_name'],
|
||||
// A never-used sequence reports last_value = start value with is_called = false.
|
||||
'last' => $state ? (int)$state['last_value'] : null,
|
||||
'called' => (bool)($state['is_called'] ?? false)];
|
||||
}
|
||||
return $result;
|
||||
}
|
||||
|
||||
/** pg_get_serial_sequence() returns a catalog name like public.oc_x_id_seq or "My"."seq"; refuse anything else. */
|
||||
private static function sequenceIdentifier(string $name): string {
|
||||
if (!preg_match('/^("[^"]+"|[a-z_][a-z0-9_$]*)\.("[^"]+"|[a-z_][a-z0-9_$]*)$/i', $name)) {
|
||||
throw new \RuntimeException('Unexpected sequence name: ' . $name);
|
||||
}
|
||||
return $name;
|
||||
}
|
||||
|
||||
private function restoreSequences(array $sequences): void {
|
||||
if (!$this->db->getDatabasePlatform() instanceof \Doctrine\DBAL\Platforms\PostgreSQLPlatform) {
|
||||
return;
|
||||
}
|
||||
foreach ($sequences as $info) {
|
||||
$seq = $this->db->fetchOne('SELECT pg_get_serial_sequence(quote_ident(?), ?)', [$info['table'], $info['column']]);
|
||||
if ($seq === false || $seq === null) {
|
||||
continue;
|
||||
}
|
||||
if ($info['last'] === null) {
|
||||
// Never used: start at 1 (or after the data, if rows were inserted with explicit ids).
|
||||
$max = (int)$this->db->fetchOne(sprintf('SELECT COALESCE(MAX(%s), 0) FROM %s',
|
||||
$this->db->quoteIdentifier($info['column']), $this->db->quoteIdentifier($info['table'])));
|
||||
$this->db->executeQuery('SELECT setval(CAST(? AS regclass), ?, false)', [$seq, $max + 1]);
|
||||
} else {
|
||||
$this->db->executeQuery('SELECT setval(CAST(? AS regclass), ?, ?)', [$seq, $info['last'], $info['called']],
|
||||
[\Doctrine\DBAL\ParameterType::STRING, \Doctrine\DBAL\ParameterType::INTEGER, \Doctrine\DBAL\ParameterType::BOOLEAN]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+18
-1
@@ -73,7 +73,15 @@ final class BackupRun {
|
||||
continue;
|
||||
}
|
||||
|
||||
$cur = $state['cur'] ?? ['path' => $relative, 'offset' => 0, 'blobs' => []];
|
||||
$version = Repository::fileVersion($path);
|
||||
$cur = $state['cur'];
|
||||
if ($cur === null || $cur['version'] !== $version) {
|
||||
// New file, or it changed since the previous step: (re)start reading it.
|
||||
if ($cur !== null) {
|
||||
$stats['retried'] = ($stats['retried'] ?? 0) + 1;
|
||||
}
|
||||
$cur = ['path' => $relative, 'offset' => 0, 'blobs' => [], 'version' => $version];
|
||||
}
|
||||
$fh = fopen($path, 'rb');
|
||||
fseek($fh, $cur['offset']);
|
||||
$finished = true;
|
||||
@@ -86,6 +94,15 @@ final class BackupRun {
|
||||
}
|
||||
}
|
||||
fclose($fh);
|
||||
if (Repository::fileVersion($path) !== $cur['version']) {
|
||||
// Changed while reading: start this file again (in this or the next step).
|
||||
$stats['retried'] = ($stats['retried'] ?? 0) + 1;
|
||||
$state['cur'] = null;
|
||||
if (microtime(true) >= $deadline) {
|
||||
break;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (!$finished) {
|
||||
$state['cur'] = $cur;
|
||||
break;
|
||||
|
||||
@@ -55,6 +55,13 @@ final class BlobIndex {
|
||||
$this->pending[$id] = true;
|
||||
}
|
||||
|
||||
/** @param list<string> $ids */
|
||||
public function dropPending(array $ids): void {
|
||||
foreach ($ids as $id) {
|
||||
unset($this->pending[$id]);
|
||||
}
|
||||
}
|
||||
|
||||
/** @param list<array{id:string, offset:int, length:int, raw:int, flags:int}> $entries */
|
||||
public function addPack(string $packId, array $entries): void {
|
||||
$stream = fopen('php://memory', 'w+b');
|
||||
|
||||
@@ -59,9 +59,17 @@ final class PackWriter {
|
||||
fwrite($this->buffer, $header . pack('N', strlen($header)));
|
||||
rewind($this->buffer);
|
||||
$path = 'packs/' . substr($packId, 0, 2) . '/' . $packId;
|
||||
$this->backend->put($path, $this->buffer); // closes the buffer
|
||||
unset($this->buffer);
|
||||
$this->index->addPack($packId, $this->entries);
|
||||
try {
|
||||
$this->backend->put($path, $this->buffer); // closes the buffer
|
||||
unset($this->buffer);
|
||||
$this->index->addPack($packId, $this->entries);
|
||||
} catch (\Throwable $e) {
|
||||
// The blobs of this pack are not durable: forget them, so a retry stores them again.
|
||||
$this->index->dropPending(array_column($this->entries, 'id'));
|
||||
unset($this->buffer);
|
||||
$this->reset();
|
||||
throw $e;
|
||||
}
|
||||
$this->reset();
|
||||
}
|
||||
|
||||
|
||||
@@ -92,12 +92,7 @@ final class Repository {
|
||||
$blobs = $prev['b'];
|
||||
$stats['reused']++;
|
||||
} else {
|
||||
$blobs = [];
|
||||
$fh = fopen($info->getPathname(), 'rb');
|
||||
while (($data = self::readFull($fh, self::BLOB_SIZE)) !== '') {
|
||||
$blobs[] = $this->storeBlob($data, $packs, $stats);
|
||||
}
|
||||
fclose($fh);
|
||||
$blobs = $this->readStable($info->getPathname(), $packs, $stats);
|
||||
$stats['read']++;
|
||||
}
|
||||
fwrite($tree, json_encode(['p' => $relative, 's' => $info->getSize(), 'm' => $info->getMTime(), 'b' => $blobs], JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR) . "\n");
|
||||
@@ -244,6 +239,36 @@ final class Repository {
|
||||
return $this->index->count();
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a file into blobs and make sure it did not change while reading (size, mtime, ctime
|
||||
* before and after). A file that keeps changing is an error rather than a torn backup.
|
||||
*
|
||||
* @return list<string>
|
||||
*/
|
||||
private function readStable(string $path, PackWriter $packs, array &$stats): array {
|
||||
for ($attempt = 1; $attempt <= 3; $attempt++) {
|
||||
$before = self::fileVersion($path);
|
||||
$blobs = [];
|
||||
$fh = fopen($path, 'rb');
|
||||
while (($data = self::readFull($fh, self::BLOB_SIZE)) !== '') {
|
||||
$blobs[] = $this->storeBlob($data, $packs, $stats);
|
||||
}
|
||||
fclose($fh);
|
||||
if (self::fileVersion($path) === $before) {
|
||||
return $blobs;
|
||||
}
|
||||
$stats['retried'] = ($stats['retried'] ?? 0) + 1;
|
||||
}
|
||||
throw new RepositoryException('File keeps changing while being read: ' . $path);
|
||||
}
|
||||
|
||||
/** Cheap file identity: changes when the content may have changed. */
|
||||
public static function fileVersion(string $path): string {
|
||||
clearstatcache(true, $path);
|
||||
$st = @stat($path);
|
||||
return $st === false ? 'missing' : implode(':', [$st['size'], $st['mtime'], $st['ctime'], $st['ino']]);
|
||||
}
|
||||
|
||||
private function storeBlob(string $data, PackWriter $packs, array &$stats): string {
|
||||
$id = $this->keys->blobId($data);
|
||||
if ($this->index->has($id)) {
|
||||
|
||||
@@ -11,6 +11,7 @@ use OCA\NgBackup\Repository\Repository;
|
||||
use OCP\Files\File;
|
||||
use OCP\Files\Folder;
|
||||
use OCP\UserMigration\IExportDestination;
|
||||
use OCP\UserMigration\UserMigrationException;
|
||||
|
||||
/**
|
||||
* Export destination for user_migration that writes straight into the encrypted repository:
|
||||
@@ -42,7 +43,7 @@ final class RepositoryExportDestination implements IExportDestination {
|
||||
while (!feof($stream)) {
|
||||
$chunk = fread($stream, 1048576);
|
||||
if ($chunk === false) {
|
||||
break;
|
||||
throw new UserMigrationException('Read error while exporting ' . $path);
|
||||
}
|
||||
$writer->write($chunk);
|
||||
}
|
||||
|
||||
@@ -35,7 +35,13 @@ final class RepositoryImportSource implements IImportSource {
|
||||
// Phase 0: buffer in php://temp (memory up to 8 MiB, then a temp file for this one file).
|
||||
$stream = fopen('php://temp/maxmemory:' . (8 * 1048576), 'w+b');
|
||||
foreach ($this->repo->readBlobs($entry['b']) as $data) {
|
||||
fwrite($stream, $data);
|
||||
for ($done = 0, $len = strlen($data); $done < $len; $done += $n) {
|
||||
$n = fwrite($stream, $done === 0 ? $data : substr($data, $done));
|
||||
if ($n === false || $n === 0) {
|
||||
fclose($stream);
|
||||
throw new UserMigrationException('Cannot buffer ' . $path);
|
||||
}
|
||||
}
|
||||
}
|
||||
rewind($stream);
|
||||
return $stream;
|
||||
@@ -48,11 +54,13 @@ final class RepositoryImportSource implements IImportSource {
|
||||
foreach (array_keys($this->manifest['entries']) as $p) {
|
||||
if ($prefix === '' || str_starts_with($p, $prefix)) {
|
||||
$rest = substr($p, strlen($prefix));
|
||||
if ($rest !== '' && !str_contains($rest, '/')) {
|
||||
$names[] = $rest;
|
||||
if ($rest !== '') {
|
||||
// First path segment: also lists folders that only exist as a parent of a file.
|
||||
$names[explode('/', $rest, 2)[0]] = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
$names = array_keys($names);
|
||||
sort($names);
|
||||
return $names;
|
||||
}
|
||||
@@ -72,7 +80,8 @@ final class RepositoryImportSource implements IImportSource {
|
||||
|
||||
public function copyToFolder(Folder $destination, string $sourcePath): void {
|
||||
foreach ($this->getFolderListing($sourcePath) as $name) {
|
||||
$path = trim($sourcePath, '/') . '/' . $name;
|
||||
$base = trim($sourcePath, '/');
|
||||
$path = $base === '' ? $name : $base . '/' . $name;
|
||||
$entry = $this->manifest['entries'][$path] ?? ['t' => 'd'];
|
||||
if ($entry['t'] === 'd') {
|
||||
$sub = $destination->nodeExists($name) ? $destination->get($name) : $destination->newFolder($name);
|
||||
|
||||
Reference in New Issue
Block a user