computer: the file API works through the folders' handles (F3) (#104)

ddiff --git a/docs/computer.md b/docs/computer.md
iindex 4f47a95..74930b8 100644
---- a/docs/computer.md
++++ b/docs/computer.md
@@@ -1,8 +1,10 @@
  # The computer plugin: your machine for your agents
  
  The `computer` plugin gives the machine's terminal, files and screen to its owner's agents (OpenVibe.Actor, plan
--T14/T17), through OpenVibe.Bot and the Node's one outbound link. It is **off** unless you turn it on, it never runs
--as root, and **you set what it may do on the machine itself**: nothing in the cloud can raise that level.
++T14/T17), through OpenVibe.Bot and the Node's one outbound link. It is **off** unless you turn it on, it runs as your
++own account (never root: the one exception is a shared desk host for other people, where the Node runs as root and
++every desk sits in its own sandbox, see "Desk hosts for other people"), and **you set what it may do on the machine
++itself**: nothing in the cloud can raise that level.
  
  ## Turn it on
  
@@@ -35,6 +37,17 @@ By hand, or on another system, add it to `plugins` in the Node's `config.json` a
  `observe` is the default. Folders are `roots` (default: the home folder of the account running the Node); a path is
  checked after symlinks are resolved, so a link out of a folder is outside it.
  
++**The folders limit the plugin's own actions**: its reads, its writes, its screen and its browser. They do **not**
++limit a program. At `trusted` a program runs without asking, and at `ask` one runs when you approve it; either way it
++runs as your account, with everything that account can reach, not only the folders. An approved program can read and
++write anywhere your account can.
++
++**How a folder is kept.** On Linux the plugin opens every folder once and resolves each name under that handle
++(`openat2` with `RESOLVE_BENEATH`, or one component at a time where the kernel is older), so a name that is swapped
++for a link after it was checked cannot move the work outside the folder. A link out of the folder is refused, and so is
++a file with more than one name (a hard link to a file outside the folder), below `full`. On macOS and Windows the
++plugin compares path names only: there a name that changes under it is still a name.
++
  **Approvals** (level `ask`): a request waits on the machine until you answer it, for at most `approval_timeout_s`
  (default 120 s) and never longer than the request's own deadline:
  
@@@ -49,6 +62,10 @@ whether stdin is given with its size and digest; a write's path, mode, size and
  or its length and digest. Control, ANSI and bidi characters are escaped. A notification or dialog that cannot show all
  of it offers no **Approve** button: answer that request from `openvibe-node approvals`.
  
++**What an approval does not limit.** Approving a program lets it run as your account, with everything that account can
++reach: the folders are the plugin's own limit, not a sandbox around the program. Read what it would do before you
++approve it.
++
  **The tray** (Linux and Windows): `openvibe-node tray` puts the Node in your notification area. The installer's
  `--computer` starts it when you sign in (`openvibe-node tray --autostart on|off` does it by hand). It runs in your
  desktop session, not in the service, and talks to the Node over the same local socket as the commands above:
ddiff --git a/plugins/computer/openvibe_computer/__init__.py b/plugins/computer/openvibe_computer/__init__.py
iindex 0461ecf..5bbd40c 100644
---- a/plugins/computer/openvibe_computer/__init__.py
++++ b/plugins/computer/openvibe_computer/__init__.py
@@@ -12,7 +12,9 @@ command can never raise it:
  * ``full``: as trusted, anywhere the account running the Node can reach.
  
  The local kill switch (``openvibe-node stop``) and a remote e-stop drop the plugin to observe, kill what is running
--and deny what waits; the core refuses acting commands while either is latched too. The plugin never runs as root.
++and deny what waits; the core refuses acting commands while either is latched too. The plugin runs as the Node's own
++account and never as root, except on a shared desk host for other people, where it runs as root and every desk sits in
++its own sandbox.
  
  Command kinds and their answers (the ack's ``data``):
  
@@@ -108,6 +110,7 @@ from openvibe_plugin import Deferred, Fault, Plugin
  from .browser import KEY_CODES, Browser, BrowserError, find_binary
  from .desk import MAX_DESKS_CAP, BrowserDesk, Desks, PrivateDesk, default_max_desks
  from .egress import EgressProxy, not_public
++from .files import Files, Handles
  
  LEVELS = ("observe", "ask", "trusted", "full")
  RANK = {name: i for i, name in enumerate(LEVELS)}
@@@ -321,6 +324,8 @@ class Computer(Plugin):
          self.platform = platform or sys.platform
          self.level = "observe"
          self.roots = []
++        self.root_forms = {}      # every root, and the names it answers to (the configured one and its own)
++        self._handles = Handles()   # the folders, opened once and kept open
          self.timeout_s = 30
          self.max_output = 64 * 1024
          self.max_read = 256 * 1024
@@@ -398,7 +403,19 @@ class Computer(Plugin):
          if not isinstance(roots, list) or not all(isinstance(r, str) and r.strip() for r in roots):
              raise Fault("bad_value", "roots is a list of folders")
          self.level = level
--        self.roots = [os.path.realpath(os.path.expanduser(r.strip())) for r in roots]
++        # A root is kept under the name it was given and under its own: a folder that is reached through a link
++        # is the same folder, and the plugin must recognize both names for it.
++        forms = {}
++        self.roots = []
++        for r in roots:
++            given = os.path.normpath(os.path.expanduser(r.strip()))
++            real = os.path.realpath(given)
++            self.roots.append(real)
++            names = forms.setdefault(real, [])
++            for name in (given, real):
++                if name not in names:
++                    names.append(name)
++        self.root_forms = forms
          self._control_dirs = self._node_control_dirs()
          self.control_files = self._node_control_files()
          self.timeout_s = _int(config.get("timeout_s"), "timeout_s", 1, 600, 30)
@@@ -852,23 +869,49 @@ class Computer(Plugin):
      def _inside(self, path):
          return any(path == r or path.startswith(r.rstrip(os.sep) + os.sep) for r in self.roots)
  
--    def _path(self, raw, level, create=False):
--        """The real path a command names, inside the roots unless the level is full, never the Node's control files."""
++    def _files(self, level):
++        """The folders this machine allows, and the handles they are reached through."""
++        return Files(self.roots, self.root_forms, level, self._handles)
++
++    def _control_guard(self, st, path):
++        """The object the plugin is about to use is one of the Node's own control files or folders.
++
++        The name may be a link or another name for the same bytes, so the object itself is compared, not the name.
++        """
++        for control in self.control_files:
++            try:
++                other = os.stat(control)
++            except OSError:
++                continue
++            if (st.st_dev, st.st_ino) == (other.st_dev, other.st_ino):
++                raise Fault("not_allowed", "%s is one of the Node's own control files" % path)
++        for control in self._control_dirs:
++            try:
++                other = os.stat(control)
++            except OSError:
++                continue
++            if (st.st_dev, st.st_ino) == (other.st_dev, other.st_ino):
++                raise Fault("not_allowed", "%s is one of the Node's own control folders" % path)
++
++    def _path(self, raw, level, create="no"):
++        """The path a command names: inside the roots unless the level is full, never the Node's control files.
++
++        On Linux the name is resolved through the folders' handles (openat2, or one component at a time where
++        the kernel has no openat2), so what the plugin later uses is the object the handle points at and not
++        the name that was checked. macOS and Windows keep the name checks they always had.
++        """
          if not isinstance(raw, str) or not raw or len(raw) > MAX_ARG_LEN or "\x00" in raw:
              raise Fault("bad_value", "path is a non-empty string")
          path = os.path.expanduser(raw)
          if not os.path.isabs(path):
              raise Fault("bad_value", "path must be absolute")
--        if create and not os.path.lexists(path):
--            parent = os.path.realpath(os.path.dirname(path))
--            real = os.path.join(parent, os.path.basename(path))
--        else:
--            real = os.path.realpath(path)
--        if self._is_control(real):
--            raise Fault("not_allowed", "%s is one of the Node's own control files or folders" % real)
--        if level != "full" and not self._inside(real):
--            raise Fault("not_allowed", "%s is outside this machine's allowed folders" % real)
--        return real
++        target = self._files(level).target(os.path.normpath(path), level, create=create, guard=self._control_guard)
++        # The Node's own files are refused by the name the agent asked for and by the name it resolves to: on macOS
++        # the folder the Node runs in is reached through a link, so only the resolved name says where it is.
++        if self._is_control(target.requested) or self._is_control(target.path):
++            raise Fault("not_allowed", "%s is one of the Node's own control files or folders" % target.path)
++        target.check_now(create)
++        return target
  
      # ---- commands --------------------------------------------------------------------------------------------------
      def handle(self, cmd):
@@@ -1536,18 +1579,34 @@ class Computer(Plugin):
                  % (_argv_text(argv), _plain(resolved), _plain(cwd or "none"), where, cinfo, account))
  
      @staticmethod
