Skip to content

_yaml

_yaml

Round-trip YAML write-back for the few sandbox sites that mutate config.yml.

Mirrors the round-trip facade in terok.lib.util.yaml — sandbox can't import from terok, so the two stay duplicated; keep in sync if the parser race-fix comment ever changes.

A fresh YAML() per call: the class carries parser state, so a module-level singleton races under concurrent use from TUI worker threads.

__all__ = ['update_section', 'write_secret_text'] module-attribute

update_section(path, section, updates)

Merge updates into data[section] at path, preserving comments.

Source code in src/terok_sandbox/_yaml.py
def update_section(path: Path, section: str, updates: dict[str, Any]) -> None:
    """Merge *updates* into ``data[section]`` at *path*, preserving comments."""
    yaml = YAML(typ="rt")
    yaml.preserve_quotes = True
    if path.exists():
        existing = yaml.load(path.read_text(encoding="utf-8")) or {}
    else:
        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        existing = {}
    if not isinstance(existing, dict):
        raise ValueError(
            f"{path} top-level is {type(existing).__name__}, expected a mapping;"
            " refusing to silently overwrite — fix or move aside the file by hand"
        )
    # ``setdefault`` returns whatever sits at the key — a stale scalar
    # written by a previous schema version would explode on ``.update``.
    if not isinstance(existing.get(section), dict):
        existing[section] = {}
    existing[section].update(updates)
    buf = StringIO()
    yaml.dump(existing, buf)
    write_secret_text(path, buf.getvalue())

write_secret_text(path, text)

Atomically create-or-replace path with mode 0o600.

Writes a temp file in the same directory, fsyncs, then os.replaces it onto path. A crash mid-write leaves either the old file intact or no file at all — never a truncated one. mkstemp opens with mode 0o600 from the kernel, so a passphrase never has a window of world-readable visibility.

Source code in src/terok_sandbox/_yaml.py
def write_secret_text(path: Path, text: str) -> None:
    """Atomically create-or-replace *path* with mode 0o600.

    Writes a temp file in the same directory, fsyncs, then
    ``os.replace``s it onto *path*.  A crash mid-write leaves either
    the old file intact or no file at all — never a truncated one.
    ``mkstemp`` opens with mode 0o600 from the kernel, so a passphrase
    never has a window of world-readable visibility.
    """
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    fd, tmp_name = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    tmp_path = Path(tmp_name)
    try:
        # POSIX ``write`` is allowed to return a short count; loop until
        # the full payload is committed.  A truncated config.yml or
        # vault.passphrase would lock the operator out of the vault.
        data = memoryview(text.encode("utf-8"))
        while data:
            data = data[os.write(fd, data) :]
        os.fsync(fd)
        os.close(fd)
        fd = -1
        os.replace(tmp_path, path)
    except BaseException:
        if fd >= 0:
            os.close(fd)
        tmp_path.unlink(missing_ok=True)
        raise