feat(voices): export and import voice bundles

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
mARTin-B78 2026-10-04 16:42:28 +02:00
parent 781bf6e7f2
commit e0daafd4e1
11 changed files with 550 additions and 2 deletions

View File

@ -7,6 +7,7 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
## [Unreleased]
### Added
- **Voice export/import bundles.** `GET /api/voices/{id}/export` streams a `<id>.voice.zip` (audio, `reference.txt`, `meta.json` with seed, design prompt, rating/tags, `provenance`, `audio_problems`; optional picture and consent recording; `manifest.json` with format version, app version and sha256 per file; no temp files). `POST /api/voices/import` validates at the boundary: 150 MB upload cap, fixed flat member names only (zip-slip/path traversal rejected), per-file, total and compression-ratio caps, hashes verified, audio decoded, meta and provenance re-validated; a taken name gets a `_2` suffix, an identical voice is a no-op, nothing is overwritten. Voice library toolbar: "Export selected" and "Import bundle" (44px). Tests: `tests/test_voice_bundle.py`, `tests/voice-bundle.spec.js`. See `architecture/voice-bundle.md`.
- **Consent/provenance field per voice.** Each voice gets `provenance` in its `.meta.json` (`basis`: own_voice, consented, licensed, public_domain or unknown; `source`, `licence`, `note`, optional `recorded_consent`, `set_at`). Existing voices count as `unknown`; `quick-import-voice` prefills the source URL (basis stays `unknown`). New `PUT /api/voices/{id}/provenance` (enum, length caps, control characters stripped, http(s) URLs only, no credentials) and `POST`/`GET /api/voices/{id}/consent-recording` (audio only, 10 MB cap, stored as `<voice>.consent`, moves/deletes with the voice). The voice inspector has a "Source & consent" section (typed note or MediaRecorder recording) and voices without a recorded basis show a "Source unknown" badge. Generation is not blocked. Data stays local. Tests: `tests/test_voice_provenance.py`, `tests/voice-provenance.spec.js`.
- **Seed Finder: Auto-pick best.** `POST /api/seed-autopick {voice, text, seeds|count}` (default 8, max 16) generates candidates sequentially through the existing temporary-voice path (voices.json is always restored), scores each 0-100 (ASR word accuracy, loudness/clipping, `audio_problems`) and returns all sub-scores plus the winner. Degrades to loudness/problems when STT is unavailable. The UI highlights the winner; saving stays a click on "Use seed". Tests: `tests/test_seed_autopick.py`, `tests/seed-autopick.spec.js`.
- **Voice Design: attribute chips.** Toggle chips (gender, age, pitch, pace, accent/region, tone/timbre, emotion; German + English labels) above the design prompt add or remove only their own phrase in the free-text description. The textarea stays fully free-form. Test: `tests/design-chips.spec.js`.

View File

@ -14,6 +14,16 @@ Living manual. Every feature updates this file in the same commit as the code.
---
## Voice export and import bundles
Purpose: back up a voice or move it between machines (DGX and NUC) as one `.voice.zip` file containing the reference audio, transcript, settings (seed, design prompt, rating, tags, source & consent, audio problems), and optionally the picture and the consent recording.
Prerequisites: Voice library open. Export needs at least one checked voice.
Steps: tick the checkboxes of the voices, press "Export selected"; one `<voice>.voice.zip` is downloaded per voice. On the other machine press "Import bundle" and choose one or more `.voice.zip` files.
Expected result: the voice appears in the library. If the name is already taken the import gets a suffix (`_2`, `_3`); an identical voice that is already there is reported as "already present". Existing voices are never overwritten.
Errors: damaged, tampered, oversized or unexpected files are rejected with a message (for example "Hash mismatch"); nothing is imported then. Messages appear as toasts, which screen readers announce.
Mobile / accessibility: both buttons are at least 44 px high and keyboard operable; the file chooser is the browser's own.
Offline: works fully offline. Privacy: a bundle contains the voice, its source/consent notes and any consent recording; share it only with people you trust.
## Voice source & consent
Purpose: record where a voice comes from and why you may use it (own voice, speaker consented, licensed, public domain, unknown).

View File

