diff --git a/.github/workflows/filesystem-limits.yml b/.github/workflows/filesystem-limits.yml new file mode 100644 index 0000000000..f21f61f478 --- /dev/null +++ b/.github/workflows/filesystem-limits.yml @@ -0,0 +1,40 @@ +name: Filesystem resource limits + +on: + push: + branches: ['fix/security-filesystem-limits'] + pull_request: + paths: + - 'modules/boss-service-filesystem/**' + - 'modules/boss-ipc/src/main/proto/boss/ipc/v1/services/filesystem.proto' + - 'plugin-platform/plugin-api-ipc/**' + - '.github/workflows/filesystem-limits.yml' + workflow_dispatch: + +permissions: + contents: read + +jobs: + filesystem: + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, ubuntu-24.04-arm, macos-latest, windows-latest] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-java@v6.0.0 + with: + distribution: temurin + java-version: '17' + - uses: gradle/actions/setup-gradle@v6 + - name: Run filesystem and paging regressions + shell: bash + run: ./gradlew :boss-service-filesystem:test :plugin-platform:plugin-api-ipc:desktopTest :boss-service-filesystem:ktlintCheck :boss-service-filesystem:detekt :plugin-platform:plugin-api-ipc:ktlintCheck :plugin-platform:plugin-api-ipc:detekt --no-daemon + - uses: actions/upload-artifact@v7 + if: always() + with: + name: filesystem-limits-${{ matrix.os }} + path: | + modules/boss-service-filesystem/build/test-results + plugin-platform/plugin-api-ipc/build/test-results diff --git a/modules/boss-ipc/src/main/proto/boss/ipc/v1/services/filesystem.proto b/modules/boss-ipc/src/main/proto/boss/ipc/v1/services/filesystem.proto index 38d31c38c2..e65b377c4a 100644 --- a/modules/boss-ipc/src/main/proto/boss/ipc/v1/services/filesystem.proto +++ b/modules/boss-ipc/src/main/proto/boss/ipc/v1/services/filesystem.proto @@ -51,7 +51,7 @@ message ScanDirectoryRequest { bool recursive = 2; // If true, scan subdirectories too bool include_hidden = 3; // If true, include hidden files/dirs (starting with '.') repeated string extensions = 4; // Filter by extension (e.g., "kt", "ts"). Empty = all. - int32 max_depth = 5; // Max recursion depth (0 = no limit) + int32 max_depth = 5; // Requested depth; server caps depth at 64 and scanned entries at 10,000 } message ScanDirectoryResponse { @@ -64,7 +64,7 @@ message ScanDirectoryResponse { message ReadFileRequest { string path = 1; int64 offset_bytes = 2; // 0 = start - int64 max_bytes = 3; // 0 = read all + int64 max_bytes = 3; // Explicit page size, capped at 1 MiB; 0 = read all or refuse an oversized file } message ReadFileResponse { diff --git a/modules/boss-service-filesystem/build.gradle.kts b/modules/boss-service-filesystem/build.gradle.kts index d7d33b0d29..264766c972 100644 --- a/modules/boss-service-filesystem/build.gradle.kts +++ b/modules/boss-service-filesystem/build.gradle.kts @@ -35,6 +35,7 @@ dependencies { implementation(project(":boss-ipc")) implementation(libs.kotlinx.coroutines.core) implementation(libs.slf4j.api) + implementation(libs.jna) runtimeOnly(libs.slf4j.simple) testImplementation(libs.kotlin.test.junit) diff --git a/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/BoundedDirectoryScan.kt b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/BoundedDirectoryScan.kt new file mode 100644 index 0000000000..be09b1890c --- /dev/null +++ b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/BoundedDirectoryScan.kt @@ -0,0 +1,105 @@ +package ai.rever.boss.service.filesystem + +import ai.rever.boss.ipc.proto.services.FileEntry +import ai.rever.boss.ipc.proto.services.ScanDirectoryRequest +import kotlinx.coroutines.ensureActive +import java.nio.file.FileVisitResult +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.SimpleFileVisitor +import java.nio.file.attribute.BasicFileAttributes +import kotlin.coroutines.CoroutineContext + +internal class BoundedDirectoryScan( + private val request: ScanDirectoryRequest, + private val context: CoroutineContext, + private val validateTarget: (String) -> Unit, +) { + private val requestedDepth = if (!request.recursive) 1 else request.maxDepth.takeIf { it > 0 } ?: Int.MAX_VALUE + private val depth = requestedDepth.coerceAtMost(FileSystemLimits.SCAN_DEPTH) + private val extensions = request.extensionsList.toSet() + private val entries = mutableListOf() + private var visited = 0 + private var responseBytes = 0 + + fun scan(root: Path): List { + val target = root.toRealPath() + validateTarget(target.toString()) + Files.walkFileTree( + target, + emptySet(), + depth, + object : SimpleFileVisitor() { + override fun preVisitDirectory( + dir: Path, + attrs: BasicFileAttributes, + ) = visit(target, root, dir, attrs) + + override fun visitFile( + file: Path, + attrs: BasicFileAttributes, + ) = visit(target, root, file, attrs) + }, + ) + return entries + } + + private fun visit( + root: Path, + visibleRoot: Path, + path: Path, + attributes: BasicFileAttributes, + ): FileVisitResult { + context.ensureActive() + enforceFileSystemLimit(++visited <= FileSystemLimits.SCAN_ENTRIES, "Directory scan entry limit reached") + val name = path.fileName?.toString().orEmpty() + return when { + path == root -> { + FileVisitResult.CONTINUE + } + + !request.includeHidden && name.startsWith('.') -> { + FileVisitResult.SKIP_SUBTREE + } + + else -> { + checkDepth(root, path, attributes) + if (attributes.isDirectory || extensions.isEmpty() || name.substringAfterLast('.', "") in extensions) { + add(visibleRoot.resolve(root.relativize(path)), attributes) + } + FileVisitResult.CONTINUE + } + } + } + + private fun checkDepth( + root: Path, + path: Path, + attributes: BasicFileAttributes, + ) { + if (attributes.isDirectory && requestedDepth > depth && root.relativize(path).nameCount == depth) { + Files.newDirectoryStream(path).use { + enforceFileSystemLimit(!it.iterator().hasNext(), "Directory scan depth limit reached") + } + } + } + + private fun add( + path: Path, + attributes: BasicFileAttributes, + ) { + val entry = + FileEntry + .newBuilder() + .setPath(path.toString()) + .setName(path.fileName.toString()) + .setIsDirectory(attributes.isDirectory) + .setSizeBytes(if (attributes.isDirectory) 0L else attributes.size()) + .setModifiedAt(attributes.lastModifiedTime().toMillis()) + .setIsHidden(path.fileName.toString().startsWith('.')) + .build() + responseBytes += entry.serializedSize + 8 + enforceFileSystemLimit(responseBytes <= FileSystemLimits.SCAN_BYTES, "Directory scan response limit reached") + entries.add(entry) + } +} diff --git a/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileSystemLimits.kt b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileSystemLimits.kt new file mode 100644 index 0000000000..14c0be2d10 --- /dev/null +++ b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileSystemLimits.kt @@ -0,0 +1,20 @@ +package ai.rever.boss.service.filesystem + +import io.grpc.Status + +/** Bounded RPC pages fit below gRPC's default message limit; callers can page large file reads explicitly. */ +internal object FileSystemLimits { + const val READ_BYTES = 1_048_576 + const val SCAN_ENTRIES = 10_000 + const val SCAN_DEPTH = 64 + const val SCAN_BYTES = 2_097_152 +} + +internal fun fileSystemLimit(message: String) = Status.RESOURCE_EXHAUSTED.withDescription(message).asRuntimeException() + +internal fun enforceFileSystemLimit( + condition: Boolean, + message: String, +) { + if (!condition) throw fileSystemLimit(message) +} diff --git a/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileSystemServiceImpl.kt b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileSystemServiceImpl.kt index 1c0d172190..1771b72b1d 100644 --- a/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileSystemServiceImpl.kt +++ b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileSystemServiceImpl.kt @@ -5,11 +5,11 @@ import ai.rever.boss.ipc.proto.services.* import com.google.protobuf.ByteString import io.grpc.Status import io.grpc.StatusException +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.flow -import kotlinx.coroutines.isActive import kotlinx.coroutines.withContext import org.slf4j.LoggerFactory import java.io.File @@ -18,14 +18,10 @@ import java.nio.file.AccessDeniedException import java.nio.file.AtomicMoveNotSupportedException import java.nio.file.DirectoryNotEmptyException import java.nio.file.FileAlreadyExistsException -import java.nio.file.FileSystems import java.nio.file.Files import java.nio.file.Path import java.nio.file.Paths import java.nio.file.StandardCopyOption -import java.nio.file.StandardWatchEventKinds -import java.nio.file.WatchEvent -import java.util.concurrent.TimeUnit /** * gRPC implementation of FileSystemService. @@ -61,32 +57,8 @@ class FileSystemServiceImpl : FileSystemServiceGrpcKt.FileSystemServiceCoroutine .build() } - val maxDepth = if (request.maxDepth > 0) request.maxDepth else Int.MAX_VALUE - val sequence: Sequence = - if (request.recursive) { - dir.walkTopDown().maxDepth(maxDepth) - } else { - dir.listFiles()?.asSequence() ?: emptySequence() - } - - val filterExtensions = request.extensionsList.toSet() - - val entries = - sequence - .filter { it != dir } - .filter { request.includeHidden || !it.name.startsWith(".") } - .filter { it.isDirectory || filterExtensions.isEmpty() || it.extension in filterExtensions } - .map { f -> - FileEntry - .newBuilder() - .setPath(f.absolutePath) - .setName(f.name) - .setIsDirectory(f.isDirectory) - .setSizeBytes(if (f.isDirectory) 0L else f.length()) - .setModifiedAt(f.lastModified()) - .setIsHidden(f.name.startsWith(".")) - .build() - }.toList() + val scanner = BoundedDirectoryScan(request, currentCoroutineContext(), ::validatePath) + val entries = scanner.scan(dir.toPath().toAbsolutePath()) ScanDirectoryResponse .newBuilder() @@ -106,23 +78,36 @@ class FileSystemServiceImpl : FileSystemServiceGrpcKt.FileSystemServiceCoroutine .build() } return@withContext try { - val totalSize = file.length() - val bytes = file.readBytes() - val offsetBytes = request.offsetBytes.coerceAtLeast(0L).toInt() - val slice = if (offsetBytes > 0 && offsetBytes < bytes.size) bytes.drop(offsetBytes).toByteArray() else bytes - val maxBytes = request.maxBytes - val (content, truncated) = - if (maxBytes > 0 && slice.size > maxBytes) { - slice.take(maxBytes.toInt()).toByteArray() to true - } else { - slice to false + require(request.offsetBytes >= 0 && request.maxBytes >= 0) { + "Read offsets and limits must be nonnegative" + } + val maximum = + request.maxBytes + .takeIf { it > 0 } + ?.coerceAtMost(FileSystemLimits.READ_BYTES.toLong()) + ?.toInt() ?: FileSystemLimits.READ_BYTES + val resolved = file.toPath().toRealPath() + validatePath(resolved.toString()) + openRegularFile(resolved).use { reader -> + val totalSize = reader.size + val bytes = reader.readPage(request.offsetBytes, maximum + 1) + currentCoroutineContext().ensureActive() + val truncated = bytes.size > maximum + // Legacy 'all' readers may ignore truncated. Refuse instead of returning partial text. + if (truncated && request.maxBytes == 0L) { + throw fileSystemLimit( + "File exceeds one response; read it using explicit byte limits and offsets", + ) } - ReadFileResponse - .newBuilder() - .setContent(ByteString.copyFrom(content)) - .setTotalSizeBytes(totalSize) - .setTruncated(truncated) - .build() + ReadFileResponse + .newBuilder() + .setContent(ByteString.copyFrom(bytes, 0, bytes.size.coerceAtMost(maximum))) + .setTotalSizeBytes(totalSize) + .setTruncated(truncated) + .build() + } + } catch (e: CancellationException) { + throw e } catch (e: Exception) { ReadFileResponse .newBuilder() @@ -296,89 +281,10 @@ class FileSystemServiceImpl : FileSystemServiceGrpcKt.FileSystemServiceCoroutine cause: Throwable?, ) = StatusException(code.withDescription(description).withCause(cause)) - override fun watchFileChanges(request: WatchFileChangesRequest): Flow = - flow { - logger.info("watchFileChanges: path={}, recursive={}", request.path, request.recursive) - validatePath(request.path) - val root = Paths.get(request.path) - if (!Files.exists(root)) { - logger.warn("watchFileChanges: path not found: {}", request.path) - return@flow - } + private val watches = FileWatchRegistry() - val kinds = - arrayOf( - StandardWatchEventKinds.ENTRY_CREATE, - StandardWatchEventKinds.ENTRY_MODIFY, - StandardWatchEventKinds.ENTRY_DELETE, - ) - - val watchService = FileSystems.getDefault().newWatchService() - try { - // Register root (and subdirs if recursive) - root.register(watchService, *kinds) - if (request.recursive && Files.isDirectory(root)) { - Files - .walk(root) - .filter { Files.isDirectory(it) && it != root } - .forEach { dir -> - try { - dir.register(watchService, *kinds) - } catch (_: Exception) { - } - } - } - - while (currentCoroutineContext().isActive) { - val key = - withContext(Dispatchers.IO) { - watchService.poll(500, TimeUnit.MILLISECONDS) - } ?: continue - - for (event in key.pollEvents()) { - if (event.kind() == StandardWatchEventKinds.OVERFLOW) continue - - @Suppress("UNCHECKED_CAST") - val ev = event as WatchEvent - val dir = key.watchable() as Path - val filePath = dir.resolve(ev.context()) - - val changeType = - when (event.kind()) { - StandardWatchEventKinds.ENTRY_CREATE -> FileChangeType.FILE_CHANGE_TYPE_CREATED - StandardWatchEventKinds.ENTRY_MODIFY -> FileChangeType.FILE_CHANGE_TYPE_MODIFIED - StandardWatchEventKinds.ENTRY_DELETE -> FileChangeType.FILE_CHANGE_TYPE_DELETED - else -> FileChangeType.FILE_CHANGE_TYPE_UNSPECIFIED - } - - // Register newly created directories for recursive watching - if (request.recursive && - event.kind() == StandardWatchEventKinds.ENTRY_CREATE && - Files.isDirectory(filePath) - ) { - try { - filePath.register(watchService, *kinds) - } catch (_: Exception) { - } - } - - emit( - FileChangeEvent - .newBuilder() - .setPath(filePath.toAbsolutePath().toString()) - .setChangeType(changeType) - .setTimestamp(System.currentTimeMillis()) - .build(), - ) - } - - if (!key.reset()) { - logger.info("watchFileChanges: watch key invalid, stopping: {}", request.path) - break - } - } - } finally { - watchService.close() - } - } + override fun watchFileChanges(request: WatchFileChangesRequest): Flow { + validatePath(request.path) + return watches.watch(request) + } } diff --git a/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileWatchRegistry.kt b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileWatchRegistry.kt new file mode 100644 index 0000000000..8680604ea7 --- /dev/null +++ b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/FileWatchRegistry.kt @@ -0,0 +1,198 @@ +package ai.rever.boss.service.filesystem + +import ai.rever.boss.ipc.proto.services.FileChangeEvent +import ai.rever.boss.ipc.proto.services.FileChangeType +import ai.rever.boss.ipc.proto.services.WatchFileChangesRequest +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.FlowCollector +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.flowOn +import java.io.IOException +import java.nio.file.FileSystems +import java.nio.file.FileVisitResult +import java.nio.file.Files +import java.nio.file.LinkOption.NOFOLLOW_LINKS +import java.nio.file.NoSuchFileException +import java.nio.file.Path +import java.nio.file.SimpleFileVisitor +import java.nio.file.StandardWatchEventKinds.ENTRY_CREATE +import java.nio.file.StandardWatchEventKinds.ENTRY_DELETE +import java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY +import java.nio.file.StandardWatchEventKinds.OVERFLOW +import java.nio.file.WatchKey +import java.nio.file.WatchService +import java.nio.file.attribute.BasicFileAttributes +import java.util.concurrent.Semaphore +import java.util.concurrent.TimeUnit +import kotlin.coroutines.CoroutineContext + +/** Every collector owns its registrations until cancellation, failure, or deletion of its root. */ +internal class FileWatchRegistry { + private val slots = Semaphore(32) + + fun watch(request: WatchFileChangesRequest): Flow = + flow { + if (!slots.tryAcquire()) throw fileSystemLimit("Too many active file watches") + try { + val root = Path.of(request.path).toAbsolutePath().normalize() + require(Files.isDirectory(root, NOFOLLOW_LINKS)) { "Watch root must be a directory" } + FileSystems.getDefault().newWatchService().use { service -> + val registrations = Registrations(service, root, currentCoroutineContext()) + registrations.add(root, request.recursive) + var active = true + while (active) { + currentCoroutineContext().ensureActive() + val key = service.poll(500, TimeUnit.MILLISECONDS) + if (key != null) { + emitChanges(key, request.recursive, registrations) + if (!key.reset()) { + registrations.remove(key) + active = key.watchable() != root + } + } + } + } + } finally { + slots.release() + } + }.flowOn(Dispatchers.IO) + + private suspend fun FlowCollector.emitChanges( + key: WatchKey, + recursive: Boolean, + registrations: Registrations, + ) { + val directory = key.watchable() as Path + for (event in key.pollEvents()) { + currentCoroutineContext().ensureActive() + if (event.kind() == OVERFLOW) throw fileSystemLimit("File watch overflow; rescan and reconnect") + val relative = event.context() as? Path ?: continue + val path = directory.resolve(relative) + if (recursive && event.kind() == ENTRY_CREATE && Files.isDirectory(path, NOFOLLOW_LINKS)) { + registrations.add(path, true) + } + val type = + when (event.kind()) { + ENTRY_CREATE -> FileChangeType.FILE_CHANGE_TYPE_CREATED + ENTRY_MODIFY -> FileChangeType.FILE_CHANGE_TYPE_MODIFIED + ENTRY_DELETE -> FileChangeType.FILE_CHANGE_TYPE_DELETED + else -> FileChangeType.FILE_CHANGE_TYPE_UNSPECIFIED + } + emit( + FileChangeEvent + .newBuilder() + .setPath(path.toString()) + .setChangeType(type) + .setTimestamp(System.currentTimeMillis()) + .build(), + ) + } + } + + private class Registrations( + private val service: WatchService, + private val root: Path, + private val context: CoroutineContext, + ) { + private val keys = mutableSetOf() + + fun remove(key: WatchKey) { + keys.remove(key) + } + + fun add( + path: Path, + recursive: Boolean, + ) { + var visited = 0 + Files.walkFileTree( + path, + emptySet(), + if (recursive) FileSystemLimits.SCAN_DEPTH + 1 else 0, + object : SimpleFileVisitor() { + private fun visit( + entry: Path, + attrs: BasicFileAttributes, + ): FileVisitResult { + context.ensureActive() + enforceFileSystemLimit( + ++visited <= FileSystemLimits.SCAN_ENTRIES, + "File watch scan limit reached", + ) + if (attrs.isDirectory) { + enforceFileSystemLimit( + root.relativize(entry).nameCount <= FileSystemLimits.SCAN_DEPTH, + "File watch depth limit reached", + ) + keys.removeAll { !it.isValid } + enforceFileSystemLimit(keys.size < 1024, "File watch directory limit reached") + try { + keys.add(entry.register(service, ENTRY_CREATE, ENTRY_MODIFY, ENTRY_DELETE)) + } catch (e: IOException) { + if (!disappeared(entry, e)) throw e + } + } + return FileVisitResult.CONTINUE + } + + override fun visitFileFailed( + file: Path, + exc: IOException, + ): FileVisitResult { + context.ensureActive() + if (disappeared(file, exc)) return FileVisitResult.CONTINUE + throw exc + } + + override fun preVisitDirectory( + dir: Path, + attrs: BasicFileAttributes, + ) = visit(dir, attrs) + + override fun visitFile( + file: Path, + attrs: BasicFileAttributes, + ) = visit(file, attrs) + }, + ) + } + + private fun disappeared( + path: Path, + failure: IOException, + ): Boolean = + path != root && + when (failure) { + is NoSuchFileException -> { + true + } + + else -> { + System.getProperty("os.name").startsWith("Windows") && confirmDeletion(path) + } + } + + private fun confirmDeletion(path: Path): Boolean { + // Windows can deny an open while deletion is pending. Only suppress the failure + // after an attribute read confirms absence; persistent permission failures still surface. + repeat(5) { + context.ensureActive() + Thread.sleep(10) + val absent = + try { + Files.readAttributes(path, BasicFileAttributes::class.java, NOFOLLOW_LINKS) + false + } catch (_: NoSuchFileException) { + true + } catch (_: IOException) { + false // The watch API can erase the Windows error type; only proven absence is suppressed. + } + if (absent) return true + } + return false + } + } +} diff --git a/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/RegularFileReader.kt b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/RegularFileReader.kt new file mode 100644 index 0000000000..28a1e6d1f9 --- /dev/null +++ b/modules/boss-service-filesystem/src/main/kotlin/ai/rever/boss/service/filesystem/RegularFileReader.kt @@ -0,0 +1,177 @@ +package ai.rever.boss.service.filesystem + +import com.sun.jna.Function +import com.sun.jna.Library +import com.sun.jna.Memory +import com.sun.jna.Native +import com.sun.jna.NativeLibrary +import com.sun.jna.Platform +import com.sun.jna.Pointer +import com.sun.jna.WString +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive +import java.io.ByteArrayOutputStream +import java.io.IOException +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.attribute.BasicFileAttributes + +/** Checks the opened object, never a pathname checked before a potentially blocking open. */ +internal interface RegularFileReader : AutoCloseable { + val size: Long + + fun read( + offset: Long, + destination: ByteArray, + count: Int, + ): Int +} + +internal fun openRegularFile(path: Path): RegularFileReader { + check(Native.POINTER_SIZE == 8) { "Regular-file reads require a supported 64-bit platform" } + return when { + Platform.isWindows() -> { + WindowsRegularFile(path) + } + + Platform.isMac() || (Platform.isLinux() && Platform.ARCH in setOf("x86-64", "aarch64")) -> { + PosixRegularFile(path) + } + + else -> { + throw IOException("Regular-file reads are unavailable on this platform") + } + } +} + +internal suspend fun RegularFileReader.readPage( + offset: Long, + maximum: Int, +): ByteArray { + if (offset >= size) return byteArrayOf() + val buffer = ByteArray(minOf(65_536, maximum)) + val output = ByteArrayOutputStream() + val limit = minOf(maximum.toLong(), Long.MAX_VALUE - offset).toInt() + while (output.size() < limit) { + currentCoroutineContext().ensureActive() + val position = offset + output.size() + val count = read(position, buffer, minOf(buffer.size, limit - output.size())) + if (count == 0) break + output.write(buffer, 0, count) + } + return output.toByteArray() +} + +private class PosixRegularFile( + path: Path, +) : RegularFileReader { + private val descriptor: Int + override val size: Long + + init { + // O_RDONLY | O_NONBLOCK | O_NOFOLLOW | O_CLOEXEC. Values from Darwin fcntl.h + // and Linux asm-generic/fcntl.h / arch/arm64/include/uapi/asm/fcntl.h. + val noFollow = if (Platform.ARCH == "aarch64") 0x8000 else 0x20000 + val flags = if (Platform.isMac()) 0x0004 or 0x0100 or 0x01000000 else 0x0800 or noFollow or 0x80000 + descriptor = libc.getFunction("open").invokeInt(arrayOf(path.toString(), flags)) + if (descriptor < 0) throw nativeReadError("Open") + var initialized = false + try { + // /dev/fd resolves this process's still-open descriptor on Darwin and Linux. + // Failure to inspect it is fatal; never fall back to checking the original path. + val attributes = Files.readAttributes(Path.of("/dev/fd/$descriptor"), BasicFileAttributes::class.java) + require(attributes.isRegularFile) { "Only regular files can be read" } + size = attributes.size() + initialized = true + } finally { + if (!initialized) close() + } + } + + override fun read( + offset: Long, + destination: ByteArray, + count: Int, + ): Int { + val result = libc.getFunction("pread").invokeLong(arrayOf(descriptor, destination, count.toLong(), offset)) + if (result < 0) throw nativeReadError("Read") + return result.toInt() + } + + override fun close() { + libc.getFunction("close").invokeInt(arrayOf(descriptor)) + } + + companion object { + private val libc = + NativeLibrary.getInstance(Platform.C_LIBRARY_NAME, mapOf(Library.OPTION_STRING_ENCODING to "UTF-8")) + } +} + +private class WindowsRegularFile( + path: Path, +) : RegularFileReader { + private val handle: Pointer + override val size: Long + + init { + // OPEN_EXISTING, FILE_FLAG_OPEN_REPARSE_POINT, non-inheritable handle, shared read/write/delete. + val opened = + kernel.getFunction("CreateFileW", Function.ALT_CONVENTION).invokePointer( + arrayOf(WString(windowsExtendedPath(path)), 0x80000000.toInt(), 7, null, 3, 0x00200000, null), + ) + if (opened == null || Pointer.nativeValue(opened) == -1L) throw nativeReadError("Open") + handle = opened + var initialized = false + try { + require(kernel.getFunction("GetFileType").invokeInt(arrayOf(handle)) == 1) { "Only disk files can be read" } + // BY_HANDLE_FILE_INFORMATION has thirteen DWORDs, including three FILETIMEs. + Memory(52).use { info -> + if (kernel.getFunction("GetFileInformationByHandle").invokeInt(arrayOf(handle, info)) == 0) { + throw nativeReadError("Inspect") + } + require(info.getInt(0) and (0x10 or 0x400) == 0) { "Directories and reparse points cannot be read" } + size = (info.getInt(32).toLong() shl 32) or (info.getInt(36).toLong() and 0xffffffffL) + require(size >= 0) { "File exceeds supported size" } + initialized = true + } + } finally { + if (!initialized) close() + } + } + + override fun read( + offset: Long, + destination: ByteArray, + count: Int, + ): Int { + if (kernel.getFunction("SetFilePointerEx").invokeInt(arrayOf(handle, offset, null, 0)) == 0) { + throw nativeReadError("Seek") + } + Memory(4).use { bytesRead -> + if (kernel.getFunction("ReadFile").invokeInt(arrayOf(handle, destination, count, bytesRead, null)) == 0) { + throw nativeReadError("Read") + } + return bytesRead.getInt(0) + } + } + + override fun close() { + kernel.getFunction("CloseHandle").invokeInt(arrayOf(handle)) + } + + companion object { + private val kernel = NativeLibrary.getInstance("kernel32") + } +} + +private fun nativeReadError(operation: String) = IOException("$operation failed (OS error ${Native.getLastError()})") + +private fun windowsExtendedPath(path: Path): String { + val absolute = path.toAbsolutePath().normalize().toString() + return when { + absolute.startsWith("\\\\?\\") -> absolute + absolute.startsWith("\\\\") -> "\\\\?\\UNC\\" + absolute.removePrefix("\\\\") + else -> "\\\\?\\" + absolute + } +} diff --git a/modules/boss-service-filesystem/src/test/kotlin/ai/rever/boss/service/filesystem/FileSystemLimitsTest.kt b/modules/boss-service-filesystem/src/test/kotlin/ai/rever/boss/service/filesystem/FileSystemLimitsTest.kt new file mode 100644 index 0000000000..fa0325f2fd --- /dev/null +++ b/modules/boss-service-filesystem/src/test/kotlin/ai/rever/boss/service/filesystem/FileSystemLimitsTest.kt @@ -0,0 +1,358 @@ +package ai.rever.boss.service.filesystem + +import ai.rever.boss.ipc.proto.services.FileSystemServiceGrpcKt +import ai.rever.boss.ipc.proto.services.ReadFileRequest +import ai.rever.boss.ipc.proto.services.ScanDirectoryRequest +import ai.rever.boss.ipc.proto.services.WatchFileChangesRequest +import com.sun.jna.Function +import com.sun.jna.Native +import com.sun.jna.NativeLibrary +import com.sun.jna.Platform +import com.sun.jna.Pointer +import com.sun.jna.WString +import io.grpc.ManagedChannelBuilder +import io.grpc.ServerBuilder +import io.grpc.Status +import io.grpc.StatusException +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout +import java.io.IOException +import java.io.RandomAccessFile +import java.nio.file.Files +import java.nio.file.LinkOption.NOFOLLOW_LINKS +import java.nio.file.Path +import java.nio.file.StandardWatchEventKinds.ENTRY_CREATE +import java.nio.file.attribute.BasicFileAttributes +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class FileSystemLimitsTest { + private val root = Files.createTempDirectory("filesystem-limits-") + private val server = + ServerBuilder + .forPort(0) + .addService(FileSystemServiceImpl()) + .intercept(RpcFailureDiagnostics) + .build() + .start() + private val channel = ManagedChannelBuilder.forAddress("127.0.0.1", server.port).usePlaintext().build() + private val stub = FileSystemServiceGrpcKt.FileSystemServiceCoroutineStub(channel) + + @AfterTest + fun cleanup() { + channel.shutdownNow() + server.shutdownNow() + // Files.walk does not follow symlinks; fixture aliases must never widen cleanup. + Files.walk(root).use { paths -> + paths.sorted(Comparator.reverseOrder()).forEach { Files.deleteIfExists(it) } + } + } + + @Test + fun `scan refuses an alias into a blocked system root`() = + runBlocking { + org.junit.Assume.assumeTrue(Platform.isLinux()) + val alias = Files.createSymbolicLink(root.resolve("system-alias"), Path.of("/proc")) + try { + assertFailsWith { + stub.scanDirectory(ScanDirectoryRequest.newBuilder().setPath(alias.toString()).build()) + } + } finally { + Files.deleteIfExists(alias) + } + } + + @Test + fun `native POSIX open refuses a replaced leaf link`() { + if (Platform.isWindows()) return + val target = Files.writeString(root.resolve("native-target"), "content") + openRegularFile(target).use { assertEquals(7L, it.size) } + val link = Files.createSymbolicLink(root.resolve("native-link"), target) + assertFailsWith { openRegularFile(link).use { } } + } + + @Test + fun `native open rejects a FIFO without waiting for a writer`() { + if (Platform.isWindows()) return + val fifo = root.resolve("fifo") + assertEquals(0, ProcessBuilder("mkfifo", fifo.toString()).start().waitFor()) + val worker = Executors.newSingleThreadExecutor { task -> Thread(task).apply { isDaemon = true } } + try { + val rejected = + worker.submit { + try { + openRegularFile(fifo).use { false } + } catch (_: IllegalArgumentException) { + true + } + } + assertTrue(rejected.get(2, TimeUnit.SECONDS)) + } finally { + worker.shutdownNow() + } + } + + @Test + fun `scan follows the requested root alias while leaving descendant links untraversed`() = + runBlocking { + if (Platform.isWindows()) return@runBlocking + val real = Files.createDirectory(root.resolve("real")) + Files.writeString(real.resolve("content"), "ok") + Files.createSymbolicLink(real.resolve("loop"), real) + val alias = Files.createSymbolicLink(root.resolve("alias"), real) + val result = + stub.scanDirectory( + ScanDirectoryRequest + .newBuilder() + .setPath(alias.toString()) + .setRecursive(true) + .build(), + ) + assertEquals( + setOf(alias.resolve("content").toString(), alias.resolve("loop").toString()), + result.entriesList.map { it.path }.toSet(), + ) + } + + @Test + fun `persistent Windows directory sharing failures remain visible`() = + runBlocking { + if (!Platform.isWindows()) return@runBlocking + val denied = Files.createDirectory(root.resolve("denied")) + withExclusiveDirectory(denied) { + assertTrue(Files.readAttributes(denied, BasicFileAttributes::class.java, NOFOLLOW_LINKS).isDirectory) + denied.fileSystem.newWatchService().use { watcher -> + assertFailsWith { denied.register(watcher, ENTRY_CREATE) } + } + assertFailsWith { + withTimeout(5000) { + stub + .watchFileChanges( + WatchFileChangesRequest + .newBuilder() + .setPath(root.toString()) + .setRecursive(true) + .build(), + ).first() + } + } + } + // The same operation succeeds after releasing the deliberately conflicting handle. + denied.fileSystem.newWatchService().use { watcher -> + assertTrue(denied.register(watcher, ENTRY_CREATE).isValid) + } + } + + @Test + fun `recursive watch survives disappearing descendants and continues observing the root`() = + runBlocking { + // Windows CI spent 55 seconds in synchronous fixture churn. Give fixture I/O + // its own budget; retain a separate 15-second deadline for actual watch delivery. + withTimeout(90_000) { + val sentinel = root.resolve("sentinel") + val readiness = root.resolve("watch-ready") + val ready = CompletableDeferred() + val received = + async { + stub + .watchFileChanges( + WatchFileChangesRequest + .newBuilder() + .setPath(root.toString()) + .setRecursive(true) + .build(), + ).first { + if (it.path == readiness.toString()) ready.complete(Unit) + it.path == sentinel.toString() + } + } + withTimeout(15_000) { + while (!ready.isCompleted) { + withContext(Dispatchers.IO) { Files.writeString(readiness, "ready") } + delay(25) + } + } + withContext(Dispatchers.IO) { + repeat(100) { index -> + val transient = Files.createDirectory(root.resolve("transient-$index")) + Files.createDirectory(transient.resolve("child")) + transient.toFile().deleteRecursively() + delay(5) + } + } + withTimeout(15_000) { + while (!received.isCompleted) { + withContext(Dispatchers.IO) { Files.writeString(sentinel, "still watching") } + delay(50) + } + assertEquals(sentinel.toString(), received.await().path) + } + } + } + + @Test + fun `reads through file links resolve the target before the bounded open`() = + runBlocking { + val target = Files.writeString(root.resolve("target"), "linked content") + val link = Files.createSymbolicLink(root.resolve("file-link"), target) + val response = stub.readFile(ReadFileRequest.newBuilder().setPath(link.toString()).build()) + assertTrue(response.errorMessage.isEmpty(), response.errorMessage) + assertEquals("linked content", response.content.toStringUtf8()) + } + + @Test + fun `Windows reads retain support for paths beyond MAX_PATH`() = + runBlocking { + if (!Platform.isWindows()) return@runBlocking + var directory = root + while (directory.toString().length < 300) directory = directory.resolve("long-directory-component") + Files.createDirectories(directory) + val file = Files.writeString(directory.resolve("content.txt"), "long path content") + val response = stub.readFile(ReadFileRequest.newBuilder().setPath(file.toString()).build()) + assertTrue(response.errorMessage.isEmpty(), response.errorMessage) + assertEquals("long path content", response.content.toStringUtf8()) + } + + @Test + fun `sparse files support long offsets and refuse unbounded legacy reads`() = + runBlocking { + val file = root.resolve("large") + val offset = Int.MAX_VALUE.toLong() + 32 + RandomAccessFile(file.toFile(), "rw").use { + it.seek(offset) + it.write("tail".toByteArray()) + } + val tail = + stub.readFile( + ReadFileRequest + .newBuilder() + .setPath(file.toString()) + .setOffsetBytes(offset) + .setMaxBytes(16) + .build(), + ) + assertEquals("tail", tail.content.toStringUtf8()) + assertFalse(tail.truncated) + val beyond = + stub.readFile( + ReadFileRequest + .newBuilder() + .setPath(file.toString()) + .setOffsetBytes(Long.MAX_VALUE) + .build(), + ) + assertTrue(beyond.content.isEmpty) + assertTrue(beyond.errorMessage.isEmpty()) + val legacy = stub.readFile(ReadFileRequest.newBuilder().setPath(file.toString()).build()) + assertTrue(legacy.content.isEmpty) + assertTrue(legacy.errorMessage.contains("explicit byte limits")) + val page = + stub.readFile( + ReadFileRequest + .newBuilder() + .setPath(file.toString()) + .setMaxBytes(Long.MAX_VALUE) + .build(), + ) + assertEquals(FileSystemLimits.READ_BYTES, page.content.size()) + assertTrue(page.truncated) + } + + @Test + fun `requested scan depth is honored and excessive server depth is explicit`() = + runBlocking { + var dir = root + repeat(FileSystemLimits.SCAN_DEPTH + 1) { dir = Files.createDirectory(dir.resolve("d")) } + val limited = + stub.scanDirectory( + ScanDirectoryRequest + .newBuilder() + .setPath(root.toString()) + .setRecursive(true) + .setMaxDepth(2) + .build(), + ) + assertEquals(2, limited.entriesCount) + val failure = + assertFailsWith { + stub.scanDirectory( + ScanDirectoryRequest + .newBuilder() + .setPath(root.toString()) + .setRecursive(true) + .build(), + ) + } + assertEquals(Status.Code.RESOURCE_EXHAUSTED, failure.status.code) + } + + @Test + fun `filtered entries still count toward scan work ceiling`() = + runBlocking { + repeat(FileSystemLimits.SCAN_ENTRIES) { Files.createFile(root.resolve("$it.bin")) } + val failure = + assertFailsWith { + stub.scanDirectory( + ScanDirectoryRequest + .newBuilder() + .setPath(root.toString()) + .addExtensions("txt") + .build(), + ) + } + assertEquals(Status.Code.RESOURCE_EXHAUSTED, failure.status.code) + } + + @Test + fun `watch emits real changes and releases its slot after collection`() = + runBlocking { + repeat(34) { iteration -> + withTimeout(10_000) { + val received = + async { + val request = WatchFileChangesRequest.newBuilder().setPath(root.toString()).build() + stub.watchFileChanges(request).first() + } + // Repeated writes avoid relying on an internal registration hook or scheduler timing. + while (!received.isCompleted) { + delay(25) + Files.writeString(root.resolve("event-$iteration"), "changed") + } + assertTrue(received.await().path.startsWith(root.toString())) + } + } + } +} + +private suspend fun withExclusiveDirectory( + directory: Path, + action: suspend () -> Unit, +) { + val kernel = NativeLibrary.getInstance("kernel32") + // FILE_LIST_DIRECTORY, no sharing, OPEN_EXISTING, FILE_FLAG_BACKUP_SEMANTICS. + // Attribute reads remain allowed, so the watch registry can distinguish this from deletion. + val handle = + kernel.getFunction("CreateFileW", Function.ALT_CONVENTION).invokePointer( + arrayOf(WString(directory.toString()), 1, 0, null, 3, 0x02000000, null), + ) + check(handle != null && Pointer.nativeValue(handle) != -1L) { + "Cannot acquire exclusive directory handle: ${Native.getLastError()}" + } + try { + action() + } finally { + check(kernel.getFunction("CloseHandle", Function.ALT_CONVENTION).invokeInt(arrayOf(handle)) != 0) + } +} diff --git a/modules/boss-service-filesystem/src/test/kotlin/ai/rever/boss/service/filesystem/RpcFailureDiagnostics.kt b/modules/boss-service-filesystem/src/test/kotlin/ai/rever/boss/service/filesystem/RpcFailureDiagnostics.kt new file mode 100644 index 0000000000..d266272ad0 --- /dev/null +++ b/modules/boss-service-filesystem/src/test/kotlin/ai/rever/boss/service/filesystem/RpcFailureDiagnostics.kt @@ -0,0 +1,32 @@ +package ai.rever.boss.service.filesystem + +import io.grpc.ForwardingServerCall +import io.grpc.Metadata +import io.grpc.ServerCall +import io.grpc.ServerCallHandler +import io.grpc.ServerInterceptor +import io.grpc.Status +import org.slf4j.LoggerFactory + +/** Keep the actual server exception in CI reports; a client receives only UNKNOWN for unexpected IO. */ +internal object RpcFailureDiagnostics : ServerInterceptor { + private val logger = LoggerFactory.getLogger(RpcFailureDiagnostics::class.java) + + override fun interceptCall( + call: ServerCall, + headers: Metadata, + next: ServerCallHandler, + ): ServerCall.Listener = + next.startCall( + object : ForwardingServerCall.SimpleForwardingServerCall(call) { + override fun close( + status: Status, + trailers: Metadata, + ) { + if (!status.isOk) logger.warn("Filesystem test RPC closed with {}", status, status.cause) + super.close(status, trailers) + } + }, + headers, + ) +} diff --git a/plugin-platform/plugin-api-ipc/src/desktopMain/kotlin/ai/rever/boss/plugin/ipc/FileSystemDataProviderProxy.kt b/plugin-platform/plugin-api-ipc/src/desktopMain/kotlin/ai/rever/boss/plugin/ipc/FileSystemDataProviderProxy.kt index 0901651cb3..d04df97331 100644 --- a/plugin-platform/plugin-api-ipc/src/desktopMain/kotlin/ai/rever/boss/plugin/ipc/FileSystemDataProviderProxy.kt +++ b/plugin-platform/plugin-api-ipc/src/desktopMain/kotlin/ai/rever/boss/plugin/ipc/FileSystemDataProviderProxy.kt @@ -14,6 +14,7 @@ import ai.rever.boss.plugin.api.FileSystemDataProvider import ai.rever.boss.plugin.api.NodeLoadingStateData import com.google.protobuf.ByteString import io.grpc.ManagedChannel +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob @@ -21,6 +22,7 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withContext import kotlinx.coroutines.withTimeout +import java.io.ByteArrayOutputStream import java.io.File /** @@ -254,15 +256,38 @@ class FileSystemDataProviderProxy( override suspend fun readFile(path: String): Result = withContext(Dispatchers.Default) { try { - val response = - fsStub.readFile( - ReadFileRequest.newBuilder().setPath(path).build(), - ) - if (response.errorMessage.isNotBlank()) { - Result.failure(Exception(response.errorMessage)) - } else { - Result.success(response.content.toStringUtf8()) - } + val content = ByteArrayOutputStream() + var expectedSize: Long? = null + do { + val response = + fsStub.readFile( + ReadFileRequest + .newBuilder() + .setPath(path) + .setOffsetBytes(content.size().toLong()) + .setMaxBytes(1_048_576) + .build(), + ) + check(response.errorMessage.isBlank()) { response.errorMessage } + check( + response.totalSizeBytes in 0..MAX_TEXT_BYTES.toLong() && + content.size() + response.content.size() <= MAX_TEXT_BYTES, + ) { + "Text files larger than 8 MiB cannot be read through the plugin provider" + } + check(expectedSize == null || expectedSize == response.totalSizeBytes) { + "File changed during read; retry" + } + expectedSize = response.totalSizeBytes + check(!response.truncated || !response.content.isEmpty) { + "File service returned an empty partial page" + } + response.content.writeTo(content) + } while (response.truncated) + check(content.size().toLong() == expectedSize) { "File changed during read; retry" } + Result.success(content.toString(Charsets.UTF_8)) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { Result.failure(e) } @@ -348,3 +373,6 @@ class FileSystemDataProviderProxy( return sb.toString() } } + +// This provider returns a complete in-memory String. Larger files require a streaming editor API. +private const val MAX_TEXT_BYTES = 8_388_608 diff --git a/plugin-platform/plugin-api-ipc/src/desktopTest/kotlin/ai/rever/boss/plugin/ipc/FileSystemReadPagingTest.kt b/plugin-platform/plugin-api-ipc/src/desktopTest/kotlin/ai/rever/boss/plugin/ipc/FileSystemReadPagingTest.kt new file mode 100644 index 0000000000..6fe6a6c588 --- /dev/null +++ b/plugin-platform/plugin-api-ipc/src/desktopTest/kotlin/ai/rever/boss/plugin/ipc/FileSystemReadPagingTest.kt @@ -0,0 +1,86 @@ +package ai.rever.boss.plugin.ipc + +import ai.rever.boss.ipc.proto.services.FileSystemServiceGrpcKt +import ai.rever.boss.ipc.proto.services.ReadFileRequest +import ai.rever.boss.ipc.proto.services.ReadFileResponse +import com.google.protobuf.ByteString +import io.grpc.ManagedChannelBuilder +import io.grpc.ServerBuilder +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class FileSystemReadPagingTest { + @Test + fun `UTF8 characters split across pages are decoded once`() = + runBlocking { + val expected = "a".repeat(1_048_575) + "🙂tail" + val bytes = expected.toByteArray() + val offsets = mutableListOf() + withService({ request -> + offsets.add(request.offsetBytes) + val start = request.offsetBytes.toInt() + val count = minOf(request.maxBytes.toInt(), bytes.size - start) + ReadFileResponse + .newBuilder() + .setContent(ByteString.copyFrom(bytes, start, count)) + .setTotalSizeBytes(bytes.size.toLong()) + .setTruncated(start + count < bytes.size) + .build() + }) { proxy -> + assertEquals(expected, proxy.readFile("test.txt").getOrThrow()) + } + assertEquals(listOf(0L, 1_048_576L), offsets) + } + + @Test + fun `changing file size and oversized files fail instead of returning partial text`() = + runBlocking { + withService({ request -> + ReadFileResponse + .newBuilder() + .setContent(ByteString.copyFromUtf8("a")) + .setTotalSizeBytes(if (request.offsetBytes == 0L) 2 else 3) + .setTruncated(request.offsetBytes == 0L) + .build() + }) { proxy -> assertTrue(proxy.readFile("test").isFailure) } + withService({ + ReadFileResponse.newBuilder().setTotalSizeBytes(8_388_609).build() + }) { proxy -> assertTrue(proxy.readFile("test").isFailure) } + } + + @Test + fun `empty partial pages and service errors terminate the read`() = + runBlocking { + withService({ ReadFileResponse.newBuilder().setTruncated(true).build() }) { proxy -> + assertTrue(proxy.readFile("test").isFailure) + } + withService({ ReadFileResponse.newBuilder().setErrorMessage("denied").build() }) { proxy -> + assertEquals("denied", proxy.readFile("test").exceptionOrNull()?.message) + } + } + + private suspend fun withService( + response: (ReadFileRequest) -> ReadFileResponse, + action: suspend (FileSystemDataProviderProxy) -> Unit, + ) { + val service = + object : FileSystemServiceGrpcKt.FileSystemServiceCoroutineImplBase() { + override suspend fun readFile(request: ReadFileRequest) = response(request) + } + val server = + ServerBuilder + .forPort(0) + .addService(service) + .build() + .start() + val channel = ManagedChannelBuilder.forAddress("127.0.0.1", server.port).usePlaintext().build() + try { + action(FileSystemDataProviderProxy(channel, channel)) + } finally { + channel.shutdownNow() + server.shutdownNow() + } + } +}