--    def _child_preexec():
--        """In the child: its own process group/session, and no privilege gain on Linux."""
--        os.setsid()
--        if sys.platform.startswith("linux"):
--            _set_no_new_privs()
++    def _child_preexec(cwd_fd=None):
++        """In the child: its own process group/session, the folder it runs in, and no privilege gain on Linux."""
++        def preexec():
++            os.setsid()
++            if sys.platform.startswith("linux"):
++                _set_no_new_privs()
++            if cwd_fd is not None:
++                os.fchdir(cwd_fd)   # the folder the parent checked, not a name that may have changed
++        return preexec
  
      @staticmethod
--    def _session_options():
++    def _session_options(cwd_fd=None):
          """A new session/group for an exec'd program (Windows: its own process group, no preexec_fn there)."""
          if os.name == "nt":
              return {"creationflags": getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)}
--        return {"preexec_fn": Computer._child_preexec}
++        return {"preexec_fn": Computer._child_preexec(cwd_fd)}
++
++    def _cwd_options(self, cwd):
++        """How a program is started in a folder: the handle of the folder, checked again at this moment.
++
++        A folder that was swapped since it was checked is refused; the program never runs in a name.
++        """
++        if cwd is None:
++            return {}
++        if cwd.by_name:
++            return {"cwd": cwd.path}
++        fd = cwd.open_dir()
++        return {"cwd": None, "pass_fds": (fd,), "_cwd_fd": fd}
  
      def _exec_op(self, cmd, v, level):
          argv = v.get("argv")
@@@ -1556,9 +1615,9 @@ class Computer(Plugin):
              raise Fault("bad_value", "argv is a list of 1 to %d strings: a program and its arguments" % MAX_ARGS)
          if not _no_new_privs_available():
              raise Fault("unsupported", "no_new_privs is unavailable on this machine: running a program could gain a privilege")
--        cwd = self._path(v["cwd"], level) if v.get("cwd") is not None else (self.roots[0] if self.roots else None)
--        if cwd is not None and not os.path.isdir(cwd):
--            raise Fault("bad_value", "cwd %s is not a folder" % cwd)
++        cwd = self._path(v["cwd"], level) if v.get("cwd") is not None else self._default_cwd(level)
++        if cwd is not None:
++            cwd.remember()   # what it is now: a folder that is swapped later is refused
          stdin = v.get("stdin")
          if stdin is not None and (not isinstance(stdin, str) or len(stdin.encode("utf-8")) > MAX_STDIN):
              raise Fault("bad_value", "stdin is a string of at most 64 KB")
@@@ -1580,7 +1639,7 @@ class Computer(Plugin):
              if stdin is not None:
                  raise Fault("bad_value", "a detached program takes no stdin")
              return (lambda: self._start_on_desk(argv, cwd, env, desk),
--                    self._exec_summary(argv, cwd, None, desk, detached=True))
++                    self._exec_summary(argv, cwd.path if cwd is not None else None, None, desk, detached=True))
  
          def run():
              # The time left on the command's deadline, counted on the clock that deadline is on.
@@@ -1589,7 +1648,13 @@ class Computer(Plugin):
                  raise Fault("expired", "no time was left to run it")
              return self._exec(argv, cwd, stdin, min(timeout, left), env=env)
  
--        return run, self._exec_summary(argv, cwd, stdin, desk)
++        return run, self._exec_summary(argv, cwd.path if cwd is not None else None, stdin, desk)
++
++    def _default_cwd(self, level):
++        """The folder a program runs in when the request names none: the first root, as a handle."""
++        if not self.roots:
++            return None
++        return self._files(level).target(self.roots[0], level)
  
      def _start_on_desk(self, argv, cwd, env, desk):
          """A program that keeps running on a desk (its windows there), ended when the desk stops."""
@@@ -1597,13 +1662,20 @@ class Computer(Plugin):
          running = [p for p in self.desk_programs.get(desk, []) if p.poll() is None]
          if len(running) >= MAX_DESK_PROGRAMS:
              raise Fault("not_ready", "%d programs already run on this desk: stop the desk, or close one there" % len(running))
++        options = self._cwd_options(cwd)
++        fd = options.pop("_cwd_fd", None)
          try:
--            proc = self._spawn(argv, cwd=cwd, env=self._child_env(env), stdin=subprocess.DEVNULL,
--                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, **self._session_options())
--        except FileNotFoundError:
--            raise Fault("bad_value", "no program %s on this machine" % argv[0])
--        except PermissionError:
--            raise Fault("not_allowed", "%s may not be run by the Node's account" % argv[0])
++            try:
++                proc = self._spawn(argv, env=self._child_env(env), stdin=subprocess.DEVNULL,
++                                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
++                                   **options, **self._session_options(fd))
++            except FileNotFoundError:
++                raise Fault("bad_value", "no program %s on this machine" % argv[0])
++            except PermissionError:
++                raise Fault("not_allowed", "%s may not be run by the Node's account" % argv[0])
++        finally:
++            if fd is not None:
++                os.close(fd)
          self.desk_programs[desk] = running + [proc]
          return {"started": True, "pid": proc.pid, "desk": desk}
  
@@@ -1618,14 +1690,20 @@ class Computer(Plugin):
      def _exec(self, argv, cwd, stdin, timeout, env=None):
          self._check_identity()
          started = self.clock()
++        options = self._cwd_options(cwd)
++        fd = options.pop("_cwd_fd", None)
          try:
--            proc = self._spawn(argv, cwd=cwd, stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,
--                               stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=self._child_env(env),
--                               **self._session_options())
--        except FileNotFoundError:
--            raise Fault("bad_value", "no program %s on this machine" % argv[0])
--        except PermissionError:
--            raise Fault("not_allowed", "%s may not be run by the Node's account" % argv[0])
++            try:
++                proc = self._spawn(argv, stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,
++                                   stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=self._child_env(env),
++                                   **options, **self._session_options(fd))
++            except FileNotFoundError:
++                raise Fault("bad_value", "no program %s on this machine" % argv[0])
++            except PermissionError:
++                raise Fault("not_allowed", "%s may not be run by the Node's account" % argv[0])
++        finally:
++            if fd is not None:
++                os.close(fd)
          cap = self.max_output
          got = {"out": [b"", 0], "err": [b"", 0]}
  