@ -0,0 +1,13 @@
# Voice export/import bundles
Code: `core/voice_bundle.py`; routes in `routes/library.py` (`GET /api/voices/{id}/export`, `POST /api/voices/import`); UI in `static/js/voice-library.js` + toolbar in `static/sections/s-voices.html`.
Format (v1), flat ZIP, fixed member names only: `manifest.json`, `audio.<wav|mp3|m4a|flac|ogg|opus>`, `meta.json`, `reference.txt`, `picture.<jpg|jpeg|png|webp>`, `consent.<audio ext>`. Manifest: `format` (1), `app_version` (from `VERSION`), `voice_id`, `files` {member: sha256}. One voice per bundle; multi-select exports one bundle per voice. `meta.json` is the voice sidecar plus `audio_problems`; `provenance.recorded_consent` is only kept when a consent member exists.
Export: streamed through a non-seekable ZIP writer (no temp files); manifest is written last so hashes are computed while streaming.
Import threat model (bundle is hostile): upload cap 150 MB (`_copy_limited`); member names must be in the fixed set, so absolute paths, `..`, subdirectories and unknown extensions are rejected before anything is read; names are never used as paths (staged under fixed names in a hidden `.import-*` dir inside the voices dir, removed in `finally`); declared sizes are checked and real bytes are re-capped while reading; total uncompressed 200 MB, ratio cap 200 for members over 1 MB; manifest must list exactly the members and every sha256 must match; voice id must match `[A-Za-z0-9_-][A-Za-z0-9_.-]{0,99}`; audio must decode; `meta.json` goes through `_validate_voice_meta` and `validate_provenance` (set_at is reset). Nothing is executed.
Name handling: never overwrite. Same id with identical audio hash -> `unchanged` (idempotent). Otherwise `_2`, `_3`, ... (max 99) -> `renamed`. Installed into the active voices dir and added to the voice index.
Not done: no signature (hashes detect corruption/tampering of members but not a re-built manifest); explicit overwrite/skip choice; one request per bundle for multi-file import.

231
core/voice_bundle.py Normal file
View File

