lib.ec.secret_store.Vault: Add the store class

This commit adds the Vault store and the vault:// dispatch in FileContext.

A Vault KV v2 server maps nicely into jw-pkg's secret handling abstraction
as just another secret store backend. Environment variables identify the
storage details, analogous to the way it's handled in the SOPS backend.

- JW_SECRETS_VAULT_ADDR names the server, a unix socket address or an http
  URL; the standard VAULT_ADDR is never read; JW_SECRETS_VAULT_MOUNT the
  mount path, default secret; the token is the standard VAULT_TOKEN
- The transport is one curl per API call, through the Local context: a
  unix:// server is reached through its socket, an http:// server through
  its port
- A node's type is derived at stat time: an entry, a blob, a key below an
  entry, a branch if the path has children, otherwise a miss
- A blob's content is base64 under a single key of its KV secret, the
  marker being a path component, so a KV secret that is not a blob is
  always an entry

The test drives a dev server fixture from the workspace: stat/readdir/get
on the fixture, the default fallback, the storage marker, and the copy
forms - single key, whole entry, recursive entry, deploy to the on-host
layout, and a host subtree.

Signed-off-by: Jan Lindemann <jan@janware.com>
Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.2.0

src/python/jw/pkg/lib/FileContext.py

@@ -376,6 +376,10 @@ class FileContext(abc.ABC):
                     f'{scheme}://{uri.authority}', chroot = False
                 )
                 return KvFile(inner, uri = uri)
