diff --git a/lib/Backend/SftpReadStream.php b/lib/Backend/SftpReadStream.php new file mode 100644 index 0000000..9bd9a33 --- /dev/null +++ b/lib/Backend/SftpReadStream.php @@ -0,0 +1,262 @@ +|null + */ + private static ?\WeakMap $chunks = null; + private static int $requestId = 0x4e470000; + + /** @var resource|null set by PHP for stream wrappers */ + public $context; + + /** + * The phpseclib connection. PHP creates stream wrappers without constructor arguments; it is + * set in stream_open(), and no other method is called when that fails. + * + * @psalm-suppress PropertyNotSetInConstructor + */ + private object $sftp; + private string $handle = ''; + /** Position of the reader: the first byte of $buffer. */ + private int $pos = 0; + /** Offset of the next READ to send. */ + private int $next = 0; + private string $buffer = ''; + /** @var list request id, offset, length, in the order sent */ + private array $inFlight = []; + private int $depth = 1; + private bool $eof = false; + private bool $failed = false; + /** Length of the last short answer, until the next answer tells whether it was the end of the file. */ + private ?int $shortAnswer = null; + + /** + * @param object $storage files_external's SFTP storage (getConnection(), getRoot()) + * @return resource|false|null the stream, false when the file cannot be opened, null when the + * connection does not offer what this stream needs + */ + public static function open(object $storage, string $path) { + if (!method_exists($storage, 'getConnection') || !method_exists($storage, 'getRoot')) { + return null; + } + $sftp = $storage->getConnection(); + if (!is_object($sftp) || !method_exists($sftp, '_send_sftp_packet') || !method_exists($sftp, '_get_sftp_packet') + || !method_exists($sftp, '_realpath') || !property_exists($sftp, 'packet_type')) { + return null; + } + if (!in_array(self::PROTOCOL, stream_get_wrappers(), true)) { + stream_wrapper_register(self::PROTOCOL, self::class); + } + $root = rtrim((string)$storage->getRoot(), '/'); + $context = stream_context_create([self::PROTOCOL => ['session' => $sftp, 'path' => $root . '/' . ltrim($path, '/')]]); + $fh = @fopen(self::PROTOCOL . '://file', 'r', false, $context); + return $fh === false ? false : $fh; + } + + /** Test hook: forget the learned READ sizes. */ + public static function reset(): void { + self::$chunks = null; + } + + public function stream_open(string $url, string $mode, int $options, ?string &$openedPath): bool { + $options = stream_context_get_options($this->context ?? stream_context_get_default())[self::PROTOCOL] ?? []; + if (!isset($options['session'], $options['path']) || !is_object($options['session'])) { + return false; + } + $this->sftp = $options['session']; + $remote = $this->sftp->_realpath((string)$options['path']); + if (!is_string($remote)) { + return false; + } + $id = self::nextId(); + if (!$this->sftp->_send_sftp_packet(self::FXP_OPEN, pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0), $id)) { + return false; + } + $response = $this->sftp->_get_sftp_packet($id); + if ($this->sftp->packet_type !== self::FXP_HANDLE || !is_string($response)) { + return false; + } + $this->handle = substr($response, 4); + return true; + } + + public function stream_read(int $count): string|false { + while (strlen($this->buffer) < $count && !$this->eof) { + if (!$this->receive()) { + return false; + } + } + $data = substr($this->buffer, 0, $count); + $this->buffer = substr($this->buffer, strlen($data)); + $this->pos += strlen($data); + return $data; + } + + public function stream_eof(): bool { + return $this->eof && $this->buffer === ''; + } + + public function stream_tell(): int { + return $this->pos; + } + + public function stream_seek(int $offset, int $whence = SEEK_SET): bool { + if ($whence === SEEK_CUR) { + $offset += $this->pos; + } elseif ($whence !== SEEK_SET) { + return false; // the size is never asked for, to save a round trip + } + if ($offset < 0 || $this->failed) { + return false; + } + if ($offset >= $this->pos && $offset - $this->pos <= strlen($this->buffer)) { + $this->buffer = substr($this->buffer, $offset - $this->pos); + } else { + $this->drain(); + $this->buffer = ''; + $this->next = $offset; + $this->eof = false; + $this->depth = 1; + $this->shortAnswer = null; + } + $this->pos = $offset; + return true; + } + + /** @return array */ + public function stream_stat(): array { + return []; + } + + public function stream_close(): void { + $this->drain(); + if ($this->handle !== '') { + $id = self::nextId(); + if ($this->sftp->_send_sftp_packet(self::FXP_CLOSE, pack('Na*', strlen($this->handle), $this->handle), $id)) { + $this->sftp->_get_sftp_packet($id); + } + } + $this->handle = ''; + } + + /** Keep the pipeline full, then take the oldest answer. False on a connection or server error. */ + private function receive(): bool { + if ($this->failed) { + return false; + } + while (count($this->inFlight) < $this->depth) { + $id = self::nextId(); + $length = self::chunk($this->sftp); + $packet = pack('Na*N3', strlen($this->handle), $this->handle, intdiv($this->next, 4294967296), $this->next % 4294967296, $length); + if (!$this->sftp->_send_sftp_packet(self::FXP_READ, $packet, $id)) { + return $this->fail(); + } + $this->inFlight[] = [$id, $this->next, $length]; + $this->next += $length; + } + [$id, $offset, $length] = array_shift($this->inFlight); + $response = $this->sftp->_get_sftp_packet($id); + if (!is_string($response)) { + return $this->fail(); + } + if ($this->sftp->packet_type === self::FXP_DATA) { + $data = substr($response, 4); + if ($data === '') { + return $this->fail(); + } + $this->buffer .= $data; + $this->depth = min(self::DEPTH, $this->depth * 2); + if ($this->shortAnswer !== null) { + // More data after a short answer: that one was the server's cap, not the end. + // Never below what every server must support: a one-off small answer is no cap. + self::$chunks ??= new \WeakMap(); + self::$chunks[$this->sftp] = max(self::MIN_CHUNK, min(self::chunk($this->sftp), $this->shortAnswer)); + $this->shortAnswer = null; + } + if (strlen($data) < $length) { + // Fewer bytes than asked: the end of the file, or a server that caps a READ. The + // requests behind this one assumed a full answer, so drop them and ask again from here. + $this->drain(); + $this->next = $offset + strlen($data); + $this->shortAnswer = strlen($data); + } + return true; + } + $status = $this->sftp->packet_type === self::FXP_STATUS ? unpack('N', $response) : false; + if (is_array($status) && ($status[1] ?? null) === self::FX_EOF) { + $this->eof = true; + $this->shortAnswer = null; + $this->drain(); + return true; + } + return $this->fail(); + } + + private function fail(): bool { + $this->drain(); + $this->failed = true; + return false; + } + + /** Read and drop the answers still on their way, so the connection is free again. */ + private function drain(): void { + foreach ($this->inFlight as [$id]) { + if (!$this->failed) { + $this->sftp->_get_sftp_packet($id); + } + } + $this->inFlight = []; + } + + private static function chunk(object $sftp): int { + $learned = self::$chunks?->offsetExists($sftp) === true ? self::$chunks->offsetGet($sftp) : null; + return is_int($learned) ? $learned : self::CHUNK; + } + + private static function nextId(): int { + self::$requestId = self::$requestId >= 0x7fffffff ? 0x4e470000 : self::$requestId + 1; + return self::$requestId; + } +} diff --git a/lib/Backend/StorageBackend.php b/lib/Backend/StorageBackend.php index 58f53e2..80f4c7d 100644 --- a/lib/Backend/StorageBackend.php +++ b/lib/Backend/StorageBackend.php @@ -17,6 +17,8 @@ use OCP\Files\Storage\IWriteStreamStorage; final class StorageBackend implements IBackend { /** Object stores write whole objects atomically; other storages get a temp name + rename. */ private bool $atomicPut; + /** SFTP: read through SftpReadStream, which keeps several requests in flight. */ + private bool $sftpReads; public function __construct( private IStorage $storage, @@ -25,6 +27,7 @@ final class StorageBackend implements IBackend { $this->base = trim($base, '/'); self::validate($this->base, true); $this->atomicPut = $storage->instanceOfStorage(\OCA\Files_External\Lib\Storage\AmazonS3::class); + $this->sftpReads = $storage->instanceOfStorage(\OCA\Files_External\Lib\Storage\SFTP::class); if ($this->base !== '') { $this->mkdirs($this->base); } @@ -79,6 +82,16 @@ final class StorageBackend implements IBackend { } public function get(string $path) { + if ($this->sftpReads) { + $fh = SftpReadStream::open($this->storage, $this->abs($path)); + if (is_resource($fh)) { + return $fh; + } + if ($fh === null) { + $this->sftpReads = false; + } + // false: not opened, which need not mean missing; the storage's own stream decides + } $fh = $this->storage->fopen($this->abs($path), 'r'); if ($fh === false) { throw new BackendException('Not found: ' . $path); diff --git a/tests/Unit/Backend/SftpReadStreamTest.php b/tests/Unit/Backend/SftpReadStreamTest.php new file mode 100644 index 0000000..3d40b4a --- /dev/null +++ b/tests/Unit/Backend/SftpReadStreamTest.php @@ -0,0 +1,243 @@ + $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'repo/packs/ab/abcd'); + + $this->assertIsResource($fh); + $this->assertSame($data, stream_get_contents($fh)); + $this->assertTrue(feof($fh)); + fclose($fh); + $this->assertSame(SftpReadStream::DEPTH, $sftp->maxInFlight); + $this->assertSame(0, $sftp->pending(), 'nothing may be left on the connection after close'); + $this->assertSame([], $sftp->openHandles); + } + + public function testAServerThatCapsAReadStillGivesEveryByteInOrder(): void { + $data = random_bytes(3 * 1048576 + 7); + $sftp = new FakeSftpConnection(['/home/f' => $data], cap: 100000); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + + // The cap is learned: a next stream asks for no more than the server gives. + $sftp->reads = []; + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame([100000], array_values(array_unique($sftp->reads))); + + // Only for that connection. + $other = new FakeSftpConnection(['/home/f' => $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($other), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(261120, max($other->reads)); + } + + public function testATinyAnswerDoesNotShrinkTheReadsBelowWhatEveryServerSupports(): void { + $data = random_bytes(1048576); + $sftp = new FakeSftpConnection(['/home/f' => $data], cap: 1000); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(32768, min($sftp->reads)); + } + + public function testSeekingReadsFromTheNewOffset(): void { + $data = random_bytes(3 * 1048576); + $sftp = new FakeSftpConnection(['/home/f' => $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + + $this->assertSame(0, fseek($fh, 2000000)); + $this->assertSame(2000000, ftell($fh)); + $this->assertSame(substr($data, 2000000, 1000), fread($fh, 1000)); + $this->assertSame(0, fseek($fh, 10)); // backwards + $this->assertSame(substr($data, 10, 70000), $this->readExactly($fh, 70000)); + $this->assertSame(0, fseek($fh, 100, SEEK_CUR)); // within what is already buffered + $this->assertSame(substr($data, 70110, 5000), $this->readExactly($fh, 5000)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + } + + public function testASmallReadDoesNotFetchTheWholePipelineAhead(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(10 * 1048576)]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + fread($fh, 100); + fclose($fh); + $this->assertCount(1, $sftp->reads, 'the first request goes alone, the pipeline grows while reading on'); + } + + public function testAnEmptyFileIsAtEndAtOnce(): void { + $sftp = new FakeSftpConnection(['/home/f' => '']); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame('', stream_get_contents($fh)); + $this->assertTrue(feof($fh)); + fclose($fh); + } + + public function testAMissingFileCannotBeOpened(): void { + $sftp = new FakeSftpConnection([]); + $this->assertFalse(SftpReadStream::open(new FakeSftpStorage($sftp), 'nope')); + $this->assertSame(0, $sftp->pending()); + } + + public function testAServerErrorFailsTheReadAndLeavesTheConnectionUsable(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(3 * 1048576), '/home/g' => 'second'], failAt: 1048576); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $read = ''; + while (!feof($fh)) { + $chunk = fread($fh, 65536); + if ($chunk === false || $chunk === '') { + break; + } + $read .= $chunk; + } + $this->assertLessThan(3 * 1048576, strlen($read)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'g'); + $this->assertSame('second', stream_get_contents($fh)); + fclose($fh); + } + + public function testAConnectionWithoutTheNeededMethodsIsNotUsed(): void { + $storage = new class { + public function getRoot(): string { + return '/'; + } + + public function getConnection(): object { + return new \stdClass(); + } + }; + $this->assertNull(SftpReadStream::open($storage, 'f')); + $this->assertNull(SftpReadStream::open(new \stdClass(), 'f')); + } + + /** @param resource $fh */ + private function readExactly($fh, int $length): string { + $data = ''; + while (strlen($data) < $length && !feof($fh)) { + $data .= fread($fh, $length - strlen($data)); + } + return $data; + } +} + +final class FakeSftpStorage { + public function __construct( + private FakeSftpConnection $connection, + ) { + } + + public function getRoot(): string { + return '/home/'; + } + + public function getConnection(): FakeSftpConnection { + return $this->connection; + } +} + +/** The part of phpseclib\Net\SFTP (2.x) that SftpReadStream uses. */ +final class FakeSftpConnection { + public int $packet_type = -1; + public int $maxInFlight = 0; + /** @var list lengths asked for by READ requests */ + public array $reads = []; + /** @var array */ + public array $openHandles = []; + /** @var array answers by request id, in the order the server sent them */ + private array $answers = []; + private int $handles = 0; + + /** @param array $files */ + public function __construct( + private array $files, + private int $cap = 261120, + private ?int $failAt = null, + ) { + } + + public function pending(): int { + return count($this->answers); + } + + public function _realpath(string $path): string { + return '/' . trim(preg_replace('#/+#', '/', $path), '/'); + } + + public function _send_sftp_packet(int $type, string $data, int $id = 1): bool { + switch ($type) { + case 3: // OPEN + $path = substr($data, 4, unpack('N', $data)[1]); + if (!isset($this->files[$path])) { + $this->answers[$id] = [101, pack('N', 2)]; + break; + } + $handle = 'h' . ++$this->handles; + $this->openHandles[$handle] = $path; + $this->answers[$id] = [102, pack('Na*', strlen($handle), $handle)]; + break; + case 4: // CLOSE + $handle = substr($data, 4, unpack('N', $data)[1]); + unset($this->openHandles[$handle]); + $this->answers[$id] = [101, pack('N', 0)]; + break; + case 5: // READ + $len = unpack('N', $data)[1]; + $handle = substr($data, 4, $len); + ['hi' => $hi, 'lo' => $lo, 'n' => $n] = unpack('Nhi/Nlo/Nn', substr($data, 4 + $len)); + $offset = $hi * 4294967296 + $lo; + $this->reads[] = $n; + $file = $this->files[$this->openHandles[$handle]]; + if ($this->failAt !== null && $offset >= $this->failAt) { + $this->answers[$id] = [101, pack('N', 4)]; // FX_FAILURE + } elseif ($offset >= strlen($file)) { + $this->answers[$id] = [101, pack('N', 1)]; // FX_EOF + } else { + $chunk = substr($file, $offset, min($n, $this->cap)); + $this->answers[$id] = [103, pack('Na*', strlen($chunk), $chunk)]; + } + $this->maxInFlight = max($this->maxInFlight, count($this->answers)); + break; + default: + throw new \LogicException("unexpected packet type $type"); + } + return true; + } + + public function _get_sftp_packet(?int $id = null): string|false { + if ($this->answers === []) { + throw new \LogicException('waiting for an answer that will never come'); + } + if ($id === null || !isset($this->answers[$id])) { + throw new \LogicException("no answer for request $id"); + } + [$this->packet_type, $payload] = $this->answers[$id]; + unset($this->answers[$id]); + return $payload; + } +}