@ -0,0 +1,231 @@
"""Voice export/import bundles (ZIP). See architecture/voice-bundle.md.
Export streams a ZIP (no temp files). Import treats the archive as hostile:
only fixed flat member names, capped sizes, hashes verified, nothing executed,
no path from the archive is ever used.
"""
from __future__ import annotations
import hashlib
import json
import re
import shutil
import tempfile
import zipfile
from pathlib import Path
from typing import Iterator
from core.provenance import CONSENT_EXTS, MAX_CONSENT_BYTES, validate_provenance
from core.voice import (
_AUDIO_EXTS, _PICTURE_EXTS, _find_voice_audio, _load_meta,
_save_meta, _validate_voice_meta, _voice_problems,
)
FORMAT_VERSION = 1
MAX_BUNDLE_BYTES = 150 * 1024 * 1024 # compressed upload
MAX_TOTAL_BYTES = 200 * 1024 * 1024 # uncompressed, all members
MAX_AUDIO_BYTES = 150 * 1024 * 1024
MAX_PICTURE_BYTES = 10 * 1024 * 1024
MAX_TEXT_BYTES = 256 * 1024
MAX_RATIO = 200 # per-member compression ratio (zip-bomb)
ID_RE = re.compile(r"^[A-Za-z0-9_\-][A-Za-z0-9_\-.]{0,99}$")
_PICTURE_SET = {e.lstrip(".") for e in _PICTURE_EXTS}
_AUDIO_SET = {e.lstrip(".") for e in _AUDIO_EXTS}
_CONSENT_SET = {e.lstrip(".") for e in CONSENT_EXTS}
class BundleError(ValueError):
"""Invalid bundle (-> HTTP 400)."""
def _limit_for(name: str) -> int | None:
if name in ("manifest.json", "meta.json", "reference.txt"):
return MAX_TEXT_BYTES
stem, _, ext = name.rpartition(".")
if stem == "audio" and ext in _AUDIO_SET:
return MAX_AUDIO_BYTES
if stem == "picture" and ext in _PICTURE_SET:
return MAX_PICTURE_BYTES
if stem == "consent" and ext in _CONSENT_SET:
return MAX_CONSENT_BYTES
return None
def _app_version() -> str:
try:
return (Path(__file__).resolve().parents[1] / "VERSION").read_text().strip()
except OSError:
return "unknown"
# ── export ────────────────────────────────────────────────────────────────────
class _Sink:
"""Non-seekable write target; the generator drains it between chunks."""
def __init__(self): self.buf = bytearray(); self.pos = 0
def write(self, b): self.buf += b; self.pos += len(b); return len(b)
def tell(self): return self.pos
def flush(self): pass
def drain(self):
out = bytes(self.buf); self.buf.clear(); return out
def export_members(wav: Path) -> tuple[dict, list[tuple[str, Path | bytes]]]:
"""Return (meta_for_bundle, [(member_name, path-or-bytes)]) for a voice."""
meta = _load_meta(wav)
meta["audio_problems"] = list(meta.get("audio_problems") or _voice_problems(wav))
members: list[tuple[str, Path | bytes]] = [(f"audio{wav.suffix.lower()}", wav)]
ref = wav.with_suffix(".reference.txt")
if ref.is_file():
members.append(("reference.txt", ref))
for ext in _PICTURE_EXTS:
pic = wav.with_suffix(ext)
if pic.is_file():
members.append((f"picture{ext}", pic))
break
consent = wav.with_suffix(".consent")
rec = (meta.get("provenance") or {}).get("recorded_consent") or ""
if consent.is_file() and Path(rec).suffix.lower() in CONSENT_EXTS:
members.append((f"consent{Path(rec).suffix.lower()}", consent))
members.append(("meta.json", (json.dumps(meta, indent=2, ensure_ascii=False) + "\n").encode()))
return meta, members
def stream_bundle(voice_id: str, members: list[tuple[str, Path | bytes]]) -> Iterator[bytes]:
"""Yield a ZIP; manifest.json (with sha256 per file) is written last."""
sink, hashes = _Sink(), {}
with zipfile.ZipFile(sink, "w", zipfile.ZIP_DEFLATED) as zf:
for name, src in members:
h = hashlib.sha256()
with zf.open(name, "w", force_zip64=True) as out:
if isinstance(src, bytes):
h.update(src); out.write(src)
else:
with src.open("rb") as f:
while chunk := f.read(1024 * 1024):
h.update(chunk); out.write(chunk)
if sink.buf:
yield sink.drain()
hashes[name] = h.hexdigest()
yield sink.drain()
manifest = {"format": FORMAT_VERSION, "app_version": _app_version(),
"voice_id": voice_id, "files": hashes}
zf.writestr("manifest.json", json.dumps(manifest, indent=2))
yield sink.drain()
# ── import ────────────────────────────────────────────────────────────────────
def _read_member(zf: zipfile.ZipFile, info: zipfile.ZipInfo, limit: int, dest: Path | None = None) -> tuple[bytes, str]:
"""Read at most `limit` bytes (declared size is not trusted); returns (data|b'', sha256)."""
h, total, data = hashlib.sha256(), 0, bytearray()
out = dest.open("wb") if dest else None
try:
with zf.open(info) as f:
while chunk := f.read(1024 * 1024):
total += len(chunk)
if total > limit:
raise BundleError(f"Bundle member too large: {info.filename}")
h.update(chunk)
(out.write(chunk) if out else data.extend(chunk))
finally:
if out: out.close()
return bytes(data), h.hexdigest()
def import_bundle(zip_path: Path, voices_dir: Path, scan_dir: Path, decode_check=None) -> dict:
"""Validate and install a bundle. Returns {voice_id, status, wav}.
status: "imported" | "renamed" (name taken -> suffix) | "unchanged" (identical already present).
`decode_check(path)` must raise if the audio is not decodable.
"""
if not zipfile.is_zipfile(zip_path):
raise BundleError("Not a ZIP file")
voices_dir.mkdir(parents=True, exist_ok=True)
stage = Path(tempfile.mkdtemp(prefix=".import-", dir=voices_dir))
try:
with zipfile.ZipFile(zip_path) as zf:
infos = zf.infolist()
names = [i.filename for i in infos]
if len(set(names)) != len(names):
raise BundleError("Duplicate members in bundle")
total = 0
for i in infos:
if i.is_dir() or _limit_for(i.filename) is None: # also rejects any path / traversal / unknown name
raise BundleError("Unexpected member in bundle")
if i.file_size > _limit_for(i.filename) or (i.compress_size and i.file_size / i.compress_size > MAX_RATIO and i.file_size > 1024 * 1024):
raise BundleError("Bundle member too large or suspiciously compressed")
total += i.file_size
if total > MAX_TOTAL_BYTES:
raise BundleError("Bundle is too large when unpacked")
by_name = {i.filename: i for i in infos}
if "manifest.json" not in by_name:
raise BundleError("manifest.json missing")
raw, _ = _read_member(zf, by_name["manifest.json"], MAX_TEXT_BYTES)
try:
manifest = json.loads(raw)
except ValueError:
raise BundleError("manifest.json is not valid JSON")
if not isinstance(manifest, dict) or manifest.get("format") != FORMAT_VERSION:
raise BundleError("Unsupported bundle format version")
files = manifest.get("files")
if not isinstance(files, dict) or set(files) != set(by_name) - {"manifest.json"}:
raise BundleError("Manifest does not match bundle contents")
voice_id = manifest.get("voice_id")
if not isinstance(voice_id, str) or not ID_RE.match(voice_id) or ".." in voice_id:
raise BundleError("Invalid voice id in manifest")
audio = [n for n in files if n.startswith("audio.")]
if len(audio) != 1:
raise BundleError("Bundle needs exactly one audio file")
staged: dict[str, Path] = {}
for n, expected in files.items():
p = stage / n # n is a validated fixed name, never a path from the archive
_, digest = _read_member(zf, by_name[n], _limit_for(n), p)
if digest != expected:
raise BundleError(f"Hash mismatch: {n}")
staged[n] = p
# decode + metadata re-validation (nothing is installed yet)
if decode_check:
try:
decode_check(staged[audio[0]])
except Exception:
raise BundleError("Audio file cannot be decoded")
try:
meta = json.loads(staged["meta.json"].read_bytes()) if "meta.json" in staged else {}
consent = next((n for n in staged if n.startswith("consent.")), None)
if "provenance" in meta:
prev = {"recorded_consent": "x"} if consent else None
meta["provenance"] = validate_provenance(meta["provenance"], prev)
meta["provenance"].pop("recorded_consent", None)
if consent:
meta["provenance"]["recorded_consent"] = "consent" + Path(consent).suffix
_validate_voice_meta(meta)
except (ValueError, TypeError, AttributeError) as exc:
raise BundleError(f"Invalid metadata: {exc}")
# name handling: never overwrite; identical audio already present -> idempotent no-op
new_hash = files[audio[0]]
suffix = Path(audio[0]).suffix
vid, status, n = voice_id, "imported", 1
while (existing := _find_voice_audio(vid, scan_dir)) is not None or (voices_dir / f"{vid}{suffix}").exists():
if existing is not None and hashlib.sha256(existing.read_bytes()).hexdigest() == new_hash:
return {"voice_id": vid, "status": "unchanged", "wav": str(existing)}
n += 1
vid, status = f"{voice_id}_{n}", "renamed"
if n > 99:
raise BundleError("Too many voices with this name")
if consent:
meta.get("provenance", {}).update(recorded_consent=f"{vid}{Path(consent).suffix}")
dest = voices_dir / f"{vid}{suffix}"
shutil.move(str(staged[audio[0]]), dest)
if "reference.txt" in staged:
shutil.move(str(staged["reference.txt"]), dest.with_suffix(".reference.txt"))
for n_, p in staged.items():
if n_.startswith("picture."):
shutil.move(str(p), dest.with_suffix(Path(n_).suffix))
elif consent and n_ == consent:
shutil.move(str(p), dest.with_suffix(".consent"))
meta.pop("needs_tts_restart", None)
_save_meta(dest, meta)
return {"voice_id": vid, "status": status, "wav": str(dest)}
finally:
shutil.rmtree(stage, ignore_errors=True)