+            case 'vault':
+                from .ec.secret_store.Vault import Vault
+
+                return Vault(uri, *args, **kwargs)
             case _:
                 pass
         raise Exception(

src/python/jw/pkg/lib/ec/secret_store/Vault.py

@@ -0,0 +1,350 @@
+from __future__ import annotations
+
+import base64
+import json
+import os
+
+from datetime import datetime
+from typing import TYPE_CHECKING, Any, Iterable, override
+
+if TYPE_CHECKING:
+    from ...base import StatResult
+    from ...file_types import FileType
+    from ...Uri import Uri
+
+from ...file_types import FT_DIR
+from ...kv import parse_kv, serialize_kv
+from ...Result import Result
+from .Base import Base
+from .layout import FT_SECRET, FT_SECRET_FILE, FT_SECRET_KEY
+
+# -- The single key of a blob's KV secret: the base64 bytes
+BLOB_KEY = 'content'
+
+class Vault(Base):
+    """A secret store in a Vault KV v2 backend.
+
+    The server and the mount are deployment information, not part of
+    the address: JW_SECRETS_VAULT_ADDR names the server, a unix://
+    socket address or an http:// URL, and JW_SECRETS_VAULT_MOUNT the
+    KV v2 mount path, defaulting to secret. The standard VAULT_ADDR
+    is never read. The store URI addresses a subtree of the
+    namespace, as in the base class; the token comes from the
+    standard VAULT_TOKEN environment variable.
+
+    A store path names a KV secret, and optionally a key below an
+    entry. The KV secret of an entry is its address, the map being
+    the entry, and its keys are addressable below it; a blob is a KV
+    secret at its address, holding the base64 bytes under a single
+    key. The store does not identify content by path: a KV map whose
+    only key is the blob's content key is a blob, anything else an
+    entry.
+
+    put() writes the secret at its address: a key merged into its
+    entry, an entry wholesale, or a blob as base64 under the single
+    content key. The store accepts entry, key, and blob content
+    only; a KV v2 write is a full replacement, so a key not in the
+    content is gone.
+    """
+
+    def __init__(self, uri: str | Uri, *args: Any, **kwargs: Any) -> None:
+        super().__init__(uri, *args, **kwargs)
+        self.__addr = os.environ.get('JW_SECRETS_VAULT_ADDR')
+        self.__mount = os.environ.get('JW_SECRETS_VAULT_MOUNT', 'secret')
+        self.__secret_cache: dict[str, tuple[dict[str, str], float]] = {}
+
+    @staticmethod
+    def __mtime(created_time: str) -> float:
+        if not created_time:
+            return 0
+        return datetime.fromisoformat(created_time.replace('Z', '+00:00')).timestamp()
+
+    @staticmethod
+    def __is_blob_map(kv: dict[str, str]) -> bool:
+        # -- The map is a blob: the base64 bytes under the single
+        #    content key
+        if len(kv) != 1 or BLOB_KEY not in kv:
+            return False
+        try:
+            base64.b64decode(kv[BLOB_KEY], validate = True)
+        except (ValueError, TypeError):
+            return False
+        return True
+
+    async def __api(
+        self,
+        method: str,
+        storage_path: str,
+        data: dict[str, Any] | None = None,
+        list_path: bool = False,
+        metadata: bool = False,
+    ) -> dict[str, Any]:
+        # -- One Vault API call, <method> /v1/<mount>/<data|metadata>/<path>;
+        #    a 404 is a miss, any other failure an error
+        from ...base import InputMode
+        from ..Local import Local
+
+        if not self.__addr:
+            raise Exception('JW_SECRETS_VAULT_ADDR is not set')
+        token = os.environ.get('VAULT_TOKEN')
+        if not token:
+            raise Exception('VAULT_TOKEN is not set')
+        kind = 'metadata' if metadata or list_path else 'data'
+        api_path = f'{self.__mount}/{kind}/{storage_path}'
+        if list_path:
+            api_path += '?list=true'
+        if self.__addr.startswith('unix://'):
+            url = f'http://localhost/v1/{api_path}'
+            socket_args = ['--unix-socket', self.__addr[len('unix://'):]]
+        else:
+            base = self.__addr.rstrip('/')
+            if not base.startswith('http'):
+                base = f'http://{base}'
+            url = f'{base}/v1/{api_path}'
+            socket_args = []
+        cmd = [
+            'curl',
+            '--silent',
+            '--show-error',
+            '--fail-with-body',
+            '--request',
+            method,
+            *socket_args,
+            '--header',
+            f'X-Vault-Token: {token}',
+            # -- Required by Vault 3 to route the request to the API
+            #    rather than the web UI
+            '--header',
+            'X-Vault-Request: true',
+            url,
+        ]
+        if data is not None:
+            cmd += ['--data', json.dumps(data)]
+        ret = await Local().run(
+            cmd, cmd_input = InputMode.NonInteractive, throw = False
+        )
+        body: dict[str, Any] = {}
+        if ret.stdout:
+            body = json.loads(ret.stdout)
+        if ret.status != 0:
+            if ret.status == 22:
+                raise FileNotFoundError(storage_path)
+            raise Exception(
+                f'vault api: {method} {api_path} failed: {ret.stderr.decode()}'
+            )
+        return body
+
+    async def __secret(self, storage_path: str) -> tuple[dict[str, str], float]:
+        # -- The KV secret at the storage path: its data map and the
+        #    time derived from the created_time metadata
+        ret = self.__secret_cache.get(storage_path)
+        if ret is None:
+            body = await self.__api('GET', storage_path)
+            payload = body.get('data') or {}
+            raw = payload.get('data') or {}
+            kv = {str(k): str(v) for k, v in raw.items()}
+            meta = payload.get('metadata') or {}
+            ret = (kv, self.__mtime(str(meta.get('created_time', ''))))
+            self.__secret_cache[storage_path] = ret
+        return ret
+
+    async def __list(self, storage_path: str) -> list[str]:
+        # -- The children at the storage path, in the address space;
+        #    a subdirectory is marked with a trailing slash
+        body = await self.__api('GET', storage_path, list_path = True)
+        keys = (body.get('data') or {}).get('keys') or []
+        ret = []
+        for k in keys:
+            if k.endswith('/'):
+                ret.append(k[:-1])
+            else:
+                ret.append(k)
+        return ret
+
+    async def __vault_stat(
+        self,
+        storage_path: str,
+        follow_symlinks: bool,
+    ) -> StatResult:
+        try:
+            kv, mtime = await self.__secret(storage_path)
+        except FileNotFoundError:
+            pass
+        else:
+            if self.__is_blob_map(kv):
+                return self._stat_result('0', '0', mtime, FT_SECRET_FILE)
+            return self._stat_result('0', '0', mtime, FT_SECRET)
+        entry, sep, key = storage_path.rpartition('/')
+        if sep and key:
+            try:
+                kv, mtime = await self.__secret(entry)
+            except FileNotFoundError:
+                pass
+            else:
+                if key in kv:
+                    return self._stat_result('0', '0', mtime, FT_SECRET_KEY)
+        if await self.__list(storage_path):
+            return self._stat_result('0', '0', 0, FT_DIR)
+        raise FileNotFoundError(storage_path)
+
+    async def __vault_readdir(self, storage_path: str) -> Iterable[str]:
+        if storage_path:
+            try:
+                kv, _ = await self.__secret(storage_path)
+            except FileNotFoundError:
+                pass
+            else:
+                if self.__is_blob_map(kv):
+                    raise NotADirectoryError(storage_path)
+                return kv.keys()
+        names: set[str] = set()
+        try:
+            names.update(await self.__list(storage_path))
+        except FileNotFoundError:
+            pass
+        default_path = self._default_path_in_storage(storage_path)
+        if default_path is not None:
+            try:
+                names.update(await self.__list(default_path))
+            except FileNotFoundError:
+                pass
+        if not names:
+            raise FileNotFoundError(storage_path)
+        return sorted(names)
+
+    async def __vault_get(self, storage_path: str) -> Result:
+        # -- The entry is the KV map, serialized; the blob is the
+        #    base64 bytes under BLOB_KEY of its KV secret
+        try:
+            kv, _ = await self.__secret(storage_path)
+        except FileNotFoundError:
+            kv = None
+        if kv is not None:
+            if self.__is_blob_map(kv):
+                return Result(stdout = base64.b64decode(kv[BLOB_KEY]))
+            return Result(stdout = serialize_kv(kv))
+        if '/' in storage_path:
+            entry, key = storage_path.rsplit('/', 1)
+            try:
+                kv, _ = await self.__secret(entry)
+            except FileNotFoundError:
+                raise FileNotFoundError(storage_path) from None
+            try:
+                return Result(stdout = kv[key].encode())
+            except KeyError:
+                raise FileNotFoundError(storage_path) from None
+        raise FileNotFoundError(storage_path)
+
+    # -- Override base class methods
+
+    @override
+    async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
+        path = self._unchroot(path)
+        return await self._call_with_fallback(
+            path,
+            lambda p: self.__vault_stat(p, follow_symlinks),
+        )
+
+    @override
+    async def _get(
+        self,
+        path: str,
+        wd: str | None,
+        throw: bool,
+        verbose: bool | None,
+        title: str,
+    ) -> Result:
+        path = self._unchroot(path)
+        return await self._call_with_fallback(path, self.__vault_get)
+
+    @override
+    async def _readdir(self, path: str) -> Iterable[str]:
+        path = self._unchroot(path)
+        return await self._call_with_fallback(path, self.__vault_readdir)
+
+    async def __put_entry(self, storage_path: str, content: bytes) -> Result:
+        # -- Replace the entry: the content is the serialized entry,
+        #    and a KV v2 write is a full replacement, so a key not in
+        #    it is gone
+        kv = parse_kv(content)
+        self.__secret_cache.pop(storage_path, None)
+        await self.__api('POST', storage_path, data = {'data': kv})
+        return Result()
+
+    async def __put_key(self, storage_path: str, key: str, content: bytes) -> Result:
+        # -- Merge the key into the entry, replacing its existing
+        #    value for the key
+        try:
+            kv, _ = await self.__secret(storage_path)
+        except FileNotFoundError:
+            kv = {}
+        kv[key] = content.decode()
+        self.__secret_cache.pop(storage_path, None)
+        await self.__api('POST', storage_path, data = {'data': kv})
+        return Result()
+
+    async def __put_blob(self, storage_path: str, content: bytes) -> Result:
+        self.__secret_cache.pop(storage_path, None)
+        await self.__api(
+            'POST',
+            storage_path,
+            data = {'data': {
+                BLOB_KEY: base64.b64encode(content).decode()
+            }},
+        )
+        return Result()
+
+    async def __vault_unlink(self, storage_path: str) -> None:
+        # -- Drop the secret at the storage path: the data and its
+        #    metadata, so the name leaves the listing. A DELETE is a
+        #    blind success on KV v2 - it 204s even for a path that
+        #    never existed - so probe with a GET first
+        try:
+            await self.__api('GET', storage_path)
+        except FileNotFoundError:
+            raise
+        await self.__api('DELETE', storage_path)
+        await self.__api('DELETE', storage_path, metadata = True)
+        self.__secret_cache.pop(storage_path, None)
+
+    @override
+    async def _put(
+        self,
+        path: str,
+        content: bytes,
+        wd: str | None,
+        throw: bool,
+        verbose: bool | None,
+        title: str,
+        owner: str | None,
+        group: str | None,
+        mode: str | None,
+        atomic: bool,
+        ftype: FileType,
+    ) -> Result:
+        # -- Resolve the address to the absolute namespace path in
+        #    the host tree; a write does not take the default
+        #    fallback.
+        path = self._unchroot(path)
+        if not path:
+            raise ValueError(f'put: "{path}" is not an entry/key or blob path')
+        storage_path = self._path_in_storage(path)
+        if ftype.is_subtype(FT_SECRET):
+            return await self.__put_entry(storage_path, content)
+        if ftype.is_subtype(FT_SECRET_KEY):
+            entry, sep, key = storage_path.rpartition('/')
+            if not sep or not key:
+                raise ValueError(f'put: "{path}" is not of the form path/key')
+            return await self.__put_key(entry, key, content)
+        if ftype.is_subtype(FT_SECRET_FILE):
+            return await self.__put_blob(storage_path, content)
+        raise ValueError(f'put: ftype {ftype} is not a store content type')
+
+    @override
+    async def _unlink(self, path: str) -> None:
+        # -- Remove the node; a write does not take the default
+        #    fallback
+        path = self._unchroot(path)
+        if not path:
+            raise ValueError(f'unlink: "{path}" is not a node path')
+        await self.__vault_unlink(self._path_in_storage(path))

test/unit/python/jw/pkg/lib/ec/secret_store/Vault/Makefile

@@ -0,0 +1,7 @@
+TOPDIR = ../../../../../../../../../
+
+include $(TOPDIR)/make/proj.mk
+include $(JWBDIR)/make/py-run.mk
+
+all:
+test: run

test/unit/python/jw/pkg/lib/ec/secret_store/Vault/test.py

@@ -0,0 +1,383 @@
+import asyncio
+import base64
+import json
+import os
+import stat
+import subprocess
+import tempfile
+import uuid
+
+from pathlib import Path
+from typing import Any, override
+
+from jw.pkg.lib.distros.suse.Distro import Distro as SuseDistro
+from jw.pkg.lib.ec.Local import Local
+from jw.pkg.lib.ec.secret_store.KvFile import KvFile
+from jw.pkg.lib.ec.secret_store.layout import FT_SECRET, FT_SECRET_FILE, FT_SECRET_KEY
+from jw.pkg.lib.ec.secret_store.Vault import BLOB_KEY
+from jw.pkg.lib.file_types import FT_DIR
+from jw.pkg.lib.FileContext import FileContext
+from jw.pkg.lib.kv import parse_kv
+from jw.pkg.lib.util import copy
+
+# -- The dev server is a fixture of the workspace, not a dependency
+#    of the tree: test/vault, started by
+#    test/scripts/setup-vault-test-server.sh
+def vault_base() -> Path:
+    return Path(__file__).resolve().parents[10] / 'test' / 'vault'
+
+def vault_sock() -> str:
+    return str(vault_base() / 'vault.sock')
+
+def vault_api(method: str,
+              api_path: str,
+              data: dict[str, Any] | None = None) -> subprocess.CompletedProcess[str]:
+    cmd = [
+        'curl',
+        '--silent',
+        '--show-error',
+        '--fail-with-body',
+        '--request',
+        method,
+        '--unix-socket',
+        vault_sock(),
+        '--header',
+        f'X-Vault-Token: {os.environ["VAULT_TOKEN"]}',
+        # -- Required by Vault 3 to route the request to the API
+        '--header',
+        'X-Vault-Request: true',
+        f'http://localhost/v1/{api_path}',
+    ]
+    if data is not None:
+        cmd += ['--data', json.dumps(data)]
+    return subprocess.run(cmd, capture_output = True, text = True)
+
+def vault_put_kv(rel: str, kv: dict[str, str]) -> None:
+    vault_api('POST', f'secret/data/{rel}', {'data': kv}).check_returncode()
+
+def vault_get_meta_version(rel: str) -> int:
+    ret = vault_api('GET', f'secret/data/{rel}')
+    ret.check_returncode()
+    return int(json.loads(ret.stdout)['data']['metadata']['version'])
+
+def have_vault() -> bool:
+    # -- Skip the test if the dev server is not up
+    if not os.path.exists(vault_sock()) or not (vault_base() / 'token').exists():
+        return False
+    os.environ['JW_SECRETS_VAULT_ADDR'] = f'unix://{vault_sock()}'
+    os.environ['VAULT_TOKEN'] = (vault_base() / 'token').read_text().strip()
+    ret = vault_api('GET', 'secret/metadata?list=true')
+    return ret.returncode == 0
+
+async def run_tests() -> None:
+    if not have_vault():
+        return
+
+    # -- A unique host per run, so the test does not collide with
+    #    whatever else the dev server holds
+    host = f'vault-{uuid.uuid4().hex[:8]}.example.com'
+    rev = '/'.join(reversed(host.split('.')))
+
+    # -- The fixture, seeded through the API: an entry with two
+    #    keys, a key-only default counterpart, a default-only entry,
+    #    a blob, and a branch holding a blob. Entries and blobs
+    #    alike are KV secrets at their store paths
+    vault_put_kv(f'{rev}/etc/a.jw-secret', {'a': '1', 'b': '2'})
+    vault_put_kv('default/etc/a.jw-secret', {'a': 'default-a'})
+    vault_put_kv('default/etc/d.jw-secret', {'d': 'default-d'})
+    vault_put_kv(
+        f'{rev}/etc/c.pem.jw-secret-file',
+        {BLOB_KEY: base64.b64encode(b'-----BEGIN CERT-----\n').decode()},
+    )
+    vault_put_kv(
+        f'{rev}/etc/new/x.pem.jw-secret-file',
+        {BLOB_KEY: base64.b64encode(b'-----BEGIN NEW BRANCH CERT-----\n').decode()},
+    )
+
+    with tempfile.TemporaryDirectory() as tmp:
+        store = FileContext.create(f'vault://{host}')
+        assert store is not None
+        await store.open()
+
+        # -- stat: the node's type is derived; the size is 0, as in
+        #    /proc
+        st = await store.stat('etc/a.jw-secret')
+        assert st.ftype is FT_SECRET and stat.S_ISREG(st.mode)
+        assert st.size == 0
+        st = await store.stat('etc/a.jw-secret/b')
+        assert st.ftype is FT_SECRET_KEY and stat.S_ISREG(st.mode)
+        st = await store.stat('etc/c.pem.jw-secret-file')
+        assert st.ftype is FT_SECRET_FILE and stat.S_ISREG(st.mode)
+        st = await store.stat('etc/new')
+        assert st.ftype is FT_DIR and stat.S_ISDIR(st.mode)
+        try:
+            await store.stat('etc/missing')
+            raise AssertionError('expected FileNotFoundError')
+        except FileNotFoundError:
+            pass
+        # -- file_exists() works on the store, being built on stat()
+        assert await store.file_exists('etc/a.jw-secret')
+        assert not await store.file_exists('etc/missing')
+
+        # -- The address is the KV secret's path: the blob at
+        #    etc/c.pem.jw-secret-file is the KV secret at that path
+        body = vault_api('GET', f'secret/data/{rev}/etc/c.pem.jw-secret-file')
+        body.check_returncode()
+        assert json.loads(body.stdout)['data']['data'] == {
+            BLOB_KEY: base64.b64encode(b'-----BEGIN CERT-----\n').decode(),
+        }
+
+        # -- readdir: the entry lists its keys
+        assert set(await store.readdir('etc/a.jw-secret')) == {'a', 'b'}
+        # -- the branch lists the names below it as stored, the
+        #    default tree unioned in
+        assert set(await store.readdir('etc')) == \
+            {'a.jw-secret', 'c.pem.jw-secret-file', 'd.jw-secret', 'new'}
+        # -- the root lists the host tree
+        assert set(await store.readdir('')) == {'etc'}
+        # -- a blob is not a directory
+        try:
+            await store.readdir('etc/c.pem.jw-secret-file')
+            raise AssertionError('expected NotADirectoryError')
+        except NotADirectoryError:
+            pass
+        # -- a miss is an error
+        try:
+            await store.readdir('etc/missing')
+            raise AssertionError('expected FileNotFoundError')
+        except FileNotFoundError:
+            pass
+
+        # -- get: the key, the whole entry, the blob
+        assert (await store.get('etc/a.jw-secret/b')).stdout == b'2'
+        assert (await store.get('etc/a.jw-secret')).stdout == b'a=1\nb=2\n'
+        assert (await store.get('etc/c.pem.jw-secret-file')).stdout == \
+            b'-----BEGIN CERT-----\n'
+        # -- a miss is an error
+        try:
+            await store.get('etc/a.jw-secret/missing')
+            raise AssertionError('expected FileNotFoundError')
+        except FileNotFoundError:
+            pass
+
+        # -- The default fallback: a miss below the host is retried
+        #    below the default tree
+        assert (await store.get('etc/d.jw-secret')).stdout == b'd=default-d\n'
+        await store.close()
+
+        # -- A store address as a copy source: FileContext.create()
+        #    resolves the scheme
+        dst = os.path.join(tmp, 'copied', 'a')
+        os.makedirs(os.path.dirname(dst))
+        ret = await copy(f'vault://{host}/etc/a.jw-secret/a', dst)
+        assert ret == dst
+        with open(dst) as f:
+            assert f.read() == '1'
+
+        # A single object that is an entry: the serialized whole
+        # lands at the destination
+        dst = os.path.join(tmp, 'copied', 'entry')
+        os.makedirs(os.path.dirname(dst), exist_ok = True)
+        ret = await copy(f'vault://{host}/etc/a.jw-secret', dst)
+        assert ret == dst
+        with open(dst) as f:
+            assert f.read() == 'a=1\nb=2\n'
+
+        # -- A store entry as a recursive copy source: the entry is
+        #    gettable as a whole, and the walk terminates at it
+        tree = os.path.join(tmp, 'tree')
+        ret = await copy(f'vault://{host}/etc/a.jw-secret', tree, recursive = True)
+        assert ret == tree
+        with open(tree) as f:
+            assert f.read() == 'a=1\nb=2\n'
+
+        # -- Deploy: a recursive copy from the store to the on-host
+        #    layout; the kvfile scheme wraps the target context,
+        #    chrooted at the root of the on-host tree. The host tree
+        #    is a precondition: etc/ exists
+        host_tree = os.path.join(tmp, 'host-tree')
+        os.makedirs(os.path.join(host_tree, 'etc'))
+        ret = await copy(f'vault://{host}', f'kvfile://{host_tree}', recursive = True)
+        assert isinstance(ret, list)
+        assert set(ret) == {
+            'etc/a.jw-secret',
+            'etc/c.pem.jw-secret-file',
+            'etc/d.jw-secret',
+            'etc/new/x.pem.jw-secret-file',
+        }
+        # -- Entries land as entry files, mode 0400; the host
+        #    value wins over default, and the default-only entry d
+        #    lands from the union listing
+        for entry, kv in (
+            ('etc/a.jw-secret', {'a': '1', 'b': '2'}),
+            ('etc/d.jw-secret', {'d': 'default-d'}),
+        ):
+            path = os.path.join(host_tree, *entry.split('/'))
+            assert os.path.isfile(path)
+            assert stat.S_IMODE(os.stat(path).st_mode) == 0o400
+            with open(path, 'rb') as fb:
+                assert parse_kv(fb.read()) == kv
+        # -- The blob lands as a .jw-secret-file at its config path,
+        #    and no .jw-secret garbage is created for the branch
+        pem = os.path.join(host_tree, 'etc', 'c.pem.jw-secret-file')
+        with open(pem) as f:
+            assert f.read() == '-----BEGIN CERT-----\n'
+        assert not os.path.exists(os.path.join(host_tree, 'etc.jw-secret'))
+        # -- A branch is a real directory: its blob lands suffixed
+        #    below it
+        assert os.path.isdir(os.path.join(host_tree, 'etc', 'new'))
+        new_pem = os.path.join(host_tree, 'etc', 'new', 'x.pem.jw-secret-file')
+        with open(new_pem) as f:
+            assert f.read() == '-----BEGIN NEW BRANCH CERT-----\n'
+        assert not os.path.exists(os.path.join(host_tree, 'etc', 'new.jw-secret'))
+
+        # -- A recursive copy onto a plain host tree: an entry
+        #    lands as a single file with its serialized content, a
+        #    branch as a directory, and a blob as a file
+        plain_host = os.path.join(tmp, 'plain-host')
+        os.makedirs(plain_host)
+        ret = await copy(f'vault://{host}', plain_host, recursive = True)
+        assert isinstance(ret, list)
+        assert set(ret) == {
+            f'{plain_host}/etc/a.jw-secret',
+            f'{plain_host}/etc/c.pem.jw-secret-file',
+            f'{plain_host}/etc/d.jw-secret',
+            f'{plain_host}/etc/new/x.pem.jw-secret-file',
+        }
+        with open(os.path.join(plain_host, 'etc', 'a.jw-secret')) as f:
+            assert f.read() == 'a=1\nb=2\n'
+        assert not os.path.exists(os.path.join(plain_host, 'etc', 'a', 'a'))
+        with open(os.path.join(plain_host, 'etc', 'c.pem.jw-secret-file')) as f:
+            assert f.read() == '-----BEGIN CERT-----\n'
+        assert os.path.isdir(os.path.join(plain_host, 'etc', 'new'))
+        with open(os.path.join(plain_host, 'etc', 'new', 'x.pem.jw-secret-file')) as f:
+            assert f.read() == '-----BEGIN NEW BRANCH CERT-----\n'
+
+        # -- A host subtree: the fallback mirrors below the subtree.
+        #    root stays the URI path; the namespace root is separate
+        store = FileContext.create(f'vault://{host}/etc')
+        assert store is not None
+        assert store.root == '/etc'
+        await store.open()
+        assert (await store.get('a.jw-secret/a')).stdout == b'1'
+        assert (await store.get('d.jw-secret/d')).stdout == b'default-d'
+        await store.close()
+
+        # -- put(): a key put merges into its entry, a leaf put
+        #    writes a blob, a leaf below an existing entry lands as
+        #    a key there, and a top-level object is a blob at a
+        #    top-level config path
+        store = FileContext.create(f'vault://{host}')
+        await store.open()
+        await store.put('etc/w1.jw-secret/k1', b'v1', ftype = FT_SECRET_KEY)
+        assert (await store.get('etc/w1.jw-secret/k1')).stdout == b'v1'
+        await store.put(
+            'etc/w2.pem.jw-secret-file',
+            b'-----BEGIN KEY-----\n',
+            ftype = FT_SECRET_FILE
+        )
+        assert (await store.get('etc/w2.pem.jw-secret-file')).stdout == \
+            b'-----BEGIN KEY-----\n'
+        await store.put('etc/a.jw-secret/extra', b'x', ftype = FT_SECRET_KEY)
+        assert set(await store.readdir('etc/a.jw-secret')) == {'a', 'b', 'extra'}
+        await store.put('etc/a.jw-secret/a', b'9', ftype = FT_SECRET_KEY)
+        assert (await store.get('etc/a.jw-secret/a')).stdout == b'9'
+        # -- a typed put replaces the entry: the serialized content
+        #    is written wholesale, and a KV v2 write is a full
+        #    replacement, so a key not in it is gone
+        await store.put('etc/a.jw-secret', b'a=9\nextra=e\n', ftype = FT_SECRET)
+        assert set(await store.readdir('etc/a.jw-secret')) == {'a', 'extra'}
+        assert (await store.get('etc/a.jw-secret')).stdout == b'a=9\nextra=e\n'
+        try:
+            await store.get('etc/a.jw-secret/b')
+            raise AssertionError('expected FileNotFoundError')
+        except FileNotFoundError:
+            pass
+        await store.put('top', b'1', ftype = FT_SECRET_FILE)
+        assert (await store.get('top')).stdout == b'1'
+        # -- a put that changes nothing leaves the secret version
+        #    untouched
+        version = vault_get_meta_version(f'{rev}/etc/w1.jw-secret')
+        await store.put('etc/w1.jw-secret/k1', b'v1', ftype = FT_SECRET_KEY)
+        assert vault_get_meta_version(f'{rev}/etc/w1.jw-secret') == version
+        # -- The namespace root itself has no store home
+        try:
+            await store.put('', b'1')
+            raise AssertionError('expected ValueError')
+        except ValueError:
+            pass
+        await store.close()
+
+        # -- unlink(): drops the entry, or the blob, and the name
+        #    leaves the listing
+        store = FileContext.create(f'vault://{host}')
+        await store.open()
+        await store.unlink('etc/w1.jw-secret')
+        assert not await store.file_exists('etc/w1.jw-secret')
+        await store.unlink('etc/w2.pem.jw-secret-file')
+        assert not await store.file_exists('etc/w2.pem.jw-secret-file')
+        try:
+            await store.unlink('etc/missing')
+            raise AssertionError('expected FileNotFoundError')
+        except FileNotFoundError:
+            pass
+        await store.close()
+
+        # -- Fetch: a recursive copy from the on-host layout into
+        #    the store. The kvfile source is handed the host's
+        #    package set, and the entries and the blob land below
+        #    the destination host's prefix
+        fetch_host = f'vault-{uuid.uuid4().hex[:8]}.fetch.example.com'
+        host_tree = os.path.join(tmp, 'fetch-host')
+        os.makedirs(os.path.join(host_tree, 'etc'))
+        with open(os.path.join(host_tree, 'etc', 'foo.conf.jw-secret'), 'w') as fh:
+            fh.write('f1=1\nf2=2\n')
+        with open(os.path.join(host_tree, 'etc', 'cert.pem.jw-secret-file'), 'w') as fh:
+            fh.write('-----BEGIN FETCHED CERT-----\n')
+        # -- As on a real host, the deployed config and cert are
+        #    present too: the key paths below the entry have no
+        #    host counterpart
+        with open(os.path.join(host_tree, 'etc', 'foo.conf'), 'w') as fh:
+            fh.write('deployed config\n')
+        with open(os.path.join(host_tree, 'etc', 'cert.pem'), 'w') as fh:
+            fh.write('deployed cert\n')
+
+        class FakeDistro(SuseDistro):
+
+            def __init__(self, files: list[str]) -> None:
+                super().__init__(Local(), id = 'suse')
+                self.files = files
+
+            @override
+            async def _pkg_files(self, names: list[str] | None) -> list[str]:
+                return self.files
+
+        fake = FakeDistro(
+            [
+                f'{host_tree}/etc/foo.conf.jw-tmpl',
+                f'{host_tree}/etc/cert.pem.jw-tmpl',
+            ]
+        )
+        src = KvFile(Local(), distro = fake)
+        base = host_tree.lstrip('/')
+        ret = await copy(src, f'vault://{fetch_host}', recursive = True)
+        assert isinstance(ret, list)
+        # -- The copy transports the on-host file names unchanged
+        assert set(ret) == {
+            f'{base}/etc/foo.conf.jw-secret',
+            f'{base}/etc/cert.pem.jw-secret-file',
+        }
+        fstore = FileContext.create(f'vault://{fetch_host}')
+        await fstore.open()
+        assert set(await fstore.readdir(f'{base}/etc/foo.conf.jw-secret')) == \
+            {'f1', 'f2'}
+        assert (await fstore.get(f'{base}/etc/foo.conf.jw-secret/f1')).stdout == \
+            b'1'
+        assert (await fstore.get(f'{base}/etc/cert.pem.jw-secret-file')).stdout == \
+            b'-----BEGIN FETCHED CERT-----\n'
+        await fstore.close()
+
+def main() -> None:
+    asyncio.run(run_tests())
+
+main()