Voice / vlib /registry.py

Update voice.py, vlib, tests, bundles

fcff745 about 2 months ago

11.7 kB

"""Voice registry, ops log, and write guards."""

from pathlib import Path
import json
import os
import sys
import time
from vlib import ctx
from vlib.ui import _confirm, _fail, _noninteractive, _note, _say, _step, _warn
from vlib.tensors import _is_gguf_file, _is_quant_fetched, _write_safetensors_streaming, _write_single_tensor_gguf, write_safetensors
from vlib.sources import _NAME_RULES, _sanitize_voice_name, _slugify_voice_name


def log_operation(record):
    record = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), **record}
    try:
        ctx.VOICES_DIR.mkdir(parents=True, exist_ok=True)
        try:
            os.chmod(ctx.VOICES_DIR, 0o700)
        except Exception:
            pass
        _assert_not_symlink(ctx.OPERATIONS_LOG)
        with open(ctx.OPERATIONS_LOG, "a") as f:
            f.write(json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n")
            try:
                f.flush()
                os.fsync(f.fileno())
            except Exception:
                pass
        try:
            os.chmod(ctx.OPERATIONS_LOG, 0o600)
        except Exception:
            pass
        try:
            dirfd = os.open(str(ctx.VOICES_DIR), os.O_DIRECTORY)
            try:
                os.fsync(dirfd)
            finally:
                os.close(dirfd)
        except Exception:
            pass
    except Exception:
        if ctx.VERBOSE:
            _warn("could not write operation log")


def _atomic_jsonl_append(path, record):
    """Append one JSON line atomically with flock + fsync. Optimised: single chmod, cached JSON."""
    path = Path(path)
    # fast path: parent exists, avoid mkdir/chmod churn
    if not path.parent.exists():
        path.parent.mkdir(parents=True, exist_ok=True)
        try:
            os.chmod(path.parent, 0o700)
        except Exception:
            pass
    if not path.exists():
        try:
            fd = os.open(str(path), os.O_WRONLY | os.O_CREAT, 0o600)
            os.close(fd)
        except Exception:
            pass
    else:
        # occasional chmod drift check, not every append
        try:
            if oct(path.stat().st_mode & 0o777) != "0o600":
                os.chmod(path, 0o600)
        except Exception:
            pass
    _assert_not_symlink(path)
    line = json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n"
    with open(path, "a") as f:
        try:
            import fcntl
            fcntl.flock(f, fcntl.LOCK_EX)
            locked = True
        except Exception:
            locked = False
        try:
            f.write(line)
            f.flush()
            try:
                os.fsync(f.fileno())
            except Exception:
                pass
        finally:
            if locked:
                try:
                    import fcntl
                    fcntl.flock(f, fcntl.LOCK_UN)
                except Exception:
                    pass
    try:
        dirfd = os.open(str(path.parent), os.O_DIRECTORY)
        try:
            os.fsync(dirfd)
        finally:
            os.close(dirfd)
    except Exception:
        pass


def _write_outbox(report_id, record):
    ctx.REPORTS_OUTBOX.mkdir(parents=True, exist_ok=True)
    try:
        os.chmod(ctx.REPORTS_OUTBOX, 0o700)
    except Exception:
        pass
    _assert_not_symlink(ctx.REPORTS_OUTBOX)
    out_path = ctx.REPORTS_OUTBOX / f"{report_id}.json"
    _assert_not_symlink(out_path)
    tmp = out_path.with_suffix(".tmp")
    fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        os.write(fd, (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8"))
        try:
            os.fsync(fd)
        except Exception:
            pass
    finally:
        os.close(fd)
    tmp.replace(out_path)
    try:
        dirfd = os.open(str(ctx.REPORTS_OUTBOX), os.O_DIRECTORY)
        try:
            os.fsync(dirfd)
        finally:
            os.close(dirfd)
    except Exception:
        pass
    return out_path


def _append_report(record):
    _atomic_jsonl_append(ctx.REPORTS_JSONL, record)
    _write_outbox(record["report_id"], record)
    log_operation({"op": "report", "report_id": record["report_id"], "status": "ok", "voice": record.get("voice"), "target": record.get("target")})


def _iter_voices():
    if not ctx.VOICES_DIR.exists():
        return []
    return [d for d in sorted(ctx.VOICES_DIR.iterdir(), key=lambda p: p.name.lower())
            if d.is_dir() and not d.name.startswith(".") and d.name != "base"]


def _registry_resolve(name_or_path):
    p = Path(name_or_path).expanduser()
    if p.is_file():
        return p, None
    if p.is_dir():
        for fname in ("voice.safetensors", "voice.gguf"):
            if (p / fname).exists():
                return p / fname, _read_meta(p / "voice.json")
    target = name_or_path.strip().lower()
    for d in _iter_voices():
        if d.name.lower() == target:
            for fname in ("voice.safetensors", "voice.gguf"):
                if (d / fname).exists():
                    return d / fname, _read_meta(d / "voice.json")
    return None, None


def _read_meta(json_path):
    try:
        return json.loads(json_path.read_text()) if json_path.exists() else None
    except Exception:
        return None


def _registry_name_exists(name):
    target = name.strip().lower()
    return any(d.name.lower() == target for d in _iter_voices())


def _registry_folder(name):
    return ctx.VOICES_DIR / name.strip().lower()


def _assert_not_symlink(path):
    # Resolve-first, fail-closed. The old code compared raw strings
    # (str.startswith): a sibling such as ".voices2" counted as "inside"
    # ".voices", uninspected parents were skipped by the same prefix test,
    # and any resolve failure was swallowed (fail-open). Now: resolve before
    # any comparison, test membership with real path relations, walk every
    # component below VOICES_DIR, and refuse when resolution itself fails.
    # Scope: registry paths (under VOICES_DIR). Unrelated user paths (e.g. an
    # explicit --out in /tmp) keep the old behavior: direct-symlink check only,
    # since the user chose them and system roots above them may legitimately
    # be links. TOCTOU (link swapped between check and write) is out of scope:
    # local CLI, self-owned dir.
    try:
        p = Path(path)
        abs_p = p.absolute()
        norm = Path(os.path.normpath(abs_p))
        base_abs = ctx.VOICES_DIR.absolute()
        base_res = ctx.VOICES_DIR.resolve()
        rp = p.resolve()
        if p.is_symlink():
            _fail(f"  ✗ Refusing to write through symlink: {path}")
        prefix_hit = str(abs_p).startswith(str(base_abs))
        inside = norm == base_abs or base_abs in norm.parents
        if inside or prefix_hit:
            # every component from the base down must be symlink-free; a
            # prefix-sibling (".voices2") has no relative leg to walk, but it
            # still faces the containment check below.
            if inside:
                cur = base_abs
                for part in norm.relative_to(base_abs).parts:
                    cur = cur / part
                    if cur.is_symlink():
                        _fail(f"  ✗ Refusing to write through symlink: {cur}")
            if rp != base_res and base_res not in rp.parents:
                _fail(f"  ✗ Refusing to write outside VOICES_DIR: {path}")
    except SystemExit:
        raise
    except Exception:
        _fail(f"  ✗ Refusing to write through unresolvable path: {path}")


def _write_voice(folder, tensor_name, fetched, meta):
    _assert_not_symlink(folder)
    _step("Saving voice…")
    folder.mkdir(parents=True, exist_ok=True)
    try:
        os.chmod(folder, 0o700)
    except Exception:
        pass
    _assert_not_symlink(folder)
    kind, raw_path, data, dtype, shape = fetched
    if _is_quant_fetched(fetched):
        fpath = folder / "voice.gguf"
        tmp = folder / "voice.gguf.tmp"
        _write_single_tensor_gguf(tensor_name, raw_path, raw_path.stat().st_size, dtype, shape, str(tmp))
    else:
        fpath = folder / "voice.safetensors"
        tmp = folder / "voice.safetensors.tmp"
        if kind == "file":
            _write_safetensors_streaming(tensor_name, str(raw_path), raw_path.stat().st_size, dtype, shape, str(tmp))
        else:
            write_safetensors({tensor_name: (data, dtype, shape)}, str(tmp))
    tmp.replace(fpath)
    try:
        os.chmod(fpath, 0o600)
    except Exception:
        pass
    meta = dict(meta)
    meta["artifact_format"] = "gguf" if _is_quant_fetched(fetched) else "safetensors"
    meta["ggml_type"] = dtype if _is_quant_fetched(fetched) else None
    meta["bytes"] = fpath.stat().st_size
    jtmp = folder / "voice.json.tmp"
    # atomic 0600 json write
    fd = os.open(str(jtmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        os.write(fd, (json.dumps(meta, indent=2) + "\n").encode("utf-8"))
        try:
            os.fsync(fd)
        except Exception:
            pass
    finally:
        os.close(fd)
    jtmp.replace(folder / "voice.json")
    return fpath


def _resolve_name(args, default_name):
    provided = getattr(args, "name", None)
    if provided and str(provided).strip():
        name = str(provided).strip()
        if _sanitize_voice_name(name) is None:
            _fail(f"  ✗ Invalid name '{name}'. {_NAME_RULES}")
        return name
    default_slug = _slugify_voice_name(default_name)
    if _noninteractive(args):
        return default_slug
    try:
        entered = input(f"  → Name the voice or press enter to use [{default_slug}]: ").strip()
        name = entered or default_slug
    except (EOFError, KeyboardInterrupt):
        _say("  ⚠ Cancelled.")
        sys.exit(1)
    if _sanitize_voice_name(name) is None:
        _fail(f"  ✗ Invalid name '{name}'. {_NAME_RULES}")
    return name


def _out_path(args, target, voice_arg):
    if getattr(args, "out", None):
        out = os.path.expanduser(args.out)
        # jail: resolve but allow any absolute path user explicitly requested; block symlink traversal
        if out.endswith(("/", os.sep)) or os.path.isdir(out) or not os.path.splitext(out)[1]:
            out = os.path.join(out, f"{_stem(voice_arg)}-{_stem(target)}{'.gguf' if _is_gguf_file(target) else '.safetensors'}")
        # normalize
        out = os.path.abspath(out)
        return out
    # sane default: ~/Voiced/<voice>-<model>/<voice>-<model>.gguf (not alongside target)
    # avoids overwriting and keeps Voiced folder tidy
    voiced_base = Path.home() / "Voiced"
    sub = f"{_stem(voice_arg)}-{_stem(target)}"
    ext = ".gguf" if _is_gguf_file(target) else ".safetensors"
    return str(voiced_base / sub / f"{sub}{ext}")


def _stem(p):
    return os.path.splitext(os.path.basename(p))[0]


def _check_overwrite(out_path, args):
    if os.path.islink(out_path):
        _fail(f"  ✗ Refusing to write through symlink: {out_path}")
    if os.path.exists(out_path):
        if _noninteractive(args) and not getattr(args, "yes", False):
            _fail(f"  ✗ Output '{out_path}' exists. Use --yes to overwrite in non-interactive mode.")
        if not _confirm(f"  Output '{out_path}' already exists. Overwrite? [y/N]: ", args):
            _say("  Cancelled.")
            sys.exit(1)
        _note(f"  Overwriting '{out_path}'.")
    out_dir = os.path.dirname(os.path.abspath(out_path))
    if out_dir and os.path.islink(out_dir):
        _fail(f"  ✗ Refusing to write through symlink: {out_dir}")
    if out_dir and not os.path.isdir(out_dir):
        try:
            os.makedirs(out_dir, exist_ok=True)
        except Exception:
            _fail(f"  ✗ Could not create folder \"{out_dir}\".")

	"""Voice registry, ops log, and write guards."""

	from pathlib import Path
	import json
	import os
	import sys
	import time
	from vlib import ctx
	from vlib.ui import _confirm, _fail, _noninteractive, _note, _say, _step, _warn
	from vlib.tensors import _is_gguf_file, _is_quant_fetched, _write_safetensors_streaming, _write_single_tensor_gguf, write_safetensors
	from vlib.sources import _NAME_RULES, _sanitize_voice_name, _slugify_voice_name


	def log_operation(record):
	record = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), **record}
	try:
	ctx.VOICES_DIR.mkdir(parents=True, exist_ok=True)
	try:
	os.chmod(ctx.VOICES_DIR, 0o700)
	except Exception:
	pass
	_assert_not_symlink(ctx.OPERATIONS_LOG)
	with open(ctx.OPERATIONS_LOG, "a") as f:
	f.write(json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n")
	try:
	f.flush()
	os.fsync(f.fileno())
	except Exception:
	pass
	try:
	os.chmod(ctx.OPERATIONS_LOG, 0o600)
	except Exception:
	pass
	try:
	dirfd = os.open(str(ctx.VOICES_DIR), os.O_DIRECTORY)
	try:
	os.fsync(dirfd)
	finally:
	os.close(dirfd)
	except Exception:
	pass
	except Exception:
	if ctx.VERBOSE:
	_warn("could not write operation log")


	def _atomic_jsonl_append(path, record):
	"""Append one JSON line atomically with flock + fsync. Optimised: single chmod, cached JSON."""
	path = Path(path)
	# fast path: parent exists, avoid mkdir/chmod churn
	if not path.parent.exists():
	path.parent.mkdir(parents=True, exist_ok=True)
	try:
	os.chmod(path.parent, 0o700)
	except Exception:
	pass
	if not path.exists():
	try:
	fd = os.open(str(path), os.O_WRONLY \| os.O_CREAT, 0o600)
	os.close(fd)
	except Exception:
	pass
	else:
	# occasional chmod drift check, not every append
	try:
	if oct(path.stat().st_mode & 0o777) != "0o600":
	os.chmod(path, 0o600)
	except Exception:
	pass
	_assert_not_symlink(path)
	line = json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n"
	with open(path, "a") as f:
	try:
	import fcntl
	fcntl.flock(f, fcntl.LOCK_EX)
	locked = True
	except Exception:
	locked = False
	try:
	f.write(line)
	f.flush()
	try:
	os.fsync(f.fileno())
	except Exception:
	pass
	finally:
	if locked:
	try:
	import fcntl
	fcntl.flock(f, fcntl.LOCK_UN)
	except Exception:
	pass
	try:
	dirfd = os.open(str(path.parent), os.O_DIRECTORY)
	try:
	os.fsync(dirfd)
	finally:
	os.close(dirfd)
	except Exception:
	pass


	def _write_outbox(report_id, record):
	ctx.REPORTS_OUTBOX.mkdir(parents=True, exist_ok=True)
	try:
	os.chmod(ctx.REPORTS_OUTBOX, 0o700)
	except Exception:
	pass
	_assert_not_symlink(ctx.REPORTS_OUTBOX)
	out_path = ctx.REPORTS_OUTBOX / f"{report_id}.json"
	_assert_not_symlink(out_path)
	tmp = out_path.with_suffix(".tmp")
	fd = os.open(str(tmp), os.O_WRONLY \| os.O_CREAT \| os.O_TRUNC, 0o600)
	try:
	os.write(fd, (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8"))
	try:
	os.fsync(fd)
	except Exception:
	pass
	finally:
	os.close(fd)
	tmp.replace(out_path)
	try:
	dirfd = os.open(str(ctx.REPORTS_OUTBOX), os.O_DIRECTORY)
	try:
	os.fsync(dirfd)
	finally:
	os.close(dirfd)
	except Exception:
	pass
	return out_path


	def _append_report(record):
	_atomic_jsonl_append(ctx.REPORTS_JSONL, record)
	_write_outbox(record["report_id"], record)
	log_operation({"op": "report", "report_id": record["report_id"], "status": "ok", "voice": record.get("voice"), "target": record.get("target")})


	def _iter_voices():
	if not ctx.VOICES_DIR.exists():
	return []
	return [d for d in sorted(ctx.VOICES_DIR.iterdir(), key=lambda p: p.name.lower())
	if d.is_dir() and not d.name.startswith(".") and d.name != "base"]


	def _registry_resolve(name_or_path):
	p = Path(name_or_path).expanduser()
	if p.is_file():
	return p, None
	if p.is_dir():
	for fname in ("voice.safetensors", "voice.gguf"):
	if (p / fname).exists():
	return p / fname, _read_meta(p / "voice.json")
	target = name_or_path.strip().lower()
	for d in _iter_voices():
	if d.name.lower() == target:
	for fname in ("voice.safetensors", "voice.gguf"):
	if (d / fname).exists():
	return d / fname, _read_meta(d / "voice.json")
	return None, None


	def _read_meta(json_path):
	try:
	return json.loads(json_path.read_text()) if json_path.exists() else None
	except Exception:
	return None


	def _registry_name_exists(name):
	target = name.strip().lower()
	return any(d.name.lower() == target for d in _iter_voices())


	def _registry_folder(name):
	return ctx.VOICES_DIR / name.strip().lower()


	def _assert_not_symlink(path):
	# Resolve-first, fail-closed. The old code compared raw strings
	# (str.startswith): a sibling such as ".voices2" counted as "inside"
	# ".voices", uninspected parents were skipped by the same prefix test,
	# and any resolve failure was swallowed (fail-open). Now: resolve before
	# any comparison, test membership with real path relations, walk every
	# component below VOICES_DIR, and refuse when resolution itself fails.
	# Scope: registry paths (under VOICES_DIR). Unrelated user paths (e.g. an
	# explicit --out in /tmp) keep the old behavior: direct-symlink check only,
	# since the user chose them and system roots above them may legitimately
	# be links. TOCTOU (link swapped between check and write) is out of scope:
	# local CLI, self-owned dir.
	try:
	p = Path(path)
	abs_p = p.absolute()
	norm = Path(os.path.normpath(abs_p))
	base_abs = ctx.VOICES_DIR.absolute()
	base_res = ctx.VOICES_DIR.resolve()
	rp = p.resolve()
	if p.is_symlink():
	_fail(f" ✗ Refusing to write through symlink: {path}")
	prefix_hit = str(abs_p).startswith(str(base_abs))
	inside = norm == base_abs or base_abs in norm.parents
	if inside or prefix_hit:
	# every component from the base down must be symlink-free; a
	# prefix-sibling (".voices2") has no relative leg to walk, but it
	# still faces the containment check below.
	if inside:
	cur = base_abs
	for part in norm.relative_to(base_abs).parts:
	cur = cur / part
	if cur.is_symlink():
	_fail(f" ✗ Refusing to write through symlink: {cur}")
	if rp != base_res and base_res not in rp.parents:
	_fail(f" ✗ Refusing to write outside VOICES_DIR: {path}")
	except SystemExit:
	raise
	except Exception:
	_fail(f" ✗ Refusing to write through unresolvable path: {path}")


	def _write_voice(folder, tensor_name, fetched, meta):
	_assert_not_symlink(folder)
	_step("Saving voice…")
	folder.mkdir(parents=True, exist_ok=True)
	try:
	os.chmod(folder, 0o700)
	except Exception:
	pass
	_assert_not_symlink(folder)
	kind, raw_path, data, dtype, shape = fetched
	if _is_quant_fetched(fetched):
	fpath = folder / "voice.gguf"
	tmp = folder / "voice.gguf.tmp"
	_write_single_tensor_gguf(tensor_name, raw_path, raw_path.stat().st_size, dtype, shape, str(tmp))
	else:
	fpath = folder / "voice.safetensors"
	tmp = folder / "voice.safetensors.tmp"
	if kind == "file":
	_write_safetensors_streaming(tensor_name, str(raw_path), raw_path.stat().st_size, dtype, shape, str(tmp))
	else:
	write_safetensors({tensor_name: (data, dtype, shape)}, str(tmp))
	tmp.replace(fpath)
	try:
	os.chmod(fpath, 0o600)
	except Exception:
	pass
	meta = dict(meta)
	meta["artifact_format"] = "gguf" if _is_quant_fetched(fetched) else "safetensors"
	meta["ggml_type"] = dtype if _is_quant_fetched(fetched) else None
	meta["bytes"] = fpath.stat().st_size
	jtmp = folder / "voice.json.tmp"
	# atomic 0600 json write
	fd = os.open(str(jtmp), os.O_WRONLY \| os.O_CREAT \| os.O_TRUNC, 0o600)
	try:
	os.write(fd, (json.dumps(meta, indent=2) + "\n").encode("utf-8"))
	try:
	os.fsync(fd)
	except Exception:
	pass
	finally:
	os.close(fd)
	jtmp.replace(folder / "voice.json")
	return fpath


	def _resolve_name(args, default_name):
	provided = getattr(args, "name", None)
	if provided and str(provided).strip():
	name = str(provided).strip()
	if _sanitize_voice_name(name) is None:
	_fail(f" ✗ Invalid name '{name}'. {_NAME_RULES}")
	return name
	default_slug = _slugify_voice_name(default_name)
	if _noninteractive(args):
	return default_slug
	try:
	entered = input(f" → Name the voice or press enter to use [{default_slug}]: ").strip()
	name = entered or default_slug
	except (EOFError, KeyboardInterrupt):
	_say(" ⚠ Cancelled.")
	sys.exit(1)
	if _sanitize_voice_name(name) is None:
	_fail(f" ✗ Invalid name '{name}'. {_NAME_RULES}")
	return name


	def _out_path(args, target, voice_arg):
	if getattr(args, "out", None):
	out = os.path.expanduser(args.out)
	# jail: resolve but allow any absolute path user explicitly requested; block symlink traversal
	if out.endswith(("/", os.sep)) or os.path.isdir(out) or not os.path.splitext(out)[1]:
	out = os.path.join(out, f"{_stem(voice_arg)}-{_stem(target)}{'.gguf' if _is_gguf_file(target) else '.safetensors'}")
	# normalize
	out = os.path.abspath(out)
	return out
	# sane default: ~/Voiced/<voice>-<model>/<voice>-<model>.gguf (not alongside target)
	# avoids overwriting and keeps Voiced folder tidy
	voiced_base = Path.home() / "Voiced"
	sub = f"{_stem(voice_arg)}-{_stem(target)}"
	ext = ".gguf" if _is_gguf_file(target) else ".safetensors"
	return str(voiced_base / sub / f"{sub}{ext}")


	def _stem(p):
	return os.path.splitext(os.path.basename(p))[0]


	def _check_overwrite(out_path, args):
	if os.path.islink(out_path):
	_fail(f" ✗ Refusing to write through symlink: {out_path}")
	if os.path.exists(out_path):
	if _noninteractive(args) and not getattr(args, "yes", False):
	_fail(f" ✗ Output '{out_path}' exists. Use --yes to overwrite in non-interactive mode.")
	if not _confirm(f" Output '{out_path}' already exists. Overwrite? [y/N]: ", args):
	_say(" Cancelled.")
	sys.exit(1)
	_note(f" Overwriting '{out_path}'.")
	out_dir = os.path.dirname(os.path.abspath(out_path))
	if out_dir and os.path.islink(out_dir):
	_fail(f" ✗ Refusing to write through symlink: {out_dir}")
	if out_dir and not os.path.isdir(out_dir):
	try:
	os.makedirs(out_dir, exist_ok=True)
	except Exception:
	_fail(f" ✗ Could not create folder \"{out_dir}\".")