View File

@ -9,7 +9,7 @@ from datetime import datetime
from pathlib import Path
from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import FileResponse, Response
from fastapi.responses import FileResponse, Response, StreamingResponse
from core.config import _load_settings
from core.constants import _VOICES_DIR_DEFAULT, _OUTPUT_DIR_DEFAULT, _VOICE_TARGET_DBFS, _VOICE_PEAK_DBFS, _BENCHMARK_SENTENCES, _MAX_TTS_OUTPUT_SECONDS, _MAX_UPLOAD_BYTES, _MAX_PICTURE_BYTES, _MAX_SOUND_BYTES, _ALLOW_PRIVATE_DOWNLOADS
@ -551,6 +551,39 @@ async def serve_consent_recording(voice_id: str):
return FileResponse(str(f), media_type=mime, headers={"Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"})
@router.get("/api/voices/{voice_id}/export")
async def export_voice_bundle(voice_id: str):
from core import voice_bundle as vb
wav, _ = _provenance_voice(voice_id)
_, members = await asyncio.to_thread(vb.export_members, wav)
safe = re.sub(r"[^A-Za-z0-9_.-]", "_", wav.stem)
return StreamingResponse(vb.stream_bundle(wav.stem, members), media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{safe}.voice.zip"',
"Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"})
@router.post("/api/voices/import")
async def import_voice_bundle(file: UploadFile = File(...)):
from pydub import AudioSegment
from core import voice_bundle as vb
settings = _load_settings()
tmp = TEMP_DIR / f"bundle-{uuid.uuid4().hex}.zip"
try:
with tmp.open("wb") as f:
_copy_limited(file.file, f, vb.MAX_BUNDLE_BYTES)
try:
res = await asyncio.to_thread(
vb.import_bundle, tmp, _active_voices_dir(settings),
Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)),
lambda p: AudioSegment.from_file(str(p)))
except vb.BundleError as exc:
raise HTTPException(400, str(exc))
finally:
tmp.unlink(missing_ok=True)
await asyncio.to_thread(upsert_voice_in_index, settings, Path(res["wav"]))
return {"ok": True, "voice_id": res["voice_id"], "status": res["status"]}
@router.post("/api/voices/sync-folders")
async def sync_voice_folders():
settings = _load_settings()

File diff suppressed because one or more lines are too long

View File

@ -720,6 +720,56 @@ async function loadVoiceLibrary(options = {}) {
}
$('refresh-voices-btn').addEventListener('click', () => { _vlCacheClear(); loadVoiceLibrary({ refresh: true }); });
// ── Voice bundles (export / import .voice.zip) ───────────────────────────────
async function exportVoiceBundles(ids) {
let ok = 0;
for (const id of ids) {
try {
const r = await fetch('/api/voices/' + encodeURIComponent(id) + '/export');
if (!r.ok) { const e = await r.json().catch(() => ({})); throw new Error(e.detail || r.statusText); }
const url = URL.createObjectURL(await r.blob());
const a = document.createElement('a');
a.href = url; a.download = id + '.voice.zip';
document.body.appendChild(a); a.click(); a.remove();
setTimeout(() => URL.revokeObjectURL(url), 10000);
ok++;
} catch (e) { toast(`Export of ${id} failed: ${e.message}`, 'error'); }
}
if (ok) toast(`Exported ${ok} voice bundle${ok === 1 ? '' : 's'}`, 'success');
return ok;
}
async function importVoiceBundles(files) {
const msgs = []; let failed = 0;
for (const f of files) {
try {
const fd = new FormData(); fd.append('file', f);
const r = await fetch('/api/voices/import', { method: 'POST', body: fd });
const d = await r.json().catch(() => ({}));
if (!r.ok) throw new Error(d.detail || r.statusText);
msgs.push(d.status === 'unchanged' ? `${d.voice_id} already present` : d.status === 'renamed' ? `${d.voice_id} (name was taken)` : d.voice_id);
} catch (e) { failed++; toast(`Import of ${f.name} failed: ${e.message}`, 'error'); }
}
if (msgs.length) {
if (typeof window.invalidateVoiceLibraryCache === 'function') window.invalidateVoiceLibraryCache();
await loadVoiceLibrary({ refresh: true });
toast('Imported: ' + msgs.join(', '), failed ? 'error' : 'success');
}
return msgs.length;
}
window.exportVoiceBundles = exportVoiceBundles;
window.importVoiceBundles = importVoiceBundles;
$('export-voices-btn').addEventListener('click', async () => {
if (!_bulkSelected.size) { toast('Check one or more voices first', 'error'); return; }
$('export-voices-btn').disabled = true;
try { await exportVoiceBundles([..._bulkSelected]); } finally { $('export-voices-btn').disabled = false; }
});
$('import-voices-btn').addEventListener('click', () => $('import-voices-input').click());
$('import-voices-input').addEventListener('change', async ev => {
const files = [...ev.target.files]; ev.target.value = '';
if (!files.length) return;
$('import-voices-btn').disabled = true;
try { await importVoiceBundles(files); } finally { $('import-voices-btn').disabled = false; }
});
$('sync-voice-folders-btn').addEventListener('click', async () => {
$('sync-voice-folders-btn').disabled = true;
status('Syncing active_voices and hidden_voices…');

View File

@ -136,6 +136,9 @@
<div class="vl-toolbar">
<button class="btn-secondary vl-tb-btn" id="refresh-voices-btn" title="Refresh voice list"><span class="mdi mdi-refresh"></span> Refresh</button>
<button class="btn-secondary vl-tb-btn" id="sync-voice-folders-btn" title="Sync active/hidden folders">Sync</button>
<button class="btn-secondary vl-tb-btn vl-tb-bundle" id="export-voices-btn" title="Export the checked voices as .voice.zip bundles (backup / share)"><span class="mdi mdi-export" aria-hidden="true"></span> Export selected</button>
<button class="btn-secondary vl-tb-btn vl-tb-bundle" id="import-voices-btn" title="Import a .voice.zip bundle; an existing voice is never overwritten"><span class="mdi mdi-import" aria-hidden="true"></span> Import bundle</button>
<input type="file" id="import-voices-input" accept=".zip,application/zip" multiple hidden>
<button class="btn-secondary vl-tb-btn" id="calculate-db-btn" title="Calculate dBFS">Calc dB</button>
<button class="btn-secondary vl-tb-btn" id="benchmark-voices-btn" title="Benchmark TTS speed">Benchmark</button>
<button class="btn-secondary vl-tb-btn" id="copy-active-voices-btn" title="Copy selected voices (or all active if none checked)">Copy selected</button>

View File

@ -2325,6 +2325,7 @@ code { background: var(--panel); border-radius: 4px; padding: 1px 5px; font-fami
padding: 8px 10px 7px; border-bottom: 1px solid var(--border);
}
.vl-tb-btn { padding: 3px 8px !important; font-size: 12px !important; flex: 0 0 auto; }
.vl-tb-bundle { min-height: 44px; min-width: 44px; }
.vl-filters {
display: flex; align-items: center; gap: 5px; flex-wrap: wrap;
padding: 7px 10px; border-bottom: 1px solid var(--border);

156
tests/test_voice_bundle.py Normal file
View File

@ -0,0 +1,156 @@
"""Voice export/import bundles: roundtrip and hostile-input rejection (temp dirs only)."""
import hashlib
import io
import json
import os
import sys
import tempfile
import unittest
import wave
import zipfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-bundle-test-config-"))
from core import voice_bundle as vb
from core.voice import _load_meta, _save_meta
def _wav(path: Path):
with wave.open(str(path), "wb") as w:
w.setnchannels(1); w.setsampwidth(2); w.setframerate(8000); w.writeframes(b"\x01\x00" * 800)
def _decode(p: Path):
with wave.open(str(p), "rb") as w:
assert w.getnframes() > 0
def _zip(files: dict, manifest=None, voice_id="EN_F_a"):
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
for n, d in files.items():
zf.writestr(n, d)
if manifest is not False:
m = manifest or {"format": 1, "voice_id": voice_id,
"files": {n: hashlib.sha256(d).hexdigest() for n, d in files.items()}}
zf.writestr("manifest.json", json.dumps(m))
return buf.getvalue()
class BundleTests(unittest.TestCase):
def setUp(self):
self.tmp = Path(tempfile.mkdtemp(prefix="ttvc-bundle-"))
self.src, self.dst = self.tmp / "src", self.tmp / "dst"
self.src.mkdir(); self.dst.mkdir()
self.wav_bytes = self._wav_bytes()
def tearDown(self):
import shutil; shutil.rmtree(self.tmp, ignore_errors=True)
def _wav_bytes(self):
p = self.tmp / "x.wav"; _wav(p); return p.read_bytes()
def _install(self, data: bytes, **kw):
z = self.tmp / "in.zip"; z.write_bytes(data)
return vb.import_bundle(z, self.dst, self.dst, decode_check=_decode)
def _good(self, **extra):
files = {"audio.wav": self.wav_bytes, "meta.json": json.dumps({"rating": 4, "seed": 7, **extra}).encode()}
return files
def test_roundtrip_all_parts(self):
wav = self.src / "EN_F_a.wav"; _wav(wav)
wav.with_suffix(".reference.txt").write_text("hello")
wav.with_suffix(".png").write_bytes(b"\x89PNG fake")
wav.with_suffix(".consent").write_bytes(b"consentaudio")
_save_meta(wav, {"rating": 3, "voice_design_prompt": "warm", "audio_problems": [],
"provenance": {"basis": "consented", "source": "https://x.org", "recorded_consent": "EN_F_a.webm"}})
_, members = vb.export_members(wav)
data = b"".join(vb.stream_bundle("EN_F_a", members))
self.assertTrue(zipfile.is_zipfile(io.BytesIO(data)))
res = self._install(data)
self.assertEqual(res["status"], "imported")
out = self.dst / "EN_F_a.wav"
self.assertEqual(out.read_bytes(), wav.read_bytes())
self.assertEqual((self.dst / "EN_F_a.reference.txt").read_text(), "hello")
self.assertEqual((self.dst / "EN_F_a.consent").read_bytes(), b"consentaudio")
self.assertTrue((self.dst / "EN_F_a.png").exists())
meta = _load_meta(out)
self.assertEqual(meta["voice_design_prompt"], "warm")
self.assertEqual(meta["provenance"]["basis"], "consented")
self.assertEqual(meta["provenance"]["recorded_consent"], "EN_F_a.webm")
self.assertEqual([p.name for p in self.dst.iterdir() if p.name.startswith(".import")], [])
def test_collision_suffix_and_idempotent(self):
data = _zip(self._good())
self.assertEqual(self._install(data)["status"], "imported")
self.assertEqual(self._install(data)["status"], "unchanged") # same audio again: no duplicate
self.assertEqual(len(list(self.dst.glob("*.wav"))), 1)
other = _zip({**self._good(), "audio.wav": self.wav_bytes + b"\x00\x00"})
r = self._install(other)
self.assertEqual((r["status"], r["voice_id"]), ("renamed", "EN_F_a_2"))
self.assertTrue((self.dst / "EN_F_a.wav").exists()) # original untouched
def test_missing_optional_parts(self):
r = self._install(_zip({"audio.wav": self.wav_bytes}))
self.assertEqual(r["status"], "imported")
self.assertFalse((self.dst / "EN_F_a.consent").exists())
def test_tampered_hash(self):
files = self._good()
m = {"format": 1, "voice_id": "EN_F_a", "files": {n: hashlib.sha256(d).hexdigest() for n, d in files.items()}}
files["audio.wav"] = self.wav_bytes + b"tamper"
with self.assertRaisesRegex(vb.BundleError, "Hash mismatch"):
self._install(_zip(files, manifest=m))
self.assertEqual(list(self.dst.iterdir()), [])
def test_zip_slip_and_unknown_members(self):
for bad in ("../evil.wav", "/etc/passwd", "sub/audio.wav", "run.sh", "audio.exe"):
with self.assertRaises(vb.BundleError, msg=bad):
self._install(_zip({**self._good(), bad: b"x"}))
self.assertEqual(list(self.dst.iterdir()), [])
def test_bad_voice_id_in_manifest(self):
for vid in ("../x", "a/b", ".hidden", "", "a" * 200):
with self.assertRaises(vb.BundleError, msg=vid):
self._install(_zip(self._good(), voice_id=vid))
def test_zip_bomb(self):
files = {**self._good(), "reference.txt": b"a" * (vb.MAX_TEXT_BYTES + 1)}
with self.assertRaises(vb.BundleError):
self._install(_zip(files))
files = {"audio.wav": b"\x00" * (20 * 1024 * 1024)} # extreme ratio
with self.assertRaises(vb.BundleError):
self._install(_zip(files))
def test_lying_declared_size_still_capped(self):
orig = vb.MAX_TEXT_BYTES
try:
data = _zip({**self._good(), "reference.txt": b"b" * 5000})
vb.MAX_TEXT_BYTES = 2000
with self.assertRaises(vb.BundleError):
self._install(data)
finally:
vb.MAX_TEXT_BYTES = orig
def test_invalid_meta_and_provenance(self):
for meta in ({"rating": 99}, {"provenance": {"basis": "pirated"}}, {"provenance": {"source": "ftp://u:p@h"}}):
with self.assertRaises(vb.BundleError):
self._install(_zip({"audio.wav": self.wav_bytes, "meta.json": json.dumps(meta).encode()}))
def test_undecodable_audio_and_not_zip(self):
with self.assertRaises(vb.BundleError):
self._install(_zip({"audio.wav": b"not audio at all"}))
with self.assertRaises(vb.BundleError):
self._install(b"garbage")
self.assertEqual(list(self.dst.iterdir()), [])
def test_missing_manifest(self):
with self.assertRaises(vb.BundleError):
self._install(_zip(self._good(), manifest=False))
if __name__ == "__main__":
unittest.main()

View File

@ -0,0 +1,50 @@
// @ts-check
// Voice bundle export/import in the library toolbar (mocked routes only; never touches the live library).
const { test, expect } = require('@playwright/test');
async function setup(page, calls) {
await page.route('**/api/voices/import', async route => {
calls.push('import');
const name = route.request().postData() || '';
if (name.includes('bad.zip')) return route.fulfill({ status: 400, json: { detail: 'Hash mismatch: audio.wav' } });
return route.fulfill({ json: { ok: true, voice_id: 'EN_F_x_2', status: 'renamed' } });
});
await page.route('**/api/voices/*/export', route => {
calls.push('export');
return route.fulfill({ body: 'PK', contentType: 'application/zip' });
});
await page.route('**/api/voices*', route => route.fulfill({ json: [] }));
await page.route('**/api/settings*', route => route.fulfill({ json: {} }));
await page.goto('/', { waitUntil: 'domcontentloaded' });
await page.waitForFunction(() => typeof window.importVoiceBundles === 'function', null, { timeout: 15_000 });
}
test('bundle buttons are labelled and at least 44px', async ({ page }) => {
await setup(page, []);
for (const id of ['export-voices-btn', 'import-voices-btn']) {
const b = page.locator('#' + id);
await expect(b).toHaveText(/Export selected|Import bundle/);
const box = await b.boundingBox().catch(() => null);
const h = await b.evaluate(el => parseFloat(getComputedStyle(el).minHeight));
expect(h).toBeGreaterThanOrEqual(44);
if (box) expect(box.height).toBeGreaterThanOrEqual(44);
}
});
test('export downloads a bundle per voice', async ({ page }) => {
const calls = [];
await setup(page, calls);
const dl = page.waitForEvent('download');
await page.evaluate(() => window.exportVoiceBundles(['EN_F_x']));
expect((await dl).suggestedFilename()).toBe('EN_F_x.voice.zip');
expect(calls).toContain('export');
});
test('import reports renamed voices and shows server errors', async ({ page }) => {
const calls = [];
await setup(page, calls);
await page.evaluate(() => window.importVoiceBundles([new File(['PK'], 'ok.zip')]));
await expect(page.locator('#toast')).toContainText('EN_F_x_2 (name was taken)');
await page.evaluate(() => window.importVoiceBundles([new File(['PK'], 'bad.zip')]));
await expect(page.locator('#toast')).toContainText('Import of bad.zip failed: Hash mismatch');
});