@@@ -1759,62 +1837,32 @@ class Computer(Plugin):
          return "other"
  
      def _fs_read_op(self, op, v, level):
--        path = self._path(v.get("path"), level)
--        try:
--            st = os.stat(path)
--        except FileNotFoundError:
--            raise Fault("bad_value", "%s does not exist" % path)
--        except PermissionError:
--            raise Fault("not_allowed", "the Node's account may not read %s" % path)
++        target = self._path(v.get("path"), level)
++        st = target.stat()
          if op == "stat":
--            return {"path": path, "type": self._kind(st.st_mode), "size": st.st_size, "mtime": int(st.st_mtime),
++            return {"path": target.path, "type": self._kind(st.st_mode), "size": st.st_size, "mtime": int(st.st_mtime),
                      "mode": oct(stat.S_IMODE(st.st_mode))}
          if op == "list":
              if not stat.S_ISDIR(st.st_mode):
--                raise Fault("bad_value", "%s is not a folder" % path)
--            entries, total = [], 0
--            try:
--                with os.scandir(path) as it:
--                    for e in sorted(it, key=lambda e: e.name):
--                        total += 1
--                        if len(entries) >= MAX_LIST:
--                            continue
--                        try:
--                            s = e.stat(follow_symlinks=False)
--                            entries.append({"name": e.name, "type": self._kind(s.st_mode), "size": s.st_size,
--                                            "mtime": int(s.st_mtime)})
--                        except OSError:
--                            entries.append({"name": e.name, "type": "other"})
--            except PermissionError:
--                raise Fault("not_allowed", "the Node's account may not list %s" % path)
--            return {"path": path, "entries": entries, "total": total, "truncated": total > len(entries)}
++                raise Fault("bad_value", "%s is not a folder" % target.path)
++            entries, total = target.list(MAX_LIST)
++            return {"path": target.path, "entries": entries, "total": total, "truncated": total > len(entries)}
          if not stat.S_ISREG(st.st_mode):
--            raise Fault("bad_value", "%s is not a file" % path)
++            raise Fault("bad_value", "%s is not a file" % target.path)
          offset = _int(v.get("offset"), "offset", 0, 2 ** 53, 0)
          length = _int(v.get("length"), "length", 0, self.max_read, self.max_read)
--        try:
--            with open(path, "rb") as f:
--                f.seek(offset)
--                data = f.read(length)
--        except PermissionError:
--            raise Fault("not_allowed", "the Node's account may not read %s" % path)
--        return {"path": path, "size": st.st_size, "offset": offset, "length": len(data),
--                "truncated": offset + len(data) < st.st_size, "content_b64": base64.b64encode(data).decode("ascii")}
++        data, size = target.read(offset, length)
++        return {"path": target.path, "size": size, "offset": offset, "length": len(data),
++                "truncated": offset + len(data) < size, "content_b64": base64.b64encode(data).decode("ascii")}
  
      def _fs_write_op(self, op, v, level):
--        path = self._path(v.get("path"), level, create=True)
++        target = self._path(v.get("path"), level, create="any" if op == "mkdir" else "parent")
          if op == "mkdir":
              def run():
                  self._check_identity()
--                existed = os.path.isdir(path)
--                try:
--                    os.makedirs(path, exist_ok=True)
--                except FileExistsError:
--                    raise Fault("bad_value", "%s exists and is not a folder" % path)
--                except PermissionError:
--                    raise Fault("not_allowed", "the Node's account may not create %s" % path)
--                return {"path": path, "created": not existed}
--            return run, "make the folder %s" % _plain(path)
++                created = target.mkdir()
++                return {"path": target.path, "created": created}
++            return run, "make the folder %s" % _plain(target.path)
          mode = v.get("mode") or "create"
          if mode not in ("create", "overwrite", "append"):
              raise Fault("bad_value", "mode is create, overwrite or append")
@@@ -1830,33 +1878,12 @@ class Computer(Plugin):
  
          def run():
              self._check_identity()
--            if os.path.isdir(path):
--                raise Fault("bad_value", "%s is a folder" % path)
--            if mode == "create" and os.path.lexists(path):
--                raise Fault("bad_value", "%s exists (mode overwrite or append replaces or extends it)" % path)
--            try:
--                if mode == "append":
--                    with open(path, "ab") as f:
--                        f.write(data)
--                else:
--                    fd, tmp = tempfile.mkstemp(prefix=".openvibe-", dir=os.path.dirname(path))
--                    try:
--                        with os.fdopen(fd, "wb") as f:
--                            f.write(data)
--                        if os.path.exists(path):
--                            shutil.copymode(path, tmp)
--                        os.replace(tmp, path)
--                    except BaseException:
--                        if os.path.exists(tmp):
--                            os.unlink(tmp)
--                        raise
--            except PermissionError:
--                raise Fault("not_allowed", "the Node's account may not write %s" % path)
--            return {"path": path, "written": len(data), "mode": mode}
++            target.write(data, mode)
++            return {"path": target.path, "written": len(data), "mode": mode}
  
          verb = {"create": "create", "overwrite": "overwrite", "append": "append to"}[mode]
          return run, "%s %s (mode %s, %d bytes, sha256 %s)" % (
--            verb, _plain(path), mode, len(data), _digest(data))
++            verb, _plain(target.path), mode, len(data), _digest(data))
  
      # ---- input -----------------------------------------------------------------------------------------------------
      # ---- browser ---------------------------------------------------------------------------------------------------
ddiff --git a/plugins/computer/openvibe_computer/files.py b/plugins/computer/openvibe_computer/files.py
nnew file mode 100644
iindex 0000000..75fd685
---- /dev/null
++++ b/plugins/computer/openvibe_computer/files.py
@@@ -0,0 +1,600 @@
++"""The computer plugin's files: the folders it may touch, and how it keeps that promise.
++
++On Linux every allowed folder is opened **once**, as a handle, and every request is resolved
++*relative to that handle*: with openat2 (RESOLVE_BENEATH | RESOLVE_NO_MAGICLINKS) where the
++kernel has it (Linux 5.6 and later), and otherwise one component at a time with O_NOFOLLOW.
++The object the plugin then reads, lists, writes or runs a program in is the one the handle points
++at, so a name that is swapped after the check cannot move the work outside the folder.
++
++macOS and Windows keep the name checks they always had: there the plugin compares path names,
++and a name that changes under it is still a name (see docs/computer.md).
++
++A file with more than one name (a hard link) is refused below level full: such a file is
++another name for the same bytes, and the folder limit is about the object, not only about its
++name.
++"""
++
++import ctypes
++import errno
++import os
++import shutil
++import stat
++import sys
++import tempfile
++
++from openvibe_plugin import Fault
++
++MAX_PATH_LEN = 4096
++
++# openat2 (Linux 5.6 and later): the kernel resolves a name relative to a directory handle and
++# refuses to leave it.
++_SYS_OPENAT2 = 437
++_AT_FDCWD = -100
++RESOLVE_NO_MAGICLINKS = 0x02
++RESOLVE_BENEATH = 0x08
++
++_O_PATH = getattr(os, "O_PATH", 0)
++_O_CLOEXEC = getattr(os, "O_CLOEXEC", 0)
++_O_DIRECTORY = getattr(os, "O_DIRECTORY", 0)
++_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
++
++
++class _How(ctypes.Structure):
++    _fields_ = [("flags", ctypes.c_uint64), ("mode", ctypes.c_uint64), ("resolve", ctypes.c_uint64)]
++
++
++def _libc():
++    try:
++        return ctypes.CDLL(None, use_errno=True)
++    except OSError:
++        return None
++
++
++def _openat2(dirfd, path, flags, mode=0, resolve=0):
++    """The kernel's own openat2(2): a name resolved under a handle, never outside it."""
++    libc = _libc()
++    how = _How(flags=flags, mode=mode, resolve=resolve)
++    data = os.fsencode(path)
++    buf = ctypes.create_string_buffer(data, len(data) + 1)
++    ctypes.set_errno(0)
++    rc = libc.syscall(ctypes.c_long(_SYS_OPENAT2), ctypes.c_int(dirfd), buf,
++                      ctypes.byref(how), ctypes.c_size_t(ctypes.sizeof(how)))
++    if rc < 0:
++        err = ctypes.get_errno()
++        raise OSError(err, os.strerror(err), path)
++    return int(rc)
++
++
++def _openat2_usable():
++    """Whether this kernel has openat2."""
++    libc = _libc()
++    if libc is None or not hasattr(libc, "syscall"):
++        return False
++    try:
++        fd = _openat2(_AT_FDCWD, ".", os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC, 0, RESOLVE_NO_MAGICLINKS)
++    except Exception:
++        return False
++    os.close(fd)
++    return True
++
++
++class Handles:
++    """The allowed folders, each opened once and kept open: the plugin never names one again."""
++
++    def __init__(self):
++        # The handles are Linux's; the machine's own platform decides, not what a test says about it.
++        self.linux = sys.platform.startswith("linux")
++        self.openat2 = self.linux and _openat2_usable()
++        self._fds = {}
++
++    def fd(self, path):
++        """The handle of a folder: opened on the first use, then the same one every time."""
++        fd = self._fds.get(path)
++        if fd is None:
++            try:
++                fd = os.open(path, os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC)
++            except FileNotFoundError:
++                raise Fault("not_allowed", "%s does not exist" % path)
++            except PermissionError:
++                raise Fault("not_allowed", "the Node's account may not use %s" % path)
++            except OSError as e:
++                raise Fault("not_allowed", "%s cannot be used: %s" % (path, e.strerror))
++            self._fds[path] = fd
++        return fd
++
++    def close(self):
++        for fd in self._fds.values():
++            try:
++                os.close(fd)
++            except OSError:
++                pass
++        self._fds = {}
++
++    def open_beneath(self, base, rel, flags, mode=0, level="observe"):
++        """Open a name under the handle of a folder. Returns the handle of what it names."""
++        fd = self.fd(base)
++        name = rel or "."
++        if self.openat2:
++            # Below full the name may not leave the folder; at full it may go anywhere the account can.
++            resolve = 0 if level == "full" else (RESOLVE_BENEATH | RESOLVE_NO_MAGICLINKS)
++            try:
++                return _openat2(fd, name, flags, mode, resolve)
++            except OSError as e:
++                raise _fault(e, base, rel)
++        return _walk(fd, name, flags, mode, level == "full")
++
++
++def _fault(err, base, rel):
++    """The fault a failed resolution is: what the kernel said, in the plugin's words."""
++    where = os.path.join(base, rel) if rel else base
++    if err.errno == errno.EXDEV:
++        return Fault("not_allowed", "%s is outside the folder it was checked in" % where)
++    if err.errno == errno.ELOOP:
++        return Fault("not_allowed", "%s is a link this machine does not follow" % where)
++    if err.errno == errno.ENOENT:
++        return Fault("bad_value", "%s does not exist" % where)
++    if err.errno in (errno.EACCES, errno.EPERM):
++        return Fault("not_allowed", "the Node's account may not use %s" % where)
++    if err.errno == errno.ENOTDIR:
++        return Fault("bad_value", "%s is not a folder" % where)
++    if err.errno == errno.EEXIST:
++        return Fault("bad_value", "%s exists" % where)
++    if err.errno == errno.EISDIR:
++        return Fault("bad_value", "%s is a folder" % where)
++    if err.errno == errno.ENAMETOOLONG:
++        return Fault("bad_value", "%s has a name that is too long" % where)
++    return Fault("not_allowed", "%s cannot be used: %s" % (where, err.strerror))
++
++
++def _walk(fd, name, flags, mode=0, follow=False):
++    """Without openat2: the name is walked one component at a time, under the handle of the folder.
++
++    Below full no link is followed at all (a link is the one thing a name check cannot pin down);
++    at full links are followed, as they always were.
++    """
++    parts = [p for p in name.split("/") if p and p != "."]
++    if not parts:
++        return os.dup(fd)
++    # A ".." would climb out of the folder's handle. _path normalizes every name first, so none should reach here;
++    # the walker refuses one anyway, so the folder limit does not rest on its caller (openat2's RESOLVE_BENEATH does
++    # the same on the other path).
++    if not follow and ".." in parts:
++        raise Fault("not_allowed", "%s climbs out of the folder" % name)
++    current = os.dup(fd)
++    try:
++        for i, part in enumerate(parts):
++            last = i == len(parts) - 1
++            if last:
++                oflags = flags
++            else:
++                oflags = os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC
++            if not follow:
++                oflags |= _O_NOFOLLOW
++            try:
++                nxt = os.open(part, oflags, mode, dir_fd=current)
++            except OSError as e:
++                if e.errno in (errno.ELOOP, errno.EMLINK):
++                    raise Fault("not_allowed", "%s is a link: this kernel has no openat2, so no link is followed" % part)
++                raise
++            os.close(current)
++            current = nxt
++    except BaseException:
++        os.close(current)
++        raise
++    return current
++
++
++def _resolved_path(fd):
++    """The path a handle points at now (Linux): what the object is called."""
++    try:
++        return os.readlink("/proc/self/fd/%d" % fd)
++    except OSError:
++        return None
++
++
++class Files:
++    """The folders this machine allows, and the names an agent may ask for inside them."""
++
++    def __init__(self, roots, forms, level, handles):
++        self.roots = list(roots)
++        self.forms = forms or {}
++        self.level = level
++        self.handles = handles
++        # The handle-based operations are Linux's; macOS and Windows keep the name checks they always had.
++        self.native = sys.platform.startswith("linux")
++
++    def _inside(self, path):
++        return any(path == r or path.startswith(r.rstrip(os.sep) + os.sep) for r in self.roots)
++
++    def target(self, path, level, create="no", guard=None):
++        """The target of one request: what it names, and the folder it is checked in.
++
++        The name is resolved once here, so a name that leaves the folder is refused before the owner is asked;
++        every later use resolves it again, under the same handle.
++        """
++        if not self.native:
++            return self._by_name(path, level, create, guard)
++        base, rel = self._split(path, level)
++        return Target(self, base, rel, path, level, guard, create=create)
++
++    def _split(self, path, level):
++        """The folder a path belongs to, and the name under it."""
++        if level == "full":
++            return "/", path.lstrip("/")
++        best = None
++        for root in self.roots:
++            for form in self.forms.get(root, [root]):
++                form = form.rstrip(os.sep) or form
++                if path == form:
++                    rel = ""
++                elif path.startswith(form + os.sep):
++                    rel = path[len(form) + 1:]
++                else:
++                    continue
++                if best is None or len(form) > len(best[0]):
++                    best = (form, rel, root)
++        if best is None:
++            raise Fault("not_allowed", "%s is outside this machine's allowed folders" % path)
++        return best[2], best[1]
++
++    def _by_name(self, path, level, create, guard):
++        """Without handles (macOS, Windows): the name itself is checked, as it always was."""
++        if create != "no" and not os.path.lexists(path):
++            real = os.path.join(os.path.realpath(os.path.dirname(path)), os.path.basename(path))
++        else:
++            real = os.path.realpath(path)
++        if level != "full" and not self._inside(real):
++            raise Fault("not_allowed", "%s is outside this machine's allowed folders" % real)
++        return Target(self, None, None, path, level, guard, by_name=True, create=create, real=real)
++
++
++class Target:
++    """One path an agent asked for, and the handle it is reached through.
++
++    Every operation resolves the name again, under the folder's handle: what the plugin then
++    uses is the object the handle points at, not the name that was checked.
++    """
++
++    def __init__(self, files, base, rel, path, level, guard=None, by_name=False, create="no", real=None):
++        self.files = files
++        self.base = base
++        self.rel = rel
++        self.requested = path   # the name the agent asked for
++        self.level = level
++        self.guard = guard
++        self.by_name = by_name
++        self.create = create
++        self.real = real         # without handles: the name resolved the way it always was
++        self._display = None
++        self.checked = None   # what the folder was when it was checked, by its own numbers
++
++    @property
++    def path(self):
++        """Where the name points now: what the owner is shown, and what the answer names."""
++        if self._display is None:
++            self._display = self._compute_display()
++        return self._display
++
++    def _compute_display(self):
++        """The path the object is at now, as the kernel sees it."""
++        if self.by_name:
++            if self.real is not None:
++                return self.real
++            return self._name_display()
++        try:
++            if self.create == "no":
++                fd = self._open(_O_PATH | _O_CLOEXEC)
++                try:
++                    resolved = _resolved_path(fd)
++                finally:
++                    os.close(fd)
++                if resolved:
++                    return resolved
++            else:
++                parent, name = self._parent()
++                try:
++                    resolved = _resolved_path(parent)
++                finally:
++                    os.close(parent)
++                if resolved:
++                    return os.path.join(resolved, name)
++        except Fault:
++            pass
++        return self._name_display()
++
++    def _name_display(self):
++        """The name resolved the way it was before the handles: what the path names."""
++        if self.create == "no" or os.path.lexists(self.requested):
++            return os.path.realpath(self.requested)
++        return os.path.join(os.path.realpath(os.path.dirname(self.requested)), os.path.basename(self.requested))
++
++    # ---- resolution ---------------------------------------------------------------------------------
++    def _open(self, flags, mode=0):
++        """The handle of the object this path names."""
++        if self.by_name:
++            return os.open(self.path, flags, mode)
++        return self.files.handles.open_beneath(self.base, self.rel, flags, mode, self.level)
++
++    def _parent(self):
++        """The handle of the folder the name is in, and the name itself."""
++        if self.by_name:
++            parent = os.path.dirname(self.path) or os.sep
++            return os.open(parent, os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC), os.path.basename(self.path)
++        rel = self.rel
++        cut = rel.rfind("/")
++        if cut < 0:
++            return os.dup(self.files.handles.fd(self.base)), rel
++        return self.files.handles.open_beneath(self.base, rel[:cut], os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC,
++                                               0, self.level), rel[cut + 1:]
++
++    def _check(self, st, links=False):
++        """The object is a regular file with more than one name: refused below full."""
++        if links and self.level != "full" and stat.S_ISREG(st.st_mode) and st.st_nlink > 1:
++            raise Fault("not_allowed",
++                        "%s has %d names (a hard link): a file linked from outside the folder is refused"
++                        % (self.path, st.st_nlink))
++        if self.guard is not None:
++            self.guard(st, self.path)
++
++    def check_now(self, create):
++        """Resolve the name once, so what cannot be used is refused before anything waits for an answer."""
++        if self.by_name:
++            return   # without handles the name is checked where it is used, as it always was
++        if create == "no":
++            self.stat()
++        elif create == "parent":
++            parent, name = self._parent()
++            os.close(parent)
++
++    def identity(self):
++        """What the path names now, by its own numbers: a folder that was swapped is a different one.
++
++        Without handles (macOS, Windows) the name is stat'ed: a folder cannot be opened as a file there, and the
++        name is all there is to compare.
++        """
++        if self.by_name:
++            try:
++                st = os.stat(self.path)
++            except OSError:
++                return None
++            return (st.st_dev, st.st_ino)
++        fd = self.open_dir()
++        try:
++            st = os.fstat(fd)
++        finally:
++            os.close(fd)
++        return (st.st_dev, st.st_ino)
++
++    def remember(self):
++        """Remember what this folder was when it was checked: a swap is seen later."""
++        self.checked = self.identity()
++        return self.checked
++
++    # ---- operations --------------------------------------------------------------------------------
++    def stat(self):
++        """What the path names, as the kernel sees it now."""
++        if self.by_name:
++            try:
++                return os.stat(self.path)
++            except FileNotFoundError:
++                raise Fault("bad_value", "%s does not exist" % self.path)
++            except PermissionError:
++                raise Fault("not_allowed", "the Node's account may not read %s" % self.path)
++        fd = self._open(_O_PATH | _O_CLOEXEC)
++        try:
++            st = os.fstat(fd)
++        finally:
++            os.close(fd)
++        self._check(st)
++        return st
++
++    def list(self, limit):
++        """The names in a folder, as the kernel sees them now."""
++        if self.by_name:
++            try:
++                with os.scandir(self.path) as it:
++                    return self._entries(it, limit)
++            except FileNotFoundError:
++                raise Fault("bad_value", "%s does not exist" % self.path)
++            except PermissionError:
++                raise Fault("not_allowed", "the Node's account may not list %s" % self.path)
++        fd = self._open(os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC)
++        try:
++            with os.scandir(fd) as it:
++                return self._entries(it, limit)
++        finally:
++            os.close(fd)
++
++    @staticmethod
++    def _entries(it, limit):
++        entries, total = [], 0
++        for e in sorted(it, key=lambda e: e.name):
++            total += 1
++            if len(entries) >= limit:
++                continue
++            try:
++                s = e.stat(follow_symlinks=False)
++                entries.append({"name": e.name, "type": _kind(s.st_mode), "size": s.st_size, "mtime": int(s.st_mtime)})
++            except OSError:
++                entries.append({"name": e.name, "type": "other"})
++        return entries, total
++
++    def read(self, offset, length):
++        """The bytes of the file, as the kernel sees them now."""
++        if self.by_name:
++            try:
++                with open(self.path, "rb") as f:
++                    st = os.fstat(f.fileno())
++                    self._check(st, links=True)
++                    f.seek(offset)
++                    return f.read(length), st.st_size
++            except FileNotFoundError:
++                raise Fault("bad_value", "%s does not exist" % self.path)
++            except PermissionError:
++                raise Fault("not_allowed", "the Node's account may not read %s" % self.path)
++        fd = self._open(os.O_RDONLY | _O_CLOEXEC)
++        try:
++            st = os.fstat(fd)
++            self._check(st, links=True)
++            data = os.pread(fd, length, offset)
++            return data, st.st_size
++        finally:
++            os.close(fd)
++
++    def write(self, data, mode):
++        """Write into the file: a temporary file in its own folder, then the replace."""
++        if self.by_name:
++            self._write_by_name(data, mode)
++            return
++        if mode == "append":
++            fd = self._open(os.O_WRONLY | os.O_APPEND | _O_CLOEXEC)
++            try:
++                self._check(os.fstat(fd), links=True)
++                _write_all(fd, data)
++            finally:
++                os.close(fd)
++            return
++        parent, name = self._parent()
++        try:
++            if mode == "create":
++                try:
++                    os.stat(name, dir_fd=parent, follow_symlinks=False)
++                except FileNotFoundError:
++                    pass
++                else:
++                    raise Fault("bad_value", "%s exists (mode overwrite or append replaces or extends it)" % self.path)
++            else:
++                # What the name points at now: a file with two names is refused, and so is one of the Node's own.
++                try:
++                    st = os.stat(name, dir_fd=parent, follow_symlinks=True)
++                except FileNotFoundError:
++                    st = None
++                if st is not None:
++                    self._check(st, links=True)
++            tmp = ".openvibe-%s" % os.urandom(8).hex()
++            fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_CLOEXEC, 0o600, dir_fd=parent)
++            try:
++                _write_all(fd, data)
++            finally:
++                os.close(fd)
++            try:
++                st = os.stat(name, dir_fd=parent, follow_symlinks=False)
++                os.chmod(tmp, stat.S_IMODE(st.st_mode), dir_fd=parent)
++            except FileNotFoundError:
++                pass
++            os.replace(tmp, name, src_dir_fd=parent, dst_dir_fd=parent)
++        except PermissionError:
++            raise Fault("not_allowed", "the Node's account may not write %s" % self.path)
++        except IsADirectoryError:
++            raise Fault("bad_value", "%s is a folder" % self.path)
++        except FileExistsError:
++            raise Fault("bad_value", "%s exists (mode overwrite or append replaces or extends it)" % self.path)
++        finally:
++            os.close(parent)
++
++    def _write_by_name(self, data, mode):
++        """Without handles (macOS, Windows): the name is written as it always was."""
++        path = self.path
++        if os.path.isdir(path):
++            raise Fault("bad_value", "%s is a folder" % path)
++        if mode == "create" and os.path.lexists(path):
++            raise Fault("bad_value", "%s exists (mode overwrite or append replaces or extends it)" % path)
++        try:
++            if mode == "append":
++                with open(path, "ab") as f:
++                    self._check(os.fstat(f.fileno()), links=True)
++                    f.write(data)
++            else:
++                fd, tmp = tempfile.mkstemp(prefix=".openvibe-", dir=os.path.dirname(path))
++                try:
++                    with os.fdopen(fd, "wb") as f:
++                        f.write(data)
++                    if os.path.exists(path):
++                        shutil.copymode(path, tmp)
++                    os.replace(tmp, path)
++                except BaseException:
++                    if os.path.exists(tmp):
++                        os.unlink(tmp)
++                    raise
++        except FileNotFoundError:
++            raise Fault("bad_value", "%s does not exist" % path)
++        except PermissionError:
++            raise Fault("not_allowed", "the Node's account may not write %s" % path)
++
++    def mkdir(self):
++        """Make the folder, and the folders above it that are missing."""
++        if self.by_name:
++            existed = os.path.isdir(self.path)
++            try:
++                os.makedirs(self.path, exist_ok=True)
++            except FileExistsError:
++                raise Fault("bad_value", "%s exists and is not a folder" % self.path)
++            except PermissionError:
++                raise Fault("not_allowed", "the Node's account may not create %s" % self.path)
++            return not existed
++        parts = [p for p in (self.rel or "").split("/") if p]
++        made = False
++        for i, part in enumerate(parts):
++            if i == 0:
++                parent = os.dup(self.files.handles.fd(self.base))
++            else:
++                parent = self.files.handles.open_beneath(self.base, "/".join(parts[:i]),
++                                                          os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC,
++                                                          0, self.level)
++            try:
++                try:
++                    os.mkdir(part, 0o777, dir_fd=parent)
++                    if i == len(parts) - 1:
++                        made = True
++                except FileExistsError:
++                    pass
++                except PermissionError:
++                    raise Fault("not_allowed", "the Node's account may not create %s" % self.path)
++            finally:
++                os.close(parent)
++        return made
++
++    def open_dir(self):
++        """The handle of a folder, for a program to run in.
++
++        A folder that was swapped since it was checked is refused: the program does not run in a name.
++        """
++        if self.by_name:
++            try:
++                return os.open(self.path, os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC)
++            except FileNotFoundError:
++                raise Fault("bad_value", "%s does not exist" % self.path)
++            except NotADirectoryError:
++                raise Fault("bad_value", "%s is not a folder" % self.path)
++            except PermissionError:
++                raise Fault("not_allowed", "the Node's account may not use %s" % self.path)
++        fd = self._open(os.O_RDONLY | _O_DIRECTORY | _O_CLOEXEC)
++        if self.checked is not None:
++            st = os.fstat(fd)
++            if (st.st_dev, st.st_ino) != self.checked:
++                os.close(fd)
++                raise Fault("not_allowed", "%s is not the folder that was checked" % self.path)
++        return fd
++
++
++def _write_all(fd, data):
++    """Write every byte, or say so."""
++    view = memoryview(data)
++    while view:
++        written = os.write(fd, view)
++        view = view[written:]
++
++
++def _kind(mode):
++    """The kind of a file, as the plugin answers it."""
++    if stat.S_ISDIR(mode):
++        return "dir"
++    if stat.S_ISLNK(mode):
++        return "link"
++    if stat.S_ISREG(mode):
++        return "file"
++    if stat.S_ISFIFO(mode):
++        return "fifo"
++    if stat.S_ISSOCK(mode):
++        return "socket"
++    return "other"
ddiff --git a/plugins/computer/tests/test_files.py b/plugins/computer/tests/test_files.py
nnew file mode 100644
iindex 0000000..b4bfb60
---- /dev/null
++++ b/plugins/computer/tests/test_files.py
@@@ -0,0 +1,331 @@
++"""The computer plugin's files: the folder is opened once, and the name is checked where it is used (F3).
++
++The review of 2026-11-05 found that the plugin checked a path and then used the name again: a folder or a file
++swapped for a link after the check moved the work outside the folders the owner allowed. These tests swap the
++name (the file itself, the folder above it, the root) between the check and the use, and a hard-linked file, and
++ask for the outside file to stay untouched.
++"""
++import base64
++import io
++import json
++import os
++import shutil
++import sys
++import tempfile
++import unittest
++
++HERE = os.path.dirname(os.path.abspath(__file__))
++sys.path.insert(0, HERE)
++sys.path.insert(0, os.path.dirname(HERE))
++sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(HERE)), "sdk"))
++
++from openvibe_computer import Computer  # noqa: E402
++from openvibe_plugin import Runtime  # noqa: E402
++from test_computer import Out, reply  # noqa: E402
++
++LINUX = sys.platform.startswith("linux")
++
++
+[email protected](LINUX, "the handle-based file operations are Linux's")
++class FileHandleTest(unittest.TestCase):
++    """The files are reached through the folders' handles: a name that is swapped cannot move the work."""
++
++    def boot(self, level="observe", **config):
++        self.dir = tempfile.mkdtemp(prefix="ov-files-")
++        self.outside = tempfile.mkdtemp(prefix="ov-outside-")
++        self.addCleanup(shutil.rmtree, self.dir, ignore_errors=True)
++        self.addCleanup(shutil.rmtree, self.outside, ignore_errors=True)
++        os.mkdir(os.path.join(self.dir, "sub"))
++        with open(os.path.join(self.dir, "notes.txt"), "w") as f:
++            f.write("hello\n")
++        with open(os.path.join(self.dir, "sub", "inner.txt"), "w") as f:
++            f.write("inner\n")
++        with open(os.path.join(self.outside, "secret.txt"), "w") as f:
++            f.write("SECRET\n")
++        with open(os.path.join(self.outside, "inner.txt"), "w") as f:
++            f.write("SECRET\n")
++        self.out = Out()
++        self.plugin = Computer(env={}, platform="linux")
++        self.rt = Runtime(self.plugin, infile=io.StringIO(""), outfile=self.out)
++        self.rt.dispatch({"op": "hello", "config": {"level": level, "roots": [self.dir], **config}})
++        self.n = 0
++        return self.out.msgs()
++
++    def cmd(self, kind, value, deadline_ms=30000):
++        self.n += 1
++        cid = "f%d" % self.n
++        self.rt.dispatch({"op": "command", "id": cid, "kind": kind, "value": value, "deadline_ms": deadline_ms})
++        return cid
++
++    def swap_after_the_check(self, swap):
++        """The swap happens right after the plugin checked the name, and before it uses it."""
++        original = self.plugin._path
++
++        def patched(raw, level, create=False):
++            target = original(raw, level, create)
++            swap()
++            return target
++        self.plugin._path = patched
++
++    def wait_for_the_approval(self, cid):
++        self.out.wait_for(lambda m: m.get("name") == "approval" and m.get("id") == cid)
++
++    def link_out(self, path, target):
++        """A link from a name in the folder to a file outside it."""
++        os.symlink(target, path)
++
++    # ---- the name is swapped after it was checked -----------------------------------------------
++    def test_a_file_swapped_after_the_check_is_not_read(self):
++        # F3: the file was there when the name was checked, and a link out of the folder replaced it.
++        self.boot("observe")
++        target = os.path.join(self.dir, "notes.txt")
++        outside = os.path.join(self.outside, "secret.txt")
++
++        def swap():
++            os.remove(target)
++            self.link_out(target, outside)
++        self.swap_after_the_check(swap)
++        r = reply(self.out, self.cmd("fs", {"op": "read", "path": target}))
++        self.assertEqual(r["op"], "nack", "the file outside the folder was read: %r" % r)
++        self.assertNotIn("SECRET", json.dumps(r))
++
++    def test_a_folder_swapped_after_the_check_is_not_listed(self):
++        # F3: the folder was there when the name was checked, and a link out of the root replaced it.
++        self.boot("observe")
++        target = os.path.join(self.dir, "sub")
++
++        def swap():
++            os.rename(target, target + "-old")
++            self.link_out(target, self.outside)
++        self.swap_after_the_check(swap)
++        r = reply(self.out, self.cmd("fs", {"op": "list", "path": target}))
++        self.assertEqual(r["op"], "nack", "the folder outside the roots was listed: %r" % r)
++        self.assertNotIn("SECRET", json.dumps(r))
++
++    def test_a_file_swapped_after_the_check_is_not_stated(self):
++        # F3: the same for a stat: what is answered is the file outside the folder.
++        self.boot("observe")
++        target = os.path.join(self.dir, "notes.txt")
++        outside = os.path.join(self.outside, "secret.txt")
++
++        def swap():
++            os.remove(target)
++            self.link_out(target, outside)
++        self.swap_after_the_check(swap)
++        r = reply(self.out, self.cmd("fs", {"op": "stat", "path": target}))
++        self.assertEqual(r["op"], "nack", "the file outside the folder was stated: %r" % r)
++
++    def test_the_folder_above_the_name_swapped_after_the_check_is_not_read(self):
++        # F3: every ancestor of the name, not only the name itself.
++        self.boot("observe")
++        folder = os.path.join(self.dir, "sub")
++        target = os.path.join(folder, "inner.txt")
++
++        def swap():
++            os.rename(folder, folder + "-old")
++            self.link_out(folder, self.outside)
++        self.swap_after_the_check(swap)
++        r = reply(self.out, self.cmd("fs", {"op": "read", "path": target}))
++        self.assertEqual(r["op"], "nack", "the file outside the folder was read: %r" % r)
++        self.assertNotIn("SECRET", json.dumps(r))
++
++    # ---- the name is swapped while the owner is asked ------------------------------------------
++    def test_a_write_whose_folder_is_swapped_while_it_waits_is_refused(self):
++        # F3: the request waits for the owner; the folder above the file is replaced by a link out of the root.
++        self.boot("ask")
++        folder = os.path.join(self.dir, "sub")
++        target = os.path.join(folder, "out.txt")
++        cid = self.cmd("fs", {"op": "write", "path": target, "content_b64": "eA=="})
++        self.wait_for_the_approval(cid)
++        os.rename(folder, folder + "-old")
++        self.link_out(folder, self.outside)
++        self.plugin.on_approval(cid, True)
++        r = reply(self.out, cid)
++        self.assertEqual(r["op"], "nack", "the file outside the folder was written: %r" % r)
++        self.assertFalse(os.path.exists(os.path.join(self.outside, "out.txt")), "a file was written outside")
++
++    def test_an_overwrite_whose_folder_is_swapped_while_it_waits_is_refused(self):
++        # F3: the same for an overwrite.
++        self.boot("ask")
++        folder = os.path.join(self.dir, "sub")
++        target = os.path.join(folder, "inner.txt")
++        cid = self.cmd("fs", {"op": "write", "path": target, "content_b64": "eA==", "mode": "overwrite"})
++        self.wait_for_the_approval(cid)
++        os.rename(folder, folder + "-old")
++        self.link_out(folder, self.outside)
++        self.plugin.on_approval(cid, True)
++        r = reply(self.out, cid)
++        self.assertEqual(r["op"], "nack", "the file outside the folder was written: %r" % r)
++        with open(os.path.join(self.outside, "secret.txt")) as f:
++            self.assertEqual(f.read(), "SECRET\n", "a file outside the folder was changed")
++
++    def test_an_append_through_a_link_written_while_it_waits_is_refused(self):
++        # F3: the file itself is replaced by a link while the owner is asked: the append must not follow it.
++        self.boot("ask")
++        target = os.path.join(self.dir, "notes.txt")
++        outside = os.path.join(self.outside, "appended.txt")
++        cid = self.cmd("fs", {"op": "write", "path": target, "content_b64": "eA==", "mode": "append"})
++        self.wait_for_the_approval(cid)
++        os.rename(target, target + "-old")
++        self.link_out(target, outside)
++        self.plugin.on_approval(cid, True)
++        r = reply(self.out, cid)
++        self.assertEqual(r["op"], "nack", "the file outside the folder was written: %r" % r)
++        self.assertFalse(os.path.exists(outside), "the target of the link was created")
++
++    def test_a_mkdir_whose_folder_is_swapped_while_it_waits_is_refused(self):
++        # F3: the folder above the new one is replaced by a link while the owner is asked.
++        self.boot("ask")
++        folder = os.path.join(self.dir, "sub")
++        target = os.path.join(folder, "made")
++        cid = self.cmd("fs", {"op": "mkdir", "path": target})
++        self.wait_for_the_approval(cid)
++        os.rename(folder, folder + "-old")
++        self.link_out(folder, self.outside)
++        self.plugin.on_approval(cid, True)
++        r = reply(self.out, cid)
++        self.assertEqual(r["op"], "nack", "the folder outside the root was made: %r" % r)
++        self.assertFalse(os.path.exists(os.path.join(self.outside, "made")), "a folder was made outside")
++
++    def test_an_exec_cwd_swapped_while_it_waits_does_not_run_outside(self):
++        # F3: the folder a program runs in is replaced by a link while the owner is asked.
++        self.boot("ask")
++        folder = os.path.join(self.dir, "sub")
++        cid = self.cmd("exec", {"argv": [sys.executable, "-c", "open('marker.txt', 'w').write('x')"], "cwd": folder})
++        self.wait_for_the_approval(cid)
++        os.rename(folder, folder + "-old")
++        self.link_out(folder, self.outside)
++        self.plugin.on_approval(cid, True)
++        r = reply(self.out, cid)
++        self.assertEqual(r["op"], "nack", "the program ran in the folder outside the roots: %r" % r)
++        self.assertFalse(os.path.exists(os.path.join(self.outside, "marker.txt")), "the program wrote outside")
++
++    def test_an_exec_cwd_swapped_before_it_is_used_is_refused(self):
++        # F3: the same, with the swap between the check and the start (no approval in the way).
++        self.boot("trusted")
++        folder = os.path.join(self.dir, "sub")
++
++        def swap():
++            os.rename(folder, folder + "-old")
++            self.link_out(folder, self.outside)
++        self.swap_after_the_check(swap)
++        cid = self.cmd("exec", {"argv": [sys.executable, "-c", "open('marker.txt', 'w').write('x')"], "cwd": folder})
++        r = reply(self.out, cid)
++        self.assertEqual(r["op"], "nack", "the program ran in the folder outside the roots: %r" % r)
++        self.assertFalse(os.path.exists(os.path.join(self.outside, "marker.txt")), "the program wrote outside")
++
++    # ---- links -----------------------------------------------------------------------------------
++    def test_a_dangling_link_does_not_create_its_target(self):
++        # F3: a link to a file that is not there yet: writing through it must not create the file outside.
++        self.boot("trusted")
++        target = os.path.join(self.dir, "dangling")
++        outside = os.path.join(self.outside, "made-by-the-link.txt")
++        self.link_out(target, outside)
++        r = reply(self.out, self.cmd("fs", {"op": "write", "path": target, "content_b64": "eA==", "mode": "append"}))
++        self.assertEqual(r["op"], "nack", "the write through the link was accepted: %r" % r)
++        self.assertFalse(os.path.exists(outside), "the target of the link was created")
++
++    def test_a_link_out_of_the_root_is_refused(self):
++        # F3: a link that was already there is refused as well.
++        self.boot("observe")
++        target = os.path.join(self.dir, "escape")
++        self.link_out(target, self.outside)
++        r = reply(self.out, self.cmd("fs", {"op": "list", "path": target}))
++        self.assertEqual((r["op"], r["fault_code"]), ("nack", "not_allowed"), r)
++
++    def test_the_root_swapped_keeps_the_work_in_the_folder_that_was_opened(self):
++        # F3: the root itself is replaced by a link out of it: the plugin keeps the folder it opened.
++        self.boot("ask")
++        target = os.path.join(self.dir, "out.txt")
++        cid = self.cmd("fs", {"op": "write", "path": target, "content_b64": "eA=="})
++        self.wait_for_the_approval(cid)
++        os.rename(self.dir, self.dir + "-moved")
++        self.link_out(self.dir, self.outside)
++        self.plugin.on_approval(cid, True)
++        reply(self.out, cid)
++        self.assertFalse(os.path.exists(os.path.join(self.outside, "out.txt")), "a file was written outside")
++
++    def test_without_openat2_a_link_is_refused(self):
++        # A kernel without openat2: the plugin walks the name itself and follows no link at all.
++        self.boot("observe")
++        self.plugin._handles.openat2 = False
++        target = os.path.join(self.dir, "link")
++        self.link_out(target, os.path.join(self.dir, "sub"))
++        r = reply(self.out, self.cmd("fs", {"op": "list", "path": target}))
++        self.assertEqual(r["op"], "nack", "the link was followed: %r" % r)
++
++
++class HardLinkTest(unittest.TestCase):
++    """A file with more than one name is another name for the same bytes: refused below full."""
++
++    def boot(self, level="observe"):
++        self.dir = tempfile.mkdtemp(prefix="ov-hard-")
++        self.outside = tempfile.mkdtemp(prefix="ov-hard-out-")
++        self.addCleanup(shutil.rmtree, self.dir, ignore_errors=True)
++        self.addCleanup(shutil.rmtree, self.outside, ignore_errors=True)
++        with open(os.path.join(self.dir, "notes.txt"), "w", newline="\n") as f:   # the same bytes on Windows
++            f.write("hello\n")
++        self.out = Out()
++        self.plugin = Computer(env={}, platform=sys.platform)
++        self.rt = Runtime(self.plugin, infile=io.StringIO(""), outfile=self.out)
++        self.rt.dispatch({"op": "hello", "config": {"level": level, "roots": [self.dir]}})
++        self.n = 0
++
++    def cmd(self, kind, value, deadline_ms=30000):
++        self.n += 1
++        cid = "h%d" % self.n
++        self.rt.dispatch({"op": "command", "id": cid, "kind": kind, "value": value, "deadline_ms": deadline_ms})
++        return cid
++
++    def link(self):
++        """The file in the folder, and a second name for it outside."""
++        inside = os.path.join(self.dir, "notes.txt")
++        outside = os.path.join(self.outside, "alias.txt")
++        try:
++            os.link(inside, outside)
++        except (OSError, NotImplementedError, AttributeError) as e:
++            self.skipTest("this machine cannot make a hard link: %s" % e)
++        return inside, outside
++
++    def test_a_hard_link_is_not_read(self):
++        self.boot("observe")
++        inside, outside = self.link()
++        r = reply(self.out, self.cmd("fs", {"op": "read", "path": inside}))
++        self.assertEqual((r["op"], r["fault_code"]), ("nack", "not_allowed"), r)
++
++    def test_a_hard_link_is_not_appended_to(self):
++        self.boot("trusted")
++        inside, outside = self.link()
++        r = reply(self.out, self.cmd("fs", {"op": "write", "path": inside, "content_b64": "eA==", "mode": "append"}))
++        self.assertEqual((r["op"], r["fault_code"]), ("nack", "not_allowed"), r)
++        with open(outside) as f:
++            self.assertEqual(f.read(), "hello\n", "the file outside the folder was changed")
++
++    def test_at_full_a_hard_link_is_read(self):
++        # At full the folders do not limit anything, so the file is read as any other.
++        self.boot("full")
++        inside, outside = self.link()
++        r = reply(self.out, self.cmd("fs", {"op": "read", "path": inside}))
++        self.assertEqual(r["op"], "ack", r)
++        self.assertEqual(base64.b64decode(r["data"]["content_b64"]), b"hello\n")
++
++
++class WalkTest(unittest.TestCase):
++    @unittest.skipUnless(LINUX, "the handle-based file operations are Linux's")
++    def test_the_walker_refuses_a_name_that_climbs_out(self):
++        # Without openat2 the walker resolves under the folder's handle; a ".." must not leave it, whoever calls it.
++        from openvibe_computer import files
++        from openvibe_plugin import Fault
++        root = tempfile.mkdtemp()
++        try:
++            os.makedirs(os.path.join(root, "proj"))
++            fd = os.open(os.path.join(root, "proj"), os.O_RDONLY | os.O_DIRECTORY)
++            try:
++                with self.assertRaises(Fault):
++                    files._walk(fd, "../outside", os.O_RDONLY)
++                with self.assertRaises(Fault):
++                    files._walk(fd, "a/../../outside", os.O_RDONLY)
++            finally:
++                os.close(fd)
++        finally:
++            shutil.rmtree(root)