From 5c25ef6f545e6528cba0d43709d7fa0a5b8a8047 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 2 Oct 2026 05:53:07 +0800 Subject: [PATCH 1/7] build: stage FUSE integration for cloud verification --- .fuse-integration.patch | 103 +++++ .github/workflows/apply-fuse-integration.yml | 52 +++ .github/workflows/fuse.yml | 30 ++ docs/fuse.md | 81 ++++ scripts/check_fuse_mount.py | 111 +++++ src/msg/client_mount.py | 428 +++++++++++++++++++ tests/test_client_mount.py | 305 +++++++++++++ tests/test_mount_service.py | 95 ++++ 8 files changed, 1205 insertions(+) create mode 100644 .fuse-integration.patch create mode 100644 .github/workflows/apply-fuse-integration.yml create mode 100644 .github/workflows/fuse.yml create mode 100644 docs/fuse.md create mode 100644 scripts/check_fuse_mount.py create mode 100644 src/msg/client_mount.py create mode 100644 tests/test_client_mount.py create mode 100644 tests/test_mount_service.py diff --git a/.fuse-integration.patch b/.fuse-integration.patch new file mode 100644 index 00000000..11293de6 --- /dev/null +++ b/.fuse-integration.patch @@ -0,0 +1,103 @@ +diff --git a/pyproject.toml b/pyproject.toml +index 9d926d9..503df93 100644 +--- a/pyproject.toml ++++ b/pyproject.toml +@@ -25,6 +25,7 @@ msg = "msg.cli:main" + msgd = "msg.daemon:main" + + [project.optional-dependencies] ++fuse = ["mfusepy>=3.1.1,<4"] + server = [ + "starlette>=0.50,<2", + "uvicorn>=0.38,<1", +diff --git a/src/msg/cli.py b/src/msg/cli.py +index 3156250..8af61c6 100644 +--- a/src/msg/cli.py ++++ b/src/msg/cli.py +@@ -507,6 +507,9 @@ def parser(): + from msg.client_yubikey import add_commands as add_yubikey_commands + + add_yubikey_commands(commands) ++ from msg.client_mount import add_commands as add_mount_commands ++ ++ add_mount_commands(commands) + return cli + + +@@ -581,6 +584,10 @@ async def run(args): + require(args.handle == args.user, 'connection_user_mismatch') + elif command != 'login': + await client.require_username(args.user) ++ if command == 'mount': ++ from msg.client_mount import run_mount ++ ++ return await run_mount(client, args) + if command in {'login', 'logout', 'auth'}: + from msg.client_oauth import run_command + +diff --git a/src/msg/plugins/files.py b/src/msg/plugins/files.py +index f56bedb..018fe3d 100644 +--- a/src/msg/plugins/files.py ++++ b/src/msg/plugins/files.py +@@ -43,9 +43,13 @@ def install(app): + @op('file.stat', obj({'id': IDENTIFIER}, ('id',)), effect='read') + async def stat(ctx, request, tx): + spec = app.registry.operation('discovery.get') +- return await spec.handler( ++ result = await spec.handler( + ctx, replace(request, arguments={**request.arguments, 'view': 'meta'}), tx + ) ++ resource_type = app.registry.resource_type( ++ result.data['type'], result.data['type_version'] ++ ) ++ return replace(result, data={**result.data, 'container': resource_type.container}) + + @op( + 'file.write', +diff --git a/uv.lock b/uv.lock +index 7da9be3..b47431a 100644 +--- a/uv.lock ++++ b/uv.lock +@@ -471,6 +471,15 @@ wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, + ] + ++[[package]] ++name = "mfusepy" ++version = "3.1.1" ++source = { registry = "https://pypi.org/simple" } ++sdist = { url = "https://files.pythonhosted.org/packages/91/47/746287c8962274f73ee25edb3840d80899464bfffbe2c435424c2d60a071/mfusepy-3.1.1.tar.gz", hash = "sha256:338ece54513d7d1a5e9492837679a0c7432ecf96a03490a2683a1ce1d19570e1", size = 34549, upload-time = "2026-03-13T00:36:52.636Z" } ++wheels = [ ++ { url = "https://files.pythonhosted.org/packages/c5/d5/56bc9326bf75f7ea0f4fe1178f5c9f20d1a1e15e288ecf66088513538ccd/mfusepy-3.1.1-py3-none-any.whl", hash = "sha256:69fb70cfc7f7cce595e6ff586f8451d8298f01f18286a157f24f564b24ec5a37", size = 28264, upload-time = "2026-03-13T00:36:51.843Z" }, ++] ++ + [[package]] + name = "more-itertools" + version = "11.1.0" +@@ -509,6 +518,9 @@ dev = [ + { name = "uvicorn" }, + { name = "valkey" }, + ] ++fuse = [ ++ { name = "mfusepy" }, ++] + server = [ + { name = "aiohttp" }, + { name = "dnspython" }, +@@ -530,6 +542,7 @@ requires-dist = [ + { name = "httpx", specifier = ">=0.28,<1" }, + { name = "jsonschema", specifier = ">=4.25,<5" }, + { name = "markdown-it-py", marker = "extra == 'server'", specifier = ">=4,<5" }, ++ { name = "mfusepy", marker = "extra == 'fuse'", specifier = ">=3.1.1,<4" }, + { name = "msgctl", extras = ["server"], marker = "extra == 'dev'" }, + { name = "psycopg", extras = ["binary"], marker = "extra == 'server'", specifier = ">=3.3,<4" }, + { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.4" }, +@@ -543,7 +556,7 @@ requires-dist = [ + { name = "valkey", marker = "extra == 'server'", specifier = ">=6.1,<7" }, + { name = "yubikey-manager", specifier = ">=5.8,<6" }, + ] +-provides-extras = ["server", "dev"] ++provides-extras = ["fuse", "server", "dev"] + + [[package]] + name = "multidict" diff --git a/.github/workflows/apply-fuse-integration.yml b/.github/workflows/apply-fuse-integration.yml new file mode 100644 index 00000000..71cb779c --- /dev/null +++ b/.github/workflows/apply-fuse-integration.yml @@ -0,0 +1,52 @@ +name: Apply reviewed FUSE integration +on: + push: + branches: [work/fuse-mount-20261001] +permissions: + contents: write +jobs: + apply: + if: github.repository == 'TokenNotIncluded/msg' && github.event.head_commit.message == 'build: stage FUSE integration for cloud verification' + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + - uses: actions/setup-python@v7 + with: + python-version: '3.15' + allow-prereleases: true + - name: Apply the reviewed local patch without replacing concurrent changes + run: | + git apply --check .fuse-integration.patch + git apply .fuse-integration.patch + sudo apt-get update -qq + sudo env NEEDRESTART_MODE=l apt-get install -y -qq fuse3 libpcsclite-dev swig + python -m pip install uv==0.12.20 + uv sync --locked --extra dev --extra fuse --python "$(command -v python)" + .venv/bin/ruff check --fix src/msg/client_mount.py src/msg/cli.py src/msg/plugins/files.py tests/test_client_mount.py tests/test_mount_service.py scripts/check_fuse_mount.py + .venv/bin/ruff format src/msg/client_mount.py src/msg/cli.py src/msg/plugins/files.py tests/test_client_mount.py tests/test_mount_service.py scripts/check_fuse_mount.py + - name: Commit complete implementation and remove this one-off staging workflow + run: | + git rm .fuse-integration.patch .github/workflows/apply-fuse-integration.yml + git add pyproject.toml uv.lock src/msg/client_mount.py src/msg/cli.py src/msg/plugins/files.py tests/test_client_mount.py tests/test_mount_service.py scripts/check_fuse_mount.py + git config user.name 'github-actions[bot]' + git config user.email '41898282+github-actions[bot]@users.noreply.github.com' + git commit -m 'feat: add authenticated read-only FUSE mounting' + git push origin HEAD:refs/heads/work/fuse-mount-20261001 + - name: Test mount contracts on Python 3.15 + run: .venv/bin/python -m pytest tests/test_client_mount.py --junitxml=artifacts/fuse-unit.xml + - name: Exercise actual Linux FUSE mount and cancellation + run: .venv/bin/python scripts/check_fuse_mount.py + - name: Export final reviewed delta + if: always() + run: | + mkdir -p artifacts + git diff c6be2080307a1a067c2117b9791e3a7a995a08de -- . ':!.fuse-integration.patch' ':!.github/workflows/apply-fuse-integration.yml' > artifacts/msg-fuse.patch + git rev-parse HEAD > artifacts/commit.txt + - uses: actions/upload-artifact@v7.0.1 + if: always() + with: + name: fuse-verification + path: artifacts/ diff --git a/.github/workflows/fuse.yml b/.github/workflows/fuse.yml new file mode 100644 index 00000000..ba3ee7a5 --- /dev/null +++ b/.github/workflows/fuse.yml @@ -0,0 +1,30 @@ +name: FUSE mount +on: + pull_request: + push: + branches: [main] + workflow_dispatch: +permissions: + contents: read +concurrency: + group: fuse-${{ github.event.pull_request.head.repo.full_name || github.repository }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true +jobs: + kernel: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v7 + with: + python-version: '3.15' + allow-prereleases: true + - name: Install native FUSE and optional package + run: | + sudo apt-get update -qq + sudo env NEEDRESTART_MODE=l apt-get install -y -qq fuse3 libpcsclite-dev swig + python -m pip install uv==0.12.20 + uv sync --locked --extra dev --extra fuse --python "$(command -v python)" + test -c /dev/fuse + - name: Verify real kernel mount, byte reads and unmount + run: uv run --locked --extra dev --extra fuse python scripts/check_fuse_mount.py diff --git a/docs/fuse.md b/docs/fuse.md new file mode 100644 index 00000000..7cd58cc7 --- /dev/null +++ b/docs/fuse.md @@ -0,0 +1,81 @@ +# 把 msg 挂载为本地文件夹 + +`msg mount` 把服务端资源树映射为只读目录，不抓取网页，也不整站下载。 +挂载后可以使用 `ls`、`cat`、`grep`、文件管理器和本地编辑器查看内容。 +本版本不支持在挂载点新建、保存、移动或删除远端资源；写入仍使用明确的 CLI 操作。 + +## 安装与启动 + +Linux 安装系统 FUSE 3，例如 Debian/Ubuntu： + +```sh +sudo apt-get install fuse3 +``` + +macOS 需要先安装 macFUSE。Linux 有真实内核挂载 CI；macOS 尚未完成实机验收。 +Windows 原生挂载不在本版支持范围；容器、WSL、受限沙箱必须自行具备 FUSE 设备与挂载权限。 +客户端不会替用户提权或修改系统配置。 + +使用包含本功能的源码版本安装可选依赖（PyPI 发布前不能假定现有发行版已包含）： + +```sh +uv tool install --from '.[fuse]' msgctl +``` + +客户端和服务端都需要更新：`file.stat` 必须提供资源类型的 `container` 字段。 +旧服务端会报 `mount_server_upgrade_required`，不会猜测目录或把错误当成空目录。 + +```sh +# 明确指定站点；采用该站点在 CLI 中已选定的账号。 +msg --server https://msg.lmm.best mount ~/msg + +# 已经设置默认站点后，可直接挂载。 +msg mount ~/msg + +# 只挂载一个远程目录。 +msg mount ~/msg-main --path /main + +# 选择已有账号；不创建新的挂载身份。 +msg --account alice mount ~/msg-alice +``` + +命令在前台运行。另一终端可执行 `ls ~/msg/main`，并对列出的实际文件名使用 `cat`。 +不要根据编号猜测帖子路径；挂载使用服务端返回的规范路径。 +挂载点不存在时会创建一个仅当前用户可访问的目录；已存在时必须为空、属于当前用户且不可被其他用户写入。 + +正常退出使用 `Ctrl-C`。也可以在另一终端卸载： + +```sh +fusermount3 -u ~/msg # Linux +umount ~/msg # macOS +``` + +## 路径与内容 + +服务端声明为容器的资源是目录，具有内容修订的非容器资源以原始字节呈现： +Markdown 保留原文，二进制文件不转成 Base64。证书等没有内容修订的叶节点提供鉴权后的 JSON 投影。 +文件打开时固定修订，顺序读取和 seek 不会拼接两个版本；关闭后重新打开才会看到新修订。 + +这是 **msg 资源命名空间**，不是服务器磁盘、任意网页或数据库的镜像。 +Git 仓库、网页托管资源不会自动解包为 Git 工作树或发布目录；HTTP 虚拟页面和操作入口也不伪装成本地文件。 +当前仅支持 `--transport http`，其他 CLI 传输模式会明确拒绝。 + +## 权限、缓存和限制 + +挂载复用现有账号、私钥/令牌、OAuth 登录态、证书以及显式身份代理配置。 +服务端仍决定每次请求能读取什么；本地挂载不会获得额外权限。 +`stat`、目录分页和每次内容块读取均经过鉴权。原始文件读取关闭内核内容缓存和属性缓存， +权限被撤销后后续读取失败；已交付给本地程序的字节当然无法收回。 + +挂载只对当前系统用户开放，不提供 `allow_other`，并固定 `ro,nodev,nosuid,noexec`。 +没有持久化内容缓存，没有自动 ACK、关注、点赞、工具执行或交易。 +少量无修订 JSON 在已打开的文件句柄中保留快照，但每次读取仍重新鉴权。 + +内容按需读取，每次最多 64 KiB，同时最多打开 64 个文件。 +每个目录最多 10,000 个条目；超过时返回 `EOVERFLOW`，不会静默截断。 +可用 `--max-entries` 调整到最多 100,000。每个文件系统请求默认最多 45 秒， +可用 `--timeout` 设置为大于 0、最多 300 秒。断网和服务端错误会转成明确的文件系统错误， +不会用旧内容或空文件掩盖失败。 + +递归 `grep` 或文件管理器索引仍可能产生很多网络请求。建议挂载需要的子目录， +不要把整个站点交给自动备份、全文索引或杀毒扫描程序。 diff --git a/scripts/check_fuse_mount.py b/scripts/check_fuse_mount.py new file mode 100644 index 00000000..bb9ac6b9 --- /dev/null +++ b/scripts/check_fuse_mount.py @@ -0,0 +1,111 @@ +"""Exercise the real FUSE kernel boundary; missing native support is a CI failure. + +Only this smoke test supplies an in-memory backend. The production FUSE adapter, +mount flags, thread bridge, offset reads and unmount lifecycle remain unchanged. +""" + +import asyncio +import errno +import os +import shutil +import subprocess +import tempfile +from pathlib import Path + +from msg.client_mount import Node, ReadOnlyMount, io_error, serve_mount + +BODY = bytes(range(256)) * 1200 + + +class FixtureBackend: + async def node(self, path): + if path not in {'/', '/data.bin'}: + raise io_error(errno.ENOENT) + return Node({ + 'id': 'root' if path == '/' else 'file', 'path': path, + 'container': path == '/', 'revision': 'v_one', 'size': len(BODY), + }) + + async def directory(self, path): + if path != '/': + raise io_error(errno.ENOTDIR) + return ['.', '..', 'data.bin'] + + async def query(self, operation, arguments): + return {} + + async def read(self, node, size, offset): + return BODY[offset:offset + size] + + +def inspect_mount(path): + assert sorted(p.name for p in path.iterdir()) == ['data.bin'] + target = path / 'data.bin' + assert target.stat().st_size == len(BODY) + with target.open('rb') as stream: + assert stream.read() == BODY + stream.seek(12345) + assert stream.read(2345) == BODY[12345:14690] + for operation in (lambda: target.write_bytes(b'no'), target.unlink, lambda: (path / 'new').mkdir()): + try: + operation() + except OSError as exc: + assert exc.errno == errno.EROFS, exc + else: + raise AssertionError('read-only mount accepted a mutation') + assert target.read_bytes() == BODY + + +async def exercise(path, *, cancel=False): + import mfusepy + + operations = ReadOnlyMount(FixtureBackend(), asyncio.get_running_loop()) + task = asyncio.create_task(serve_mount(mfusepy, operations, path)) + try: + for _ in range(250): + if task.done(): + await task + raise AssertionError('FUSE exited before mounting') + if os.path.ismount(path): + break + await asyncio.sleep(0.02) + else: + raise AssertionError('FUSE mount did not appear') + await asyncio.wait_for(asyncio.to_thread(inspect_mount, path), 15) + if cancel: + task.cancel() + try: + await asyncio.wait_for(task, 10) + except asyncio.CancelledError: + pass + else: + executable = shutil.which('fusermount3') or shutil.which('fusermount') + if not executable: + raise AssertionError('fusermount is required by the Linux smoke test') + await asyncio.to_thread( + subprocess.run, [executable, '-u', '--', str(path)], check=True, timeout=10 + ) + await asyncio.wait_for(task, 10) + assert not os.path.ismount(path) + assert not operations.handles + finally: + if not task.done(): + task.cancel() + try: + await asyncio.wait_for(task, 10) + except asyncio.CancelledError: + pass + + +async def main(): + if not Path('/dev/fuse').exists(): + raise SystemExit('FUSE smoke requires /dev/fuse; this is not a skipped acceptance test.') + with tempfile.TemporaryDirectory(prefix='msg-fuse-') as directory: + path = Path(directory) + await exercise(path) + await exercise(path, cancel=True) + print('FUSE kernel smoke passed: list/stat/read/seek/read-only/unmount/cancellation') + + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/src/msg/client_mount.py b/src/msg/client_mount.py new file mode 100644 index 00000000..dac681c2 --- /dev/null +++ b/src/msg/client_mount.py @@ -0,0 +1,428 @@ +"""Read-only FUSE projection of authorized resources; never a website crawler. + +The SDK stays on its asyncio loop. Synchronous FUSE callbacks run in a worker +and submit bounded reads back to that loop. Native bindings are loaded only by +``mount`` so the normal client remains usable without libfuse. +""" + +from __future__ import annotations + +import asyncio +import errno +import hashlib +import os +import re +import stat +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +import httpx + +from msg.core.codec import b64, canonical, wire +from msg.core.errors import Failure, require +from msg.transports.client import HTTPTransport + +READ_BYTES = 65536 +MAX_ENTRIES = 10000 +MAX_HANDLES = 64 +_ID = re.compile(r'[A-Za-z0-9_.:-]{1,160}\Z') +_MISSING = {'not_found', 'resource_not_found', 'revision_not_found', 'resource_purged', 'ancestor_inactive'} +_DENIED = { + 'authentication_required', 'permission_denied', 'credential_ceiling', 'certificate_gate', + 'capability_required', 'tool_certificate_required', 'delegation_scope', 'request_expired', + 'credential_revoked', 'credential_expired', 'invalid_signature', 'invalid_credential', +} + + +def io_error(number): + """Expose errno, not remote response bodies, private URLs or credentials.""" + return OSError(number, os.strerror(number)) + + +def remote_error(code): + return io_error( + errno.ENOENT if code in _MISSING else errno.EACCES if code in _DENIED else errno.EIO + ) + + +def canonical_path(value): + require( + isinstance(value, str) + and value.startswith('/') + and '\x00' not in value + and '\\' not in value + and (value == '/' or all(part not in {'', '.', '..'} for part in value[1:].split('/'))), + 'invalid_mount_path', + ) + return value + + +def add_commands(commands): + command = commands.add_parser('mount', help='Mount authorized resources as a read-only folder.') + command.add_argument('mountpoint', type=Path) + command.add_argument('--path', default='/', help='Remote directory to mount (default: /).') + command.add_argument( + '--max-entries', type=int, default=MAX_ENTRIES, + help='Maximum entries per directory; larger listings fail explicitly (default: 10000).', + ) + command.add_argument( + '--timeout', type=float, default=45, + help='Deadline for each filesystem request in seconds (default: 45).', + ) + command.epilog = ( + 'Runs in the foreground; Ctrl-C unmounts. Requires msgctl[fuse] and system FUSE. ' + 'No writes, automatic ACKs, follows, or persistent content cache.' + ) + + +@dataclass(frozen=True) +class Node: + metadata: Mapping + projection: bytes | None = None + + @property + def directory(self): + return self.metadata['container'] + + @property + def size(self): + if self.directory: + return 0 + return len(self.projection) if self.projection is not None else self.metadata['size'] + + def attributes(self): + created = self.metadata.get('created_at') + modified = self.metadata.get('modified_at') or created + + def nanoseconds(value): + return int(datetime.fromisoformat(value).timestamp() * 1_000_000_000) if value else 0 + + return { + 'st_mode': (stat.S_IFDIR | 0o500) if self.directory else (stat.S_IFREG | 0o400), + 'st_nlink': 2 if self.directory else 1, + 'st_uid': os.geteuid(), + 'st_gid': os.getegid(), + 'st_size': self.size, + 'st_ino': int.from_bytes( + hashlib.blake2b(self.metadata['id'].encode(), digest_size=8).digest(), 'big' + ) & ((1 << 63) - 1) or 1, + 'st_atime': nanoseconds(modified), + 'st_mtime': nanoseconds(modified), + 'st_ctime': nanoseconds(created), + 'st_blksize': READ_BYTES, + 'st_blocks': (self.size + 511) // 512, + } + + +class MountBackend: + """All paths and listings pass through the existing authorization boundary.""" + + def __init__(self, client, root='/', *, max_entries=MAX_ENTRIES): + require(type(client.transport) is HTTPTransport, 'mount_http_transport_required') + require(type(max_entries) is int and 1 <= max_entries <= 100000, 'invalid_mount_limit') + self.client = client + self.root = canonical_path(root) + self.max_entries = max_entries + + def remote_path(self, path): + path = canonical_path(path) + return self.root if path == '/' else self.root.rstrip('/') + path + + async def query(self, operation, arguments): + # A closed allowlist prevents future filesystem helpers acquiring write effects. + require(operation in {'file.stat', 'discovery.get', 'discovery.read_query'}, 'mount_read_only') + description = await self.client.transport.description() + require(description['operations'].get(operation) == 'read', 'effect_mismatch') + result = await self.client.call(operation, arguments) + if result.status != 'ok': + raise remote_error(result.error.code if result.error is not None else '') + require(isinstance(result.data, Mapping), 'invalid_mount_response') + return result.data + + async def node(self, path=None, *, resource_id=None): + target = resource_id if resource_id is not None else self.remote_path(path) + data = await self.query('file.stat', {'id': target}) + require(type(data.get('container')) is bool, 'mount_server_upgrade_required') + require( + isinstance(data.get('id'), str) and _ID.fullmatch(data['id']) is not None, + 'invalid_mount_response', + ) + if data.get('state') in {'archived', 'purged'}: + raise io_error(errno.ENOENT) + projection = None + if not data['container']: + if data.get('revision'): + require( + isinstance(data['revision'], str) + and _ID.fullmatch(data['revision']) is not None + and type(data.get('size')) is int + and 0 <= data['size'] < 2**63, + 'invalid_mount_response', + ) + else: + # Certificates and other revisionless leaves have an authorized + # JSON representation, not a fabricated empty raw file. + projection = canonical(await self.query('discovery.get', {'id': data['id']})) + b'\n' + require(len(projection) <= 1048576, 'response_too_large') + return Node(data, projection) + + async def directory(self, path): + parent = await self.node(path) + if not parent.directory: + raise io_error(errno.ENOTDIR) + parent_path = canonical_path(parent.metadata['path']).rstrip('/') + '/' + first = {'parent': parent.metadata['id'], 'limit': 100, 'fields': ['id', 'path']} + arguments = first + names, seen, cursors = [], set(), set() + for _ in range(self.max_entries + 1): + page = await self.query('discovery.read_query', arguments) + require(isinstance(page.get('items'), (list, tuple)), 'invalid_mount_response') + for item in page['items']: + require(isinstance(item, Mapping), 'invalid_mount_response') + child = canonical_path(item.get('path')) + require(child.startswith(parent_path), 'invalid_mount_response') + name = child[len(parent_path):] + require( + name not in {'', '.', '..'} and '/' not in name and len(name.encode()) <= 255, + 'invalid_mount_response', + ) + require(name not in seen, 'invalid_mount_response') + if len(names) >= self.max_entries: + raise io_error(errno.EOVERFLOW) + names.append(name) + seen.add(name) + cursor = page.get('cursor') + if cursor is None: + return ['.', '..', *names] + require( + isinstance(cursor, str) and cursor and cursor not in cursors, + 'invalid_mount_response', + ) + cursors.add(cursor) + # Never follow server-provided next URLs; the opaque cursor returns + # through the same authenticated, principal-bound read operation. + arguments = {'cursor': cursor} + raise io_error(errno.EOVERFLOW) + + async def read(self, node, size, offset): + if node.directory: + raise io_error(errno.EISDIR) + if size < 0 or offset < 0: + raise io_error(errno.EINVAL) + length = min(size, READ_BYTES, max(0, node.size - offset)) + if node.projection is not None: + # Reauthorize even a previously opened revisionless JSON snapshot. + await self.query('file.stat', {'id': node.metadata['id']}) + return node.projection[offset:offset + length] + if length == 0: + return b'' + from msg.client_oauth import read_session, refresh + + client = self.client + if client.signer_override is None and read_session(client.state) is not None: + await refresh(client) + transport = client.transport + description = await transport.description() + require(description['operations'].get('discovery.raw') == 'read', 'effect_mismatch') + rid, revision = node.metadata['id'], node.metadata['revision'] + packet = client.prepare( + 'discovery.raw', {'id': rid, 'revision': revision, 'offset': offset, 'length': length} + ) + path = f'/_id/{rid}/revisions/{revision}/raw' + end = offset + length - 1 + async with transport.http.stream( + 'GET', transport.endpoint + path, + headers={ + 'X-Msg-Request': b64(canonical(wire(packet))), + 'Range': f'bytes={offset}-{end}', + 'Accept-Encoding': 'identity', + }, + follow_redirects=False, + ) as response: + if response.status_code in {401, 403}: + raise io_error(errno.EACCES) + if response.status_code in {404, 410}: + raise io_error(errno.ENOENT) + require( + response.status_code == 206 + and response.headers.get('content-range') == f'bytes {offset}-{end}/{node.size}' + and response.headers.get('content-length') == str(length) + and response.headers.get('content-encoding', 'identity') == 'identity', + 'invalid_mount_response', + ) + payload = bytearray() + async for chunk in response.aiter_raw(): + require(len(payload) + len(chunk) <= length, 'response_too_large') + payload.extend(chunk) + require(len(payload) == length, 'invalid_mount_response') + return bytes(payload) + + +class ReadOnlyMount: + """Path-based mfusepy adapter; no imports or kernel device needed in tests.""" + + use_ns = True + + def __init__(self, backend, loop, *, timeout=45): + self.backend, self.loop, self.timeout = backend, loop, timeout + self.handles = {} + self.next_handle = 1 + + def invoke(self, coroutine): + async def bounded(): + async with asyncio.timeout(self.timeout): + return await coroutine + + future = asyncio.run_coroutine_threadsafe(bounded(), self.loop) + try: + return future.result(self.timeout + 1) + except TimeoutError: + future.cancel() + raise io_error(errno.ETIMEDOUT) from None + except Failure as exc: + raise remote_error(exc.code) from None + except httpx.HTTPError: + raise io_error(errno.EIO) from None + + def getattr(self, path, fh=None): + # Paths always reauthorize; open descriptors keep their pinned size. + if fh is not None and fh in self.handles: + node = self.handles[fh] + self.invoke(self.backend.query('file.stat', {'id': node.metadata['id']})) + else: + node = self.invoke(self.backend.node(path)) + return node.attributes() + + def readdir(self, path, fh): + return self.invoke(self.backend.directory(path)) + + def access(self, path, mode): + if mode & os.W_OK: + raise io_error(errno.EROFS) + node = self.invoke(self.backend.node(path)) + if mode & os.X_OK and not node.directory: + raise io_error(errno.EACCES) + return 0 + + def open(self, path, flags): + if flags & os.O_ACCMODE != os.O_RDONLY or flags & (os.O_TRUNC | os.O_CREAT | os.O_APPEND): + raise io_error(errno.EROFS) + if len(self.handles) >= MAX_HANDLES: + raise io_error(errno.EMFILE) + node = self.invoke(self.backend.node(path)) + if node.directory: + raise io_error(errno.EISDIR) + handle = self.next_handle + self.next_handle += 1 + self.handles[handle] = node + return handle + + def read(self, path, size, offset, fh): + if fh not in self.handles: + raise io_error(errno.EBADF) + return self.invoke(self.backend.read(self.handles[fh], size, offset)) + + def release(self, path, fh): + self.handles.pop(fh, None) + return 0 + + def flush(self, path, fh): + return 0 + + def statfs(self, path): + return { + 'f_bsize': 4096, 'f_frsize': 4096, 'f_blocks': 0, 'f_bfree': 0, 'f_bavail': 0, + 'f_files': 0, 'f_ffree': 0, 'f_favail': 0, 'f_flag': os.ST_RDONLY, 'f_namemax': 255, + } + + def listxattr(self, path): + return [] + + def getxattr(self, path, name, position=0): + raise io_error(getattr(errno, 'ENOATTR', errno.ENODATA)) + + def readlink(self, path): + raise io_error(errno.EINVAL) + + def readonly(self, *args): + raise io_error(errno.EROFS) + + create = write = truncate = unlink = rmdir = mkdir = rename = readonly + chmod = chown = utimens = symlink = link = mknod = readonly + setxattr = removexattr = fallocate = readonly + + +def mountpoint_path(value): + path = Path(value).expanduser().absolute() + require(not path.is_symlink(), 'unsafe_mountpoint') + path.mkdir(mode=0o700, exist_ok=True) + info = path.lstat() + require( + stat.S_ISDIR(info.st_mode) and info.st_uid == os.geteuid() and info.st_mode & 0o022 == 0, + 'unsafe_mountpoint', + ) + require(not os.path.ismount(path), 'mountpoint_busy') + require(next(path.iterdir(), None) is None, 'mountpoint_not_empty') + return path.resolve() + + +async def run_mount(client, args): + import math + import sys + + require(sys.platform.startswith('linux') or sys.platform == 'darwin', 'mount_platform_unsupported') + require(math.isfinite(args.timeout) and 0 < args.timeout <= 300, 'invalid_mount_timeout') + try: + import mfusepy + except ImportError: + raise Failure('mount_dependency_missing', details={'install': 'uv tool install "msgctl[fuse]"'}) from None + except (OSError, AttributeError): + raise Failure('mount_system_fuse_missing', details={'install': 'Install fuse3 (Linux) or macFUSE (macOS).'}) from None + backend = MountBackend(client, args.path, max_entries=args.max_entries) + async with asyncio.timeout(args.timeout): + root = await backend.node('/') + require(root.directory, 'mount_root_not_directory') + mountpoint = mountpoint_path(args.mountpoint) + operations = ReadOnlyMount(backend, asyncio.get_running_loop(), timeout=args.timeout) + await serve_mount(mfusepy, operations, mountpoint) + return 0 + + +async def serve_mount(binding, operations, mountpoint): + """Keep native callbacks off the SDK loop; unmount on task cancellation.""" + import shutil + + task = asyncio.create_task(asyncio.to_thread( + binding.FUSE, operations, str(mountpoint), + foreground=True, nothreads=True, ro=True, nodev=True, nosuid=True, noexec=True, + default_permissions=True, direct_io=True, attr_timeout=0, entry_timeout=0, + negative_timeout=0, max_read=READ_BYTES, fsname='msg', + )) + try: + await asyncio.shield(task) + except asyncio.CancelledError: + # Native libfuse normally handles SIGINT itself. Cancellation from an + # embedding application must not leave its worker mounted indefinitely. + for _ in range(100): + if task.done() or os.path.ismount(mountpoint): + break + await asyncio.sleep(0.02) + if not task.done() and os.path.ismount(mountpoint): + executable = shutil.which('fusermount3') or shutil.which('fusermount') + command = ([executable, '-u', '--', str(mountpoint)] if executable + else ['umount', str(mountpoint)]) + process = await asyncio.create_subprocess_exec( + *command, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL + ) + await process.wait() + await asyncio.shield(task) + raise + except RuntimeError: + raise Failure( + 'mount_failed', + details={'hint': 'Check FUSE installation and mountpoint permissions.'}, + ) from None + finally: + operations.handles.clear() diff --git a/tests/test_client_mount.py b/tests/test_client_mount.py new file mode 100644 index 00000000..221f282c --- /dev/null +++ b/tests/test_client_mount.py @@ -0,0 +1,305 @@ +"""Mount contract tests do not import native bindings or require /dev/fuse.""" + +import argparse +import asyncio +import errno +import os +import stat +import sys +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import httpx +import pytest + +from msg.client_mount import ( + MAX_HANDLES, + READ_BYTES, + MountBackend, + Node, + ReadOnlyMount, + add_commands, + canonical_path, + mountpoint_path, + run_mount, +) +from msg.core.codec import canonical, loads, unb64 +from msg.core.errors import Failure +from msg.transports.client import HTTPTransport, PathGETTransport + + +def metadata(**changes): + return { + 'id': 'r_file', 'path': '/main/a.md', 'container': False, + 'revision': 'v_first', 'size': 100, 'state': 'active', + 'created_at': '2026-10-01T00:00:00Z', 'modified_at': '2026-10-01T01:00:00Z', + **changes, + } + + +@pytest.fixture +async def client(): + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as http: + transport = HTTPTransport('https://example.test', http=http) + transport._description = {'operations': dict.fromkeys( + ('file.stat', 'discovery.get', 'discovery.read_query', 'discovery.raw'), 'read' + )} + value = SimpleNamespace( + transport=transport, signer_override=object(), + call=AsyncMock(return_value=SimpleNamespace(status='ok', data=metadata(), error=None)), + prepare=lambda operation, arguments: {'operation': operation, 'arguments': arguments}, + ) + yield value + + +def ok(data): + return SimpleNamespace(status='ok', data=data, error=None) + + +@pytest.mark.parametrize('path', ['', 'main', '//', '/main/', '/./a', '/../a', '/a//b', '/a/../b', '/a\\b', '/\x00']) +def test_invalid_paths(path): + with pytest.raises(Failure, match='invalid_mount_path'): + canonical_path(path) + + +@pytest.mark.asyncio +async def test_mount_root_and_lazy_node(client): + backend = MountBackend(client, '/main') + assert backend.remote_path('/') == '/main' + assert backend.remote_path('/a.md') == '/main/a.md' + node = await backend.node('/a.md') + assert node.size == 100 + client.call.assert_awaited_once_with('file.stat', {'id': '/main/a.md'}) + attrs = node.attributes() + assert stat.S_ISREG(attrs['st_mode']) and attrs['st_mode'] & 0o777 == 0o400 + assert attrs['st_mtime'] > attrs['st_ctime'] > 0 + assert attrs['st_ino'] == Node(metadata()).attributes()['st_ino'] + assert attrs['st_uid'] == os.geteuid() + + +@pytest.mark.asyncio +async def test_authoritative_container_required(client): + value = metadata() + del value['container'] + client.call.return_value = ok(value) + with pytest.raises(Failure, match='mount_server_upgrade_required'): + await MountBackend(client).node('/') + + +@pytest.mark.asyncio +async def test_non_http_transport_rejected(client): + client.transport = PathGETTransport('https://example.test', http=client.transport.http) + with pytest.raises(Failure, match='mount_http_transport_required'): + MountBackend(client) + + +@pytest.mark.asyncio +async def test_closed_read_allowlist_and_effect_guard(client): + backend = MountBackend(client) + with pytest.raises(Failure, match='mount_read_only'): + await backend.query('file.delete', {'id': 'r_file'}) + client.transport._description['operations']['file.stat'] = 'transaction' + with pytest.raises(Failure, match='effect_mismatch'): + await backend.node('/') + client.call.assert_not_called() + + +@pytest.mark.asyncio +async def test_paginated_directory_uses_only_opaque_cursor(client): + client.call.side_effect = [ + ok(metadata(id='r_topic', path='/main', container=True, revision=None)), + ok({'items': [{'id': 'a', 'path': '/main/a.md'}], 'cursor': 'next', 'next': 'https://evil.test/'}), + ok({'items': [{'id': 'b', 'path': '/main/你好.md'}]}), + ] + assert await MountBackend(client, '/main').directory('/') == ['.', '..', 'a.md', '你好.md'] + assert client.call.await_args_list[1].args == ( + 'discovery.read_query', {'parent': 'r_topic', 'limit': 100, 'fields': ['id', 'path']} + ) + assert client.call.await_args_list[2].args == ('discovery.read_query', {'cursor': 'next'}) + + +@pytest.mark.parametrize('bad', ['/elsewhere/a', '/main/../secret', '/main/sub/a', '/main/', '/main/' + 'x' * 256]) +@pytest.mark.asyncio +async def test_directory_rejects_unsafe_children(client, bad): + client.call.side_effect = [ + ok(metadata(path='/main', container=True)), ok({'items': [{'path': bad}]}), + ] + with pytest.raises(Failure): + await MountBackend(client).directory('/main') + + +@pytest.mark.asyncio +async def test_directory_refuses_truncation_and_repeated_cursors(client): + root = ok(metadata(path='/main', container=True)) + client.call.side_effect = [root, ok({'items': [{'path': '/main/a'}, {'path': '/main/b'}]})] + with pytest.raises(OSError) as exc: + await MountBackend(client, max_entries=1).directory('/main') + assert exc.value.errno == errno.EOVERFLOW + client.call.side_effect = [root, ok({'items': [], 'cursor': 'c'}), ok({'items': [], 'cursor': 'c'})] + with pytest.raises(Failure, match='invalid_mount_response'): + await MountBackend(client).directory('/main') + + +@pytest.mark.asyncio +async def test_revisionless_projection_reauthorizes_every_read(client): + client.call.side_effect = [ + ok(metadata(revision=None)), ok({'id': 'r_file', 'type': 'certificate'}), + ok(metadata(revision=None)), + SimpleNamespace(status='error', error=SimpleNamespace(code='permission_denied')), + ] + backend = MountBackend(client) + node = await backend.node('/main/a.md') + assert node.projection == canonical({'id': 'r_file', 'type': 'certificate'}) + b'\n' + assert await backend.read(node, 20, 0) == node.projection[:20] + with pytest.raises(OSError) as exc: + await backend.read(node, 20, 0) + assert exc.value.errno == errno.EACCES + + +class BytesStream(httpx.AsyncByteStream): + def __init__(self, body): + self.body = body + + async def __aiter__(self): + yield self.body + + +@pytest.mark.asyncio +async def test_raw_random_read_is_signed_pinned_and_bounded(client): + captured = [] + body = bytes(range(256)) * 600 + + def handler(request): + captured.append(request) + packet = loads(unb64(request.headers['x-msg-request'])) + args = packet['arguments'] + offset, length = args['offset'], args['length'] + assert packet['operation'] == 'discovery.raw' + assert args['revision'] == 'v_first' + assert request.headers['range'] == f'bytes={offset}-{offset + length - 1}' + assert request.headers['accept-encoding'] == 'identity' + assert request.url.path == '/_id/r_file/revisions/v_first/raw' + return httpx.Response(206, headers={ + 'content-range': f'bytes {offset}-{offset + length - 1}/{len(body)}', + 'content-length': str(length), + }, stream=BytesStream(body[offset:offset + length])) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + client.transport.http = http + backend = MountBackend(client) + node = Node(metadata(size=len(body))) + assert await backend.read(node, 2**20, 29) == body[29:29 + READ_BYTES] + assert await backend.read(node, 4096, len(body) - 10) == body[-10:] + assert await backend.read(node, 4096, len(body)) == b'' + assert len(captured) == 2 + with pytest.raises(OSError) as exc: + await backend.read(node, 1, -1) + assert exc.value.errno == errno.EINVAL + + +@pytest.mark.parametrize('status,headers,body,expected', [ + (302, {'location': 'https://evil.test'}, b'', Failure), + (403, {}, b'secret', OSError), + (404, {}, b'', OSError), + (200, {}, b'whole object', Failure), + (206, {'content-range': 'bytes 1-9/100', 'content-length': '10'}, b'0123456789', Failure), + (206, {'content-range': 'bytes 0-9/100', 'content-length': '10'}, b'01234567890', Failure), + (206, {'content-range': 'bytes 0-9/100', 'content-length': '10'}, b'short', Failure), +]) +@pytest.mark.asyncio +async def test_raw_response_validation(client, status, headers, body, expected): + requests = [] + + def handler(request): + requests.append(request) + return httpx.Response(status, headers=headers, stream=BytesStream(body)) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + client.transport.http = http + with pytest.raises(expected): + await MountBackend(client).read(Node(metadata()), 10, 0) + assert len(requests) == 1 + + +@pytest.mark.asyncio +async def test_adapter_open_read_release_and_write_denial(client): + backend = MountBackend(client) + backend.read = AsyncMock(return_value=b'bytes') + mount = ReadOnlyMount(backend, asyncio.get_running_loop()) + for flags in (os.O_WRONLY, os.O_RDWR, os.O_RDONLY | os.O_TRUNC, os.O_RDONLY | os.O_APPEND): + with pytest.raises(OSError) as exc: + mount.open('/main/a.md', flags) + assert exc.value.errno == errno.EROFS + client.call.assert_not_called() + fh = await asyncio.to_thread(mount.open, '/main/a.md', os.O_RDONLY) + assert await asyncio.to_thread(mount.read, '/main/a.md', 5, 8, fh) == b'bytes' + backend.read.assert_awaited_once_with(Node(metadata()), 5, 8) + assert mount.release('/main/a.md', fh) == 0 + with pytest.raises(OSError) as exc: + mount.read('/main/a.md', 5, 8, fh) + assert exc.value.errno == errno.EBADF + for method in ('create', 'write', 'truncate', 'unlink', 'mkdir', 'rmdir', 'rename', 'chmod', 'chown', 'utimens', 'symlink', 'link', 'mknod', 'setxattr', 'removexattr', 'fallocate'): + with pytest.raises(OSError) as exc: + getattr(mount, method)('/main/a.md') + assert exc.value.errno == errno.EROFS + mount.handles = dict.fromkeys(range(MAX_HANDLES)) + with pytest.raises(OSError) as exc: + mount.open('/main/a.md', os.O_RDONLY) + assert exc.value.errno == errno.EMFILE + + +@pytest.mark.asyncio +async def test_callback_errors_are_sanitized_and_timeouts_cancel(client): + backend = MountBackend(client) + mount = ReadOnlyMount(backend, asyncio.get_running_loop(), timeout=0.02) + backend.node = AsyncMock(side_effect=Failure('permission_denied', details={'secret': 'TOKEN'})) + with pytest.raises(OSError) as exc: + await asyncio.to_thread(mount.getattr, '/private') + assert exc.value.errno == errno.EACCES and 'TOKEN' not in str(exc.value) + cancelled = asyncio.Event() + + async def slow(path): + try: + await asyncio.sleep(5) + finally: + cancelled.set() + + backend.node = slow + with pytest.raises(OSError) as exc: + await asyncio.to_thread(mount.getattr, '/slow') + assert exc.value.errno == errno.ETIMEDOUT + await asyncio.wait_for(cancelled.wait(), 1) + + +def test_mountpoint_must_be_owned_empty_directory(tmp_path): + target = tmp_path / 'new' + assert mountpoint_path(target) == target + assert target.stat().st_mode & 0o777 == 0o700 + (target / 'keep').write_text('do not hide') + with pytest.raises(Failure, match='mountpoint_not_empty'): + mountpoint_path(target) + link = tmp_path / 'link' + link.symlink_to(target) + with pytest.raises(Failure, match='unsafe_mountpoint'): + mountpoint_path(link) + target.chmod(0o777) + with pytest.raises(Failure, match='unsafe_mountpoint'): + mountpoint_path(target) + + +def test_mount_cli_does_not_import_optional_binding(): + parser = argparse.ArgumentParser() + add_commands(parser.add_subparsers(dest='command')) + args = parser.parse_args(['mount', '/tmp/msg-mount', '--path', '/main']) + assert args.command == 'mount' and args.path == '/main' + assert args.max_entries == 10000 and args.timeout == 45 + assert 'mfusepy' not in sys.modules + + +@pytest.mark.asyncio +async def test_missing_dependency_is_actionable_not_an_import_crash(client, monkeypatch, tmp_path): + monkeypatch.setitem(sys.modules, 'mfusepy', None) + args = SimpleNamespace(path='/', max_entries=10000, timeout=45, mountpoint=tmp_path / 'mount') + with pytest.raises(Failure, match='mount_dependency_missing'): + await run_mount(client, args) + assert not args.mountpoint.exists() diff --git a/tests/test_mount_service.py b/tests/test_mount_service.py new file mode 100644 index 00000000..a5b781b9 --- /dev/null +++ b/tests/test_mount_service.py @@ -0,0 +1,95 @@ +"""FUSE backend integration against the actual HTTP service and database ACLs.""" + +import errno + +import httpx +import pytest +from test_service import NOW + +from msg.client import ClientState, MsgClient +from msg.client_mount import MountBackend +from msg.core.codec import b64, wire +from msg.transports.client import HTTPTransport +from msg.transports.http import create_app + + +@pytest.mark.asyncio +async def test_mount_reads_raw_bytes_and_rechecks_revoked_access(installed, tmp_path): + app, _ = installed + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url + ) as http: + owner = MsgClient( + ClientState(tmp_path / 'owner', server=app.settings.service_url), + HTTPTransport(app.settings.service_url, http=http), clock=lambda: NOW, + ) + assert (await owner.register('mount-owner')).status == 'ok' + guest = MsgClient( + ClientState(tmp_path / 'guest', server=app.settings.service_url), + HTTPTransport(app.settings.service_url, http=http), clock=lambda: NOW, + ) + created = await owner.call('content.post_create', {'parent': '/main', 'body': '原始 Markdown\n'}) + assert created.status == 'ok', wire(created) + rid = created.resources[0].id + meta = await owner.call('file.stat', {'id': rid}) + assert meta.status == 'ok' and meta.data['container'] is False, wire(meta) + path = meta.data['path'] + mount = MountBackend(guest) + pinned = await mount.node(path) + assert await mount.read(pinned, 65536, 0) == '原始 Markdown\n'.encode() + assert await mount.read(pinned, 3, 1) == '原始 Markdown\n'.encode()[1:4] + root = await mount.node('/main') + assert root.directory + assert path.rsplit('/', 1)[-1] in await mount.directory('/main') + changed = await owner.call( + 'content.post_edit', + {'id': rid, 'expected_revision': pinned.metadata['revision'], 'body': 'new revision'}, + expected=((rid, created.data['generation']),), + ) + assert changed.status == 'ok', wire(changed) + assert await mount.read(pinned, 65536, 0) == '原始 Markdown\n'.encode() + current = await mount.node(path) + assert await mount.read(current, 65536, 0) == b'new revision' + private = await owner.call( + 'content.chmod', {'id': rid, 'mode': '0600'}, + expected=((rid, changed.data['generation']),), + ) + assert private.status == 'ok', wire(private) + for action in (mount.node(path), mount.read(pinned, 10, 0)): + with pytest.raises(OSError) as exc: + await action + assert exc.value.errno == errno.EACCES + assert path.rsplit('/', 1)[-1] not in await mount.directory('/main') + own = MountBackend(owner) + assert await own.read(await own.node(path), 65536, 0) == b'new revision' + # Reading the mount does not express user intent to ACK, react or follow. + async with app.metadata.transaction(write=False) as tx: + assert tx.one('SELECT COUNT(*) FROM reactions')[0] == 0 + assert tx.one('SELECT COUNT(*) FROM agent_follows')[0] == 0 + assert tx.one('SELECT COUNT(*) FROM watches')[0] == 0 + + +@pytest.mark.asyncio +async def test_mount_binary_empty_files_and_private_subtree(installed, tmp_path): + app, _ = installed + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url + ) as http: + client = MsgClient( + ClientState(tmp_path / 'owner', server=app.settings.service_url), + HTTPTransport(app.settings.service_url, http=http), clock=lambda: NOW, + ) + assert (await client.register('mount-binary')).status == 'ok' + parent = '/@mount-binary/files' + bodies = {'bytes.bin': bytes(range(256)) * 500, 'empty.txt': b''} + for name, body in bodies.items(): + created = await client.call('file.create', {'parent': parent, 'name': name, 'data': b64(body)}) + assert created.status == 'ok', wire(created) + backend = MountBackend(client, parent) + assert set(await backend.directory('/')) == {'.', '..', *bodies} + for name, body in bodies.items(): + node = await backend.node('/' + name) + assert node.size == len(body) + result = b''.join([await backend.read(node, 65536, offset) for offset in range(0, len(body), 65536)]) + assert result == body + assert await backend.read(node, 10, len(body)) == b'' From fceda057bb7f87096bb85453f99f30ef7601895c Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 2 Oct 2026 05:54:19 +0800 Subject: [PATCH 2/7] ci: quote-free branch guard for FUSE verification --- .github/workflows/apply-fuse-integration.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/apply-fuse-integration.yml b/.github/workflows/apply-fuse-integration.yml index 71cb779c..ebdb17ea 100644 --- a/.github/workflows/apply-fuse-integration.yml +++ b/.github/workflows/apply-fuse-integration.yml @@ -6,7 +6,7 @@ permissions: contents: write jobs: apply: - if: github.repository == 'TokenNotIncluded/msg' && github.event.head_commit.message == 'build: stage FUSE integration for cloud verification' + if: github.repository == 'TokenNotIncluded/msg' && github.ref == 'refs/heads/work/fuse-mount-20261001' runs-on: ubuntu-latest timeout-minutes: 10 steps: From 186799efeb3e8540261b74e106521ec10f438eec Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Thu, 1 Oct 2026 21:54:57 +0000 Subject: [PATCH 3/7] feat: add authenticated read-only FUSE mounting --- .fuse-integration.patch | 103 ----------------- .github/workflows/apply-fuse-integration.yml | 52 --------- pyproject.toml | 1 + scripts/check_fuse_mount.py | 15 ++- src/msg/cli.py | 7 ++ src/msg/client_mount.py | 110 ++++++++++++++----- src/msg/plugins/files.py | 4 +- tests/test_client_mount.py | 109 +++++++++++++----- tests/test_mount_service.py | 24 ++-- uv.lock | 15 ++- 10 files changed, 216 insertions(+), 224 deletions(-) delete mode 100644 .fuse-integration.patch delete mode 100644 .github/workflows/apply-fuse-integration.yml diff --git a/.fuse-integration.patch b/.fuse-integration.patch deleted file mode 100644 index 11293de6..00000000 --- a/.fuse-integration.patch +++ /dev/null @@ -1,103 +0,0 @@ -diff --git a/pyproject.toml b/pyproject.toml -index 9d926d9..503df93 100644 ---- a/pyproject.toml -+++ b/pyproject.toml -@@ -25,6 +25,7 @@ msg = "msg.cli:main" - msgd = "msg.daemon:main" - - [project.optional-dependencies] -+fuse = ["mfusepy>=3.1.1,<4"] - server = [ - "starlette>=0.50,<2", - "uvicorn>=0.38,<1", -diff --git a/src/msg/cli.py b/src/msg/cli.py -index 3156250..8af61c6 100644 ---- a/src/msg/cli.py -+++ b/src/msg/cli.py -@@ -507,6 +507,9 @@ def parser(): - from msg.client_yubikey import add_commands as add_yubikey_commands - - add_yubikey_commands(commands) -+ from msg.client_mount import add_commands as add_mount_commands -+ -+ add_mount_commands(commands) - return cli - - -@@ -581,6 +584,10 @@ async def run(args): - require(args.handle == args.user, 'connection_user_mismatch') - elif command != 'login': - await client.require_username(args.user) -+ if command == 'mount': -+ from msg.client_mount import run_mount -+ -+ return await run_mount(client, args) - if command in {'login', 'logout', 'auth'}: - from msg.client_oauth import run_command - -diff --git a/src/msg/plugins/files.py b/src/msg/plugins/files.py -index f56bedb..018fe3d 100644 ---- a/src/msg/plugins/files.py -+++ b/src/msg/plugins/files.py -@@ -43,9 +43,13 @@ def install(app): - @op('file.stat', obj({'id': IDENTIFIER}, ('id',)), effect='read') - async def stat(ctx, request, tx): - spec = app.registry.operation('discovery.get') -- return await spec.handler( -+ result = await spec.handler( - ctx, replace(request, arguments={**request.arguments, 'view': 'meta'}), tx - ) -+ resource_type = app.registry.resource_type( -+ result.data['type'], result.data['type_version'] -+ ) -+ return replace(result, data={**result.data, 'container': resource_type.container}) - - @op( - 'file.write', -diff --git a/uv.lock b/uv.lock -index 7da9be3..b47431a 100644 ---- a/uv.lock -+++ b/uv.lock -@@ -471,6 +471,15 @@ wheels = [ - { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, - ] - -+[[package]] -+name = "mfusepy" -+version = "3.1.1" -+source = { registry = "https://pypi.org/simple" } -+sdist = { url = "https://files.pythonhosted.org/packages/91/47/746287c8962274f73ee25edb3840d80899464bfffbe2c435424c2d60a071/mfusepy-3.1.1.tar.gz", hash = "sha256:338ece54513d7d1a5e9492837679a0c7432ecf96a03490a2683a1ce1d19570e1", size = 34549, upload-time = "2026-03-13T00:36:52.636Z" } -+wheels = [ -+ { url = "https://files.pythonhosted.org/packages/c5/d5/56bc9326bf75f7ea0f4fe1178f5c9f20d1a1e15e288ecf66088513538ccd/mfusepy-3.1.1-py3-none-any.whl", hash = "sha256:69fb70cfc7f7cce595e6ff586f8451d8298f01f18286a157f24f564b24ec5a37", size = 28264, upload-time = "2026-03-13T00:36:51.843Z" }, -+] -+ - [[package]] - name = "more-itertools" - version = "11.1.0" -@@ -509,6 +518,9 @@ dev = [ - { name = "uvicorn" }, - { name = "valkey" }, - ] -+fuse = [ -+ { name = "mfusepy" }, -+] - server = [ - { name = "aiohttp" }, - { name = "dnspython" }, -@@ -530,6 +542,7 @@ requires-dist = [ - { name = "httpx", specifier = ">=0.28,<1" }, - { name = "jsonschema", specifier = ">=4.25,<5" }, - { name = "markdown-it-py", marker = "extra == 'server'", specifier = ">=4,<5" }, -+ { name = "mfusepy", marker = "extra == 'fuse'", specifier = ">=3.1.1,<4" }, - { name = "msgctl", extras = ["server"], marker = "extra == 'dev'" }, - { name = "psycopg", extras = ["binary"], marker = "extra == 'server'", specifier = ">=3.3,<4" }, - { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.4" }, -@@ -543,7 +556,7 @@ requires-dist = [ - { name = "valkey", marker = "extra == 'server'", specifier = ">=6.1,<7" }, - { name = "yubikey-manager", specifier = ">=5.8,<6" }, - ] --provides-extras = ["server", "dev"] -+provides-extras = ["fuse", "server", "dev"] - - [[package]] - name = "multidict" diff --git a/.github/workflows/apply-fuse-integration.yml b/.github/workflows/apply-fuse-integration.yml deleted file mode 100644 index ebdb17ea..00000000 --- a/.github/workflows/apply-fuse-integration.yml +++ /dev/null @@ -1,52 +0,0 @@ -name: Apply reviewed FUSE integration -on: - push: - branches: [work/fuse-mount-20261001] -permissions: - contents: write -jobs: - apply: - if: github.repository == 'TokenNotIncluded/msg' && github.ref == 'refs/heads/work/fuse-mount-20261001' - runs-on: ubuntu-latest - timeout-minutes: 10 - steps: - - uses: actions/checkout@v7 - with: - fetch-depth: 0 - - uses: actions/setup-python@v7 - with: - python-version: '3.15' - allow-prereleases: true - - name: Apply the reviewed local patch without replacing concurrent changes - run: | - git apply --check .fuse-integration.patch - git apply .fuse-integration.patch - sudo apt-get update -qq - sudo env NEEDRESTART_MODE=l apt-get install -y -qq fuse3 libpcsclite-dev swig - python -m pip install uv==0.12.20 - uv sync --locked --extra dev --extra fuse --python "$(command -v python)" - .venv/bin/ruff check --fix src/msg/client_mount.py src/msg/cli.py src/msg/plugins/files.py tests/test_client_mount.py tests/test_mount_service.py scripts/check_fuse_mount.py - .venv/bin/ruff format src/msg/client_mount.py src/msg/cli.py src/msg/plugins/files.py tests/test_client_mount.py tests/test_mount_service.py scripts/check_fuse_mount.py - - name: Commit complete implementation and remove this one-off staging workflow - run: | - git rm .fuse-integration.patch .github/workflows/apply-fuse-integration.yml - git add pyproject.toml uv.lock src/msg/client_mount.py src/msg/cli.py src/msg/plugins/files.py tests/test_client_mount.py tests/test_mount_service.py scripts/check_fuse_mount.py - git config user.name 'github-actions[bot]' - git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - git commit -m 'feat: add authenticated read-only FUSE mounting' - git push origin HEAD:refs/heads/work/fuse-mount-20261001 - - name: Test mount contracts on Python 3.15 - run: .venv/bin/python -m pytest tests/test_client_mount.py --junitxml=artifacts/fuse-unit.xml - - name: Exercise actual Linux FUSE mount and cancellation - run: .venv/bin/python scripts/check_fuse_mount.py - - name: Export final reviewed delta - if: always() - run: | - mkdir -p artifacts - git diff c6be2080307a1a067c2117b9791e3a7a995a08de -- . ':!.fuse-integration.patch' ':!.github/workflows/apply-fuse-integration.yml' > artifacts/msg-fuse.patch - git rev-parse HEAD > artifacts/commit.txt - - uses: actions/upload-artifact@v7.0.1 - if: always() - with: - name: fuse-verification - path: artifacts/ diff --git a/pyproject.toml b/pyproject.toml index fad6c328..b223aa3b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ msg = "msg.cli:main" msgd = "msg.daemon:main" [project.optional-dependencies] +fuse = ["mfusepy>=3.1.1,<4"] server = [ "starlette>=0.50,<2", "uvicorn>=0.38,<1", diff --git a/scripts/check_fuse_mount.py b/scripts/check_fuse_mount.py index bb9ac6b9..304cb6d5 100644 --- a/scripts/check_fuse_mount.py +++ b/scripts/check_fuse_mount.py @@ -22,8 +22,11 @@ async def node(self, path): if path not in {'/', '/data.bin'}: raise io_error(errno.ENOENT) return Node({ - 'id': 'root' if path == '/' else 'file', 'path': path, - 'container': path == '/', 'revision': 'v_one', 'size': len(BODY), + 'id': 'root' if path == '/' else 'file', + 'path': path, + 'container': path == '/', + 'revision': 'v_one', + 'size': len(BODY), }) async def directory(self, path): @@ -35,7 +38,7 @@ async def query(self, operation, arguments): return {} async def read(self, node, size, offset): - return BODY[offset:offset + size] + return BODY[offset : offset + size] def inspect_mount(path): @@ -46,7 +49,11 @@ def inspect_mount(path): assert stream.read() == BODY stream.seek(12345) assert stream.read(2345) == BODY[12345:14690] - for operation in (lambda: target.write_bytes(b'no'), target.unlink, lambda: (path / 'new').mkdir()): + for operation in ( + lambda: target.write_bytes(b'no'), + target.unlink, + lambda: (path / 'new').mkdir(), + ): try: operation() except OSError as exc: diff --git a/src/msg/cli.py b/src/msg/cli.py index 3156250e..8af61c6a 100644 --- a/src/msg/cli.py +++ b/src/msg/cli.py @@ -507,6 +507,9 @@ def parser(): from msg.client_yubikey import add_commands as add_yubikey_commands add_yubikey_commands(commands) + from msg.client_mount import add_commands as add_mount_commands + + add_mount_commands(commands) return cli @@ -581,6 +584,10 @@ async def run(args): require(args.handle == args.user, 'connection_user_mismatch') elif command != 'login': await client.require_username(args.user) + if command == 'mount': + from msg.client_mount import run_mount + + return await run_mount(client, args) if command in {'login', 'logout', 'auth'}: from msg.client_oauth import run_command diff --git a/src/msg/client_mount.py b/src/msg/client_mount.py index dac681c2..e0e1255b 100644 --- a/src/msg/client_mount.py +++ b/src/msg/client_mount.py @@ -28,11 +28,26 @@ MAX_ENTRIES = 10000 MAX_HANDLES = 64 _ID = re.compile(r'[A-Za-z0-9_.:-]{1,160}\Z') -_MISSING = {'not_found', 'resource_not_found', 'revision_not_found', 'resource_purged', 'ancestor_inactive'} +_MISSING = { + 'not_found', + 'resource_not_found', + 'revision_not_found', + 'resource_purged', + 'ancestor_inactive', +} _DENIED = { - 'authentication_required', 'permission_denied', 'credential_ceiling', 'certificate_gate', - 'capability_required', 'tool_certificate_required', 'delegation_scope', 'request_expired', - 'credential_revoked', 'credential_expired', 'invalid_signature', 'invalid_credential', + 'authentication_required', + 'permission_denied', + 'credential_ceiling', + 'certificate_gate', + 'capability_required', + 'tool_certificate_required', + 'delegation_scope', + 'request_expired', + 'credential_revoked', + 'credential_expired', + 'invalid_signature', + 'invalid_credential', } @@ -64,11 +79,15 @@ def add_commands(commands): command.add_argument('mountpoint', type=Path) command.add_argument('--path', default='/', help='Remote directory to mount (default: /).') command.add_argument( - '--max-entries', type=int, default=MAX_ENTRIES, + '--max-entries', + type=int, + default=MAX_ENTRIES, help='Maximum entries per directory; larger listings fail explicitly (default: 10000).', ) command.add_argument( - '--timeout', type=float, default=45, + '--timeout', + type=float, + default=45, help='Deadline for each filesystem request in seconds (default: 45).', ) command.epilog = ( @@ -107,7 +126,9 @@ def nanoseconds(value): 'st_size': self.size, 'st_ino': int.from_bytes( hashlib.blake2b(self.metadata['id'].encode(), digest_size=8).digest(), 'big' - ) & ((1 << 63) - 1) or 1, + ) + & ((1 << 63) - 1) + or 1, 'st_atime': nanoseconds(modified), 'st_mtime': nanoseconds(modified), 'st_ctime': nanoseconds(created), @@ -132,7 +153,9 @@ def remote_path(self, path): async def query(self, operation, arguments): # A closed allowlist prevents future filesystem helpers acquiring write effects. - require(operation in {'file.stat', 'discovery.get', 'discovery.read_query'}, 'mount_read_only') + require( + operation in {'file.stat', 'discovery.get', 'discovery.read_query'}, 'mount_read_only' + ) description = await self.client.transport.description() require(description['operations'].get(operation) == 'read', 'effect_mismatch') result = await self.client.call(operation, arguments) @@ -164,7 +187,9 @@ async def node(self, path=None, *, resource_id=None): else: # Certificates and other revisionless leaves have an authorized # JSON representation, not a fabricated empty raw file. - projection = canonical(await self.query('discovery.get', {'id': data['id']})) + b'\n' + projection = ( + canonical(await self.query('discovery.get', {'id': data['id']})) + b'\n' + ) require(len(projection) <= 1048576, 'response_too_large') return Node(data, projection) @@ -183,7 +208,7 @@ async def directory(self, path): require(isinstance(item, Mapping), 'invalid_mount_response') child = canonical_path(item.get('path')) require(child.startswith(parent_path), 'invalid_mount_response') - name = child[len(parent_path):] + name = child[len(parent_path) :] require( name not in {'', '.', '..'} and '/' not in name and len(name.encode()) <= 255, 'invalid_mount_response', @@ -215,7 +240,7 @@ async def read(self, node, size, offset): if node.projection is not None: # Reauthorize even a previously opened revisionless JSON snapshot. await self.query('file.stat', {'id': node.metadata['id']}) - return node.projection[offset:offset + length] + return node.projection[offset : offset + length] if length == 0: return b'' from msg.client_oauth import read_session, refresh @@ -233,7 +258,8 @@ async def read(self, node, size, offset): path = f'/_id/{rid}/revisions/{revision}/raw' end = offset + length - 1 async with transport.http.stream( - 'GET', transport.endpoint + path, + 'GET', + transport.endpoint + path, headers={ 'X-Msg-Request': b64(canonical(wire(packet))), 'Range': f'bytes={offset}-{end}', @@ -333,8 +359,16 @@ def flush(self, path, fh): def statfs(self, path): return { - 'f_bsize': 4096, 'f_frsize': 4096, 'f_blocks': 0, 'f_bfree': 0, 'f_bavail': 0, - 'f_files': 0, 'f_ffree': 0, 'f_favail': 0, 'f_flag': os.ST_RDONLY, 'f_namemax': 255, + 'f_bsize': 4096, + 'f_frsize': 4096, + 'f_blocks': 0, + 'f_bfree': 0, + 'f_bavail': 0, + 'f_files': 0, + 'f_ffree': 0, + 'f_favail': 0, + 'f_flag': os.ST_RDONLY, + 'f_namemax': 255, } def listxattr(self, path): @@ -372,14 +406,21 @@ async def run_mount(client, args): import math import sys - require(sys.platform.startswith('linux') or sys.platform == 'darwin', 'mount_platform_unsupported') + require( + sys.platform.startswith('linux') or sys.platform == 'darwin', 'mount_platform_unsupported' + ) require(math.isfinite(args.timeout) and 0 < args.timeout <= 300, 'invalid_mount_timeout') try: import mfusepy except ImportError: - raise Failure('mount_dependency_missing', details={'install': 'uv tool install "msgctl[fuse]"'}) from None - except (OSError, AttributeError): - raise Failure('mount_system_fuse_missing', details={'install': 'Install fuse3 (Linux) or macFUSE (macOS).'}) from None + raise Failure( + 'mount_dependency_missing', details={'install': 'uv tool install "msgctl[fuse]"'} + ) from None + except OSError, AttributeError: + raise Failure( + 'mount_system_fuse_missing', + details={'install': 'Install fuse3 (Linux) or macFUSE (macOS).'}, + ) from None backend = MountBackend(client, args.path, max_entries=args.max_entries) async with asyncio.timeout(args.timeout): root = await backend.node('/') @@ -394,12 +435,26 @@ async def serve_mount(binding, operations, mountpoint): """Keep native callbacks off the SDK loop; unmount on task cancellation.""" import shutil - task = asyncio.create_task(asyncio.to_thread( - binding.FUSE, operations, str(mountpoint), - foreground=True, nothreads=True, ro=True, nodev=True, nosuid=True, noexec=True, - default_permissions=True, direct_io=True, attr_timeout=0, entry_timeout=0, - negative_timeout=0, max_read=READ_BYTES, fsname='msg', - )) + task = asyncio.create_task( + asyncio.to_thread( + binding.FUSE, + operations, + str(mountpoint), + foreground=True, + nothreads=True, + ro=True, + nodev=True, + nosuid=True, + noexec=True, + default_permissions=True, + direct_io=True, + attr_timeout=0, + entry_timeout=0, + negative_timeout=0, + max_read=READ_BYTES, + fsname='msg', + ) + ) try: await asyncio.shield(task) except asyncio.CancelledError: @@ -411,8 +466,11 @@ async def serve_mount(binding, operations, mountpoint): await asyncio.sleep(0.02) if not task.done() and os.path.ismount(mountpoint): executable = shutil.which('fusermount3') or shutil.which('fusermount') - command = ([executable, '-u', '--', str(mountpoint)] if executable - else ['umount', str(mountpoint)]) + command = ( + [executable, '-u', '--', str(mountpoint)] + if executable + else ['umount', str(mountpoint)] + ) process = await asyncio.create_subprocess_exec( *command, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL ) diff --git a/src/msg/plugins/files.py b/src/msg/plugins/files.py index f56bedbc..86532db0 100644 --- a/src/msg/plugins/files.py +++ b/src/msg/plugins/files.py @@ -43,9 +43,11 @@ def alias(name, target, version=1): @op('file.stat', obj({'id': IDENTIFIER}, ('id',)), effect='read') async def stat(ctx, request, tx): spec = app.registry.operation('discovery.get') - return await spec.handler( + result = await spec.handler( ctx, replace(request, arguments={**request.arguments, 'view': 'meta'}), tx ) + resource_type = app.registry.resource_type(result.data['type'], result.data['type_version']) + return replace(result, data={**result.data, 'container': resource_type.container}) @op( 'file.write', diff --git a/tests/test_client_mount.py b/tests/test_client_mount.py index 221f282c..7748b4f7 100644 --- a/tests/test_client_mount.py +++ b/tests/test_client_mount.py @@ -30,22 +30,32 @@ def metadata(**changes): return { - 'id': 'r_file', 'path': '/main/a.md', 'container': False, - 'revision': 'v_first', 'size': 100, 'state': 'active', - 'created_at': '2026-10-01T00:00:00Z', 'modified_at': '2026-10-01T01:00:00Z', + 'id': 'r_file', + 'path': '/main/a.md', + 'container': False, + 'revision': 'v_first', + 'size': 100, + 'state': 'active', + 'created_at': '2026-10-01T00:00:00Z', + 'modified_at': '2026-10-01T01:00:00Z', **changes, } @pytest.fixture async def client(): - async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as http: + async with httpx.AsyncClient( + transport=httpx.MockTransport(lambda _: httpx.Response(500)) + ) as http: transport = HTTPTransport('https://example.test', http=http) - transport._description = {'operations': dict.fromkeys( - ('file.stat', 'discovery.get', 'discovery.read_query', 'discovery.raw'), 'read' - )} + transport._description = { + 'operations': dict.fromkeys( + ('file.stat', 'discovery.get', 'discovery.read_query', 'discovery.raw'), 'read' + ) + } value = SimpleNamespace( - transport=transport, signer_override=object(), + transport=transport, + signer_override=object(), call=AsyncMock(return_value=SimpleNamespace(status='ok', data=metadata(), error=None)), prepare=lambda operation, arguments: {'operation': operation, 'arguments': arguments}, ) @@ -56,7 +66,9 @@ def ok(data): return SimpleNamespace(status='ok', data=data, error=None) -@pytest.mark.parametrize('path', ['', 'main', '//', '/main/', '/./a', '/../a', '/a//b', '/a/../b', '/a\\b', '/\x00']) +@pytest.mark.parametrize( + 'path', ['', 'main', '//', '/main/', '/./a', '/../a', '/a//b', '/a/../b', '/a\\b', '/\x00'] +) def test_invalid_paths(path): with pytest.raises(Failure, match='invalid_mount_path'): canonical_path(path) @@ -108,21 +120,29 @@ async def test_closed_read_allowlist_and_effect_guard(client): async def test_paginated_directory_uses_only_opaque_cursor(client): client.call.side_effect = [ ok(metadata(id='r_topic', path='/main', container=True, revision=None)), - ok({'items': [{'id': 'a', 'path': '/main/a.md'}], 'cursor': 'next', 'next': 'https://evil.test/'}), + ok({ + 'items': [{'id': 'a', 'path': '/main/a.md'}], + 'cursor': 'next', + 'next': 'https://evil.test/', + }), ok({'items': [{'id': 'b', 'path': '/main/你好.md'}]}), ] assert await MountBackend(client, '/main').directory('/') == ['.', '..', 'a.md', '你好.md'] assert client.call.await_args_list[1].args == ( - 'discovery.read_query', {'parent': 'r_topic', 'limit': 100, 'fields': ['id', 'path']} + 'discovery.read_query', + {'parent': 'r_topic', 'limit': 100, 'fields': ['id', 'path']}, ) assert client.call.await_args_list[2].args == ('discovery.read_query', {'cursor': 'next'}) -@pytest.mark.parametrize('bad', ['/elsewhere/a', '/main/../secret', '/main/sub/a', '/main/', '/main/' + 'x' * 256]) +@pytest.mark.parametrize( + 'bad', ['/elsewhere/a', '/main/../secret', '/main/sub/a', '/main/', '/main/' + 'x' * 256] +) @pytest.mark.asyncio async def test_directory_rejects_unsafe_children(client, bad): client.call.side_effect = [ - ok(metadata(path='/main', container=True)), ok({'items': [{'path': bad}]}), + ok(metadata(path='/main', container=True)), + ok({'items': [{'path': bad}]}), ] with pytest.raises(Failure): await MountBackend(client).directory('/main') @@ -135,7 +155,11 @@ async def test_directory_refuses_truncation_and_repeated_cursors(client): with pytest.raises(OSError) as exc: await MountBackend(client, max_entries=1).directory('/main') assert exc.value.errno == errno.EOVERFLOW - client.call.side_effect = [root, ok({'items': [], 'cursor': 'c'}), ok({'items': [], 'cursor': 'c'})] + client.call.side_effect = [ + root, + ok({'items': [], 'cursor': 'c'}), + ok({'items': [], 'cursor': 'c'}), + ] with pytest.raises(Failure, match='invalid_mount_response'): await MountBackend(client).directory('/main') @@ -143,7 +167,8 @@ async def test_directory_refuses_truncation_and_repeated_cursors(client): @pytest.mark.asyncio async def test_revisionless_projection_reauthorizes_every_read(client): client.call.side_effect = [ - ok(metadata(revision=None)), ok({'id': 'r_file', 'type': 'certificate'}), + ok(metadata(revision=None)), + ok({'id': 'r_file', 'type': 'certificate'}), ok(metadata(revision=None)), SimpleNamespace(status='error', error=SimpleNamespace(code='permission_denied')), ] @@ -179,16 +204,20 @@ def handler(request): assert request.headers['range'] == f'bytes={offset}-{offset + length - 1}' assert request.headers['accept-encoding'] == 'identity' assert request.url.path == '/_id/r_file/revisions/v_first/raw' - return httpx.Response(206, headers={ - 'content-range': f'bytes {offset}-{offset + length - 1}/{len(body)}', - 'content-length': str(length), - }, stream=BytesStream(body[offset:offset + length])) + return httpx.Response( + 206, + headers={ + 'content-range': f'bytes {offset}-{offset + length - 1}/{len(body)}', + 'content-length': str(length), + }, + stream=BytesStream(body[offset : offset + length]), + ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: client.transport.http = http backend = MountBackend(client) node = Node(metadata(size=len(body))) - assert await backend.read(node, 2**20, 29) == body[29:29 + READ_BYTES] + assert await backend.read(node, 2**20, 29) == body[29 : 29 + READ_BYTES] assert await backend.read(node, 4096, len(body) - 10) == body[-10:] assert await backend.read(node, 4096, len(body)) == b'' assert len(captured) == 2 @@ -197,15 +226,18 @@ def handler(request): assert exc.value.errno == errno.EINVAL -@pytest.mark.parametrize('status,headers,body,expected', [ - (302, {'location': 'https://evil.test'}, b'', Failure), - (403, {}, b'secret', OSError), - (404, {}, b'', OSError), - (200, {}, b'whole object', Failure), - (206, {'content-range': 'bytes 1-9/100', 'content-length': '10'}, b'0123456789', Failure), - (206, {'content-range': 'bytes 0-9/100', 'content-length': '10'}, b'01234567890', Failure), - (206, {'content-range': 'bytes 0-9/100', 'content-length': '10'}, b'short', Failure), -]) +@pytest.mark.parametrize( + 'status,headers,body,expected', + [ + (302, {'location': 'https://evil.test'}, b'', Failure), + (403, {}, b'secret', OSError), + (404, {}, b'', OSError), + (200, {}, b'whole object', Failure), + (206, {'content-range': 'bytes 1-9/100', 'content-length': '10'}, b'0123456789', Failure), + (206, {'content-range': 'bytes 0-9/100', 'content-length': '10'}, b'01234567890', Failure), + (206, {'content-range': 'bytes 0-9/100', 'content-length': '10'}, b'short', Failure), + ], +) @pytest.mark.asyncio async def test_raw_response_validation(client, status, headers, body, expected): requests = [] @@ -238,7 +270,24 @@ async def test_adapter_open_read_release_and_write_denial(client): with pytest.raises(OSError) as exc: mount.read('/main/a.md', 5, 8, fh) assert exc.value.errno == errno.EBADF - for method in ('create', 'write', 'truncate', 'unlink', 'mkdir', 'rmdir', 'rename', 'chmod', 'chown', 'utimens', 'symlink', 'link', 'mknod', 'setxattr', 'removexattr', 'fallocate'): + for method in ( + 'create', + 'write', + 'truncate', + 'unlink', + 'mkdir', + 'rmdir', + 'rename', + 'chmod', + 'chown', + 'utimens', + 'symlink', + 'link', + 'mknod', + 'setxattr', + 'removexattr', + 'fallocate', + ): with pytest.raises(OSError) as exc: getattr(mount, method)('/main/a.md') assert exc.value.errno == errno.EROFS diff --git a/tests/test_mount_service.py b/tests/test_mount_service.py index a5b781b9..06cbde09 100644 --- a/tests/test_mount_service.py +++ b/tests/test_mount_service.py @@ -21,14 +21,18 @@ async def test_mount_reads_raw_bytes_and_rechecks_revoked_access(installed, tmp_ ) as http: owner = MsgClient( ClientState(tmp_path / 'owner', server=app.settings.service_url), - HTTPTransport(app.settings.service_url, http=http), clock=lambda: NOW, + HTTPTransport(app.settings.service_url, http=http), + clock=lambda: NOW, ) assert (await owner.register('mount-owner')).status == 'ok' guest = MsgClient( ClientState(tmp_path / 'guest', server=app.settings.service_url), - HTTPTransport(app.settings.service_url, http=http), clock=lambda: NOW, + HTTPTransport(app.settings.service_url, http=http), + clock=lambda: NOW, + ) + created = await owner.call( + 'content.post_create', {'parent': '/main', 'body': '原始 Markdown\n'} ) - created = await owner.call('content.post_create', {'parent': '/main', 'body': '原始 Markdown\n'}) assert created.status == 'ok', wire(created) rid = created.resources[0].id meta = await owner.call('file.stat', {'id': rid}) @@ -51,7 +55,8 @@ async def test_mount_reads_raw_bytes_and_rechecks_revoked_access(installed, tmp_ current = await mount.node(path) assert await mount.read(current, 65536, 0) == b'new revision' private = await owner.call( - 'content.chmod', {'id': rid, 'mode': '0600'}, + 'content.chmod', + {'id': rid, 'mode': '0600'}, expected=((rid, changed.data['generation']),), ) assert private.status == 'ok', wire(private) @@ -77,19 +82,24 @@ async def test_mount_binary_empty_files_and_private_subtree(installed, tmp_path) ) as http: client = MsgClient( ClientState(tmp_path / 'owner', server=app.settings.service_url), - HTTPTransport(app.settings.service_url, http=http), clock=lambda: NOW, + HTTPTransport(app.settings.service_url, http=http), + clock=lambda: NOW, ) assert (await client.register('mount-binary')).status == 'ok' parent = '/@mount-binary/files' bodies = {'bytes.bin': bytes(range(256)) * 500, 'empty.txt': b''} for name, body in bodies.items(): - created = await client.call('file.create', {'parent': parent, 'name': name, 'data': b64(body)}) + created = await client.call( + 'file.create', {'parent': parent, 'name': name, 'data': b64(body)} + ) assert created.status == 'ok', wire(created) backend = MountBackend(client, parent) assert set(await backend.directory('/')) == {'.', '..', *bodies} for name, body in bodies.items(): node = await backend.node('/' + name) assert node.size == len(body) - result = b''.join([await backend.read(node, 65536, offset) for offset in range(0, len(body), 65536)]) + result = b''.join([ + await backend.read(node, 65536, offset) for offset in range(0, len(body), 65536) + ]) assert result == body assert await backend.read(node, 10, len(body)) == b'' diff --git a/uv.lock b/uv.lock index 7da9be37..b47431ab 100644 --- a/uv.lock +++ b/uv.lock @@ -471,6 +471,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, ] +[[package]] +name = "mfusepy" +version = "3.1.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/91/47/746287c8962274f73ee25edb3840d80899464bfffbe2c435424c2d60a071/mfusepy-3.1.1.tar.gz", hash = "sha256:338ece54513d7d1a5e9492837679a0c7432ecf96a03490a2683a1ce1d19570e1", size = 34549, upload-time = "2026-03-13T00:36:52.636Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c5/d5/56bc9326bf75f7ea0f4fe1178f5c9f20d1a1e15e288ecf66088513538ccd/mfusepy-3.1.1-py3-none-any.whl", hash = "sha256:69fb70cfc7f7cce595e6ff586f8451d8298f01f18286a157f24f564b24ec5a37", size = 28264, upload-time = "2026-03-13T00:36:51.843Z" }, +] + [[package]] name = "more-itertools" version = "11.1.0" @@ -509,6 +518,9 @@ dev = [ { name = "uvicorn" }, { name = "valkey" }, ] +fuse = [ + { name = "mfusepy" }, +] server = [ { name = "aiohttp" }, { name = "dnspython" }, @@ -530,6 +542,7 @@ requires-dist = [ { name = "httpx", specifier = ">=0.28,<1" }, { name = "jsonschema", specifier = ">=4.25,<5" }, { name = "markdown-it-py", marker = "extra == 'server'", specifier = ">=4,<5" }, + { name = "mfusepy", marker = "extra == 'fuse'", specifier = ">=3.1.1,<4" }, { name = "msgctl", extras = ["server"], marker = "extra == 'dev'" }, { name = "psycopg", extras = ["binary"], marker = "extra == 'server'", specifier = ">=3.3,<4" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.4" }, @@ -543,7 +556,7 @@ requires-dist = [ { name = "valkey", marker = "extra == 'server'", specifier = ">=6.1,<7" }, { name = "yubikey-manager", specifier = ">=5.8,<6" }, ] -provides-extras = ["server", "dev"] +provides-extras = ["fuse", "server", "dev"] [[package]] name = "multidict" From 4233fcef3438e9204083b2548bd841eabe2210a4 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 2 Oct 2026 05:59:43 +0800 Subject: [PATCH 4/7] fix: negotiate FUSE read limits and avoid unmount event-loop deadlocks --- .github/workflows/fuse.yml | 33 +++++++++++++++++-- scripts/check_fuse_mount.py | 8 ++--- src/msg/client_mount.py | 9 ++++-- tests/test_mount_lifecycle.py | 61 +++++++++++++++++++++++++++++++++++ 4 files changed, 101 insertions(+), 10 deletions(-) create mode 100644 tests/test_mount_lifecycle.py diff --git a/.github/workflows/fuse.yml b/.github/workflows/fuse.yml index ba3ee7a5..5eac8819 100644 --- a/.github/workflows/fuse.yml +++ b/.github/workflows/fuse.yml @@ -2,7 +2,7 @@ name: FUSE mount on: pull_request: push: - branches: [main] + branches: [main, work/fuse-mount-20261001] workflow_dispatch: permissions: contents: read @@ -13,6 +13,22 @@ jobs: kernel: runs-on: ubuntu-latest timeout-minutes: 10 + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: msgtest + POSTGRES_PASSWORD: msgtest + POSTGRES_DB: postgres + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U msgtest" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + env: + MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database} steps: - uses: actions/checkout@v7 - uses: actions/setup-python@v7 @@ -22,9 +38,20 @@ jobs: - name: Install native FUSE and optional package run: | sudo apt-get update -qq - sudo env NEEDRESTART_MODE=l apt-get install -y -qq fuse3 libpcsclite-dev swig + sudo env NEEDRESTART_MODE=l apt-get install -y -qq fuse3 age libpcsclite-dev swig python -m pip install uv==0.12.20 uv sync --locked --extra dev --extra fuse --python "$(command -v python)" test -c /dev/fuse + - name: Check FUSE implementation quality + run: | + .venv/bin/ruff check src/msg/client_mount.py tests/test_client_mount.py tests/test_mount_lifecycle.py tests/test_mount_service.py scripts/check_fuse_mount.py + .venv/bin/ruff format --check src/msg/client_mount.py tests/test_client_mount.py tests/test_mount_lifecycle.py tests/test_mount_service.py scripts/check_fuse_mount.py - name: Verify real kernel mount, byte reads and unmount - run: uv run --locked --extra dev --extra fuse python scripts/check_fuse_mount.py + run: .venv/bin/python scripts/check_fuse_mount.py + - name: Verify client contracts and real service authorization + run: .venv/bin/python -m pytest tests/test_client_mount.py tests/test_mount_lifecycle.py tests/test_mount_service.py --junitxml=artifacts/fuse.xml + - uses: actions/upload-artifact@v7.0.1 + if: always() + with: + name: fuse-acceptance + path: artifacts/fuse.xml diff --git a/scripts/check_fuse_mount.py b/scripts/check_fuse_mount.py index 304cb6d5..61b7ba13 100644 --- a/scripts/check_fuse_mount.py +++ b/scripts/check_fuse_mount.py @@ -50,9 +50,7 @@ def inspect_mount(path): stream.seek(12345) assert stream.read(2345) == BODY[12345:14690] for operation in ( - lambda: target.write_bytes(b'no'), - target.unlink, - lambda: (path / 'new').mkdir(), + lambda: target.write_bytes(b'no'), target.unlink, lambda: (path / 'new').mkdir() ): try: operation() @@ -73,7 +71,7 @@ async def exercise(path, *, cancel=False): if task.done(): await task raise AssertionError('FUSE exited before mounting') - if os.path.ismount(path): + if await asyncio.to_thread(os.path.ismount, path): break await asyncio.sleep(0.02) else: @@ -93,7 +91,7 @@ async def exercise(path, *, cancel=False): subprocess.run, [executable, '-u', '--', str(path)], check=True, timeout=10 ) await asyncio.wait_for(task, 10) - assert not os.path.ismount(path) + assert not await asyncio.to_thread(os.path.ismount, path) assert not operations.handles finally: if not task.done(): diff --git a/src/msg/client_mount.py b/src/msg/client_mount.py index e0e1255b..e0cdf09c 100644 --- a/src/msg/client_mount.py +++ b/src/msg/client_mount.py @@ -296,6 +296,11 @@ def __init__(self, backend, loop, *, timeout=45): self.handles = {} self.next_handle = 1 + def init_with_config(self, connection, config): + # libfuse 3 requires the init handshake to agree with -o max_read. + if connection is not None and hasattr(connection, 'max_read'): + connection.max_read = READ_BYTES + def invoke(self, coroutine): async def bounded(): async with asyncio.timeout(self.timeout): @@ -461,10 +466,10 @@ async def serve_mount(binding, operations, mountpoint): # Native libfuse normally handles SIGINT itself. Cancellation from an # embedding application must not leave its worker mounted indefinitely. for _ in range(100): - if task.done() or os.path.ismount(mountpoint): + if task.done() or await asyncio.to_thread(os.path.ismount, mountpoint): break await asyncio.sleep(0.02) - if not task.done() and os.path.ismount(mountpoint): + if not task.done() and await asyncio.to_thread(os.path.ismount, mountpoint): executable = shutil.which('fusermount3') or shutil.which('fusermount') command = ( [executable, '-u', '--', str(mountpoint)] diff --git a/tests/test_mount_lifecycle.py b/tests/test_mount_lifecycle.py new file mode 100644 index 00000000..20105bb1 --- /dev/null +++ b/tests/test_mount_lifecycle.py @@ -0,0 +1,61 @@ +"""Native negotiation and cancellation must not block the SDK event loop.""" + +import asyncio +import os +import threading +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from msg.client_mount import READ_BYTES, ReadOnlyMount, serve_mount + + +def test_fuse3_read_limit_handshake(): + mount = ReadOnlyMount(None, None) + connection = SimpleNamespace(max_read=0) + mount.init_with_config(connection, None) + assert connection.max_read == READ_BYTES + mount.init_with_config(SimpleNamespace(), None) + mount.init_with_config(None, None) + + +@pytest.mark.asyncio +async def test_cancellation_checks_mount_off_the_sdk_loop(monkeypatch, tmp_path): + loop_thread = threading.get_ident() + started, stopped = threading.Event(), threading.Event() + checks = [] + + def native_mount(operations, path, **options): + assert threading.get_ident() != loop_thread + assert options['ro'] and options['direct_io'] + assert 'allow_other' not in options + started.set() + assert stopped.wait(5), 'unmount did not stop the native worker' + + def ismount(path): + checks.append(threading.get_ident()) + assert threading.get_ident() != loop_thread + return started.is_set() and not stopped.is_set() + + async def unmount(*command, **kwargs): + assert command == ('/usr/bin/fusermount3', '-u', '--', str(tmp_path)) + stopped.set() + return SimpleNamespace(wait=AsyncMock(return_value=0)) + + monkeypatch.setattr(os.path, 'ismount', ismount) + monkeypatch.setattr('shutil.which', lambda name: '/usr/bin/fusermount3') + monkeypatch.setattr(asyncio, 'create_subprocess_exec', unmount) + operations = ReadOnlyMount(None, asyncio.get_running_loop()) + operations.handles[1] = object() + task = asyncio.create_task(serve_mount(SimpleNamespace(FUSE=native_mount), operations, tmp_path)) + try: + assert await asyncio.to_thread(started.wait, 2) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 3) + assert checks and not operations.handles + finally: + stopped.set() + if not task.done(): + task.cancel() From 0687d5da21d358ae289f24a9b0151220d32bc7c2 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 2 Oct 2026 06:03:16 +0800 Subject: [PATCH 5/7] style: format FUSE lifecycle regression --- tests/test_mount_lifecycle.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_mount_lifecycle.py b/tests/test_mount_lifecycle.py index 20105bb1..265e128b 100644 --- a/tests/test_mount_lifecycle.py +++ b/tests/test_mount_lifecycle.py @@ -48,7 +48,9 @@ async def unmount(*command, **kwargs): monkeypatch.setattr(asyncio, 'create_subprocess_exec', unmount) operations = ReadOnlyMount(None, asyncio.get_running_loop()) operations.handles[1] = object() - task = asyncio.create_task(serve_mount(SimpleNamespace(FUSE=native_mount), operations, tmp_path)) + task = asyncio.create_task( + serve_mount(SimpleNamespace(FUSE=native_mount), operations, tmp_path) + ) try: assert await asyncio.to_thread(started.wait, 2) task.cancel() From db44c8ea78728358e4950cc733258e40233f0d69 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 2 Oct 2026 06:05:55 +0800 Subject: [PATCH 6/7] style: apply pinned Ruff formatting to native mount smoke --- scripts/check_fuse_mount.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/scripts/check_fuse_mount.py b/scripts/check_fuse_mount.py index 61b7ba13..256a6614 100644 --- a/scripts/check_fuse_mount.py +++ b/scripts/check_fuse_mount.py @@ -50,7 +50,9 @@ def inspect_mount(path): stream.seek(12345) assert stream.read(2345) == BODY[12345:14690] for operation in ( - lambda: target.write_bytes(b'no'), target.unlink, lambda: (path / 'new').mkdir() + lambda: target.write_bytes(b'no'), + target.unlink, + lambda: (path / 'new').mkdir(), ): try: operation() From f3c4927c5a488fb216dce0568505d4bd76b7f3fe Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 2 Oct 2026 06:06:24 +0800 Subject: [PATCH 7/7] ci: collect independent FUSE kernel and service acceptance results --- .github/workflows/fuse.yml | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/.github/workflows/fuse.yml b/.github/workflows/fuse.yml index 5eac8819..018d4e9e 100644 --- a/.github/workflows/fuse.yml +++ b/.github/workflows/fuse.yml @@ -23,7 +23,7 @@ jobs: ports: - 5432:5432 options: >- - --health-cmd "pg_isready -U msgtest" + --health-cmd "pg_isready -U msgtest -d postgres" --health-interval 10s --health-timeout 5s --health-retries 5 @@ -36,6 +36,7 @@ jobs: python-version: '3.15' allow-prereleases: true - name: Install native FUSE and optional package + id: install run: | sudo apt-get update -qq sudo env NEEDRESTART_MODE=l apt-get install -y -qq fuse3 age libpcsclite-dev swig @@ -47,8 +48,10 @@ jobs: .venv/bin/ruff check src/msg/client_mount.py tests/test_client_mount.py tests/test_mount_lifecycle.py tests/test_mount_service.py scripts/check_fuse_mount.py .venv/bin/ruff format --check src/msg/client_mount.py tests/test_client_mount.py tests/test_mount_lifecycle.py tests/test_mount_service.py scripts/check_fuse_mount.py - name: Verify real kernel mount, byte reads and unmount - run: .venv/bin/python scripts/check_fuse_mount.py + if: ${{ !cancelled() && steps.install.outcome == 'success' }} + run: timeout --signal=TERM --kill-after=5s 90s .venv/bin/python scripts/check_fuse_mount.py - name: Verify client contracts and real service authorization + if: ${{ !cancelled() && steps.install.outcome == 'success' }} run: .venv/bin/python -m pytest tests/test_client_mount.py tests/test_mount_lifecycle.py tests/test_mount_service.py --junitxml=artifacts/fuse.xml - uses: actions/upload-artifact@v7.0.1 